using Nexus.Api.Data; using Nexus.Api.Models; using Nexus.Api.Repositories; namespace Nexus.Api.Services; public sealed class StaleTaskRecoveryService( ITaskRepository taskRepository, IActivityRepository activityRepository, ILiveUpdateService liveUpdateService, INotificationService notificationService) : IStaleTaskRecoveryService { private const string StalledActivityType = "stalled"; /// /// NICHT-destruktiver Watchdog: markiert „In progress"-Tasks ohne Aktivität seit /// als hängend (Activity-Event + Notification an Iris), /// OHNE die Spalte zu ändern oder Arbeit zu verwerfen. Iris eskaliert dann (nachfragen, /// neu delegieren, ggf. auf Blocked setzen). Dedup: bereits gemeldete Hänger werden nicht /// erneut gemeldet, solange kein neuer Fortschritt (andere Activity) dazwischen liegt. /// public async Task FlagStalledInProgressTasksAsync(TimeSpan stalledThreshold, CancellationToken ct = default) { var now = DateTimeOffset.UtcNow; var threshold = now - stalledThreshold; var allTasks = await taskRepository.GetAllAsync(ct); var inProgress = allTasks .Where(t => string.Equals(t.State, TaskStateHelper.ToStateString(TaskState.InProgress), StringComparison.OrdinalIgnoreCase)) .ToList(); if (inProgress.Count == 0) return 0; var activities = await activityRepository.GetRecentForTasksAsync(inProgress.Select(t => t.Id), ct); var activityByTask = activities .Where(a => a.TaskId.HasValue) .GroupBy(a => a.TaskId!.Value) .ToDictionary(g => g.Key, g => g.OrderByDescending(a => a.CreatedAt).ToList()); var flaggedCount = 0; foreach (var task in inProgress) { activityByTask.TryGetValue(task.Id, out var taskActivity); var latest = taskActivity?.FirstOrDefault(); var lastProgressAt = latest?.CreatedAt ?? task.UpdatedAt; if (lastProgressAt >= threshold) continue; // Dedup: schon als hängend gemeldet und seither kein neuer Fortschritt. if (latest is not null && string.Equals(latest.Type, StalledActivityType, StringComparison.OrdinalIgnoreCase)) continue; var silentFor = now - lastProgressAt; await activityRepository.AddAsync(new ActivityEvent { Type = StalledActivityType, Message = $"Watchdog: keine Aktivität seit {FormatDuration(silentFor)} (Schwelle {FormatDuration(stalledThreshold)}). Task bleibt In progress, Iris zur Eskalation benachrichtigt.", TaskId = task.Id }, ct); await notificationService.CreateAsync( "task_stalled", $"Task hängt: {task.Title}", $"Seit {FormatDuration(silentFor)} keine Aktivität. Bitte nachfassen, neu delegieren oder blockieren.", "iris", task.Id, ct); flaggedCount++; } if (flaggedCount > 0) liveUpdateService.Publish("tasks.board.snapshot", await BuildBoardSnapshotAsync(ct), "board"); return flaggedCount; } /// /// Destruktiver Fallback (nur manuell via Endpoint / expliziter Cron): setzt hängende /// „In progress"-Tasks hart auf Backlog zurück. Verwirft laufenden Kontext — daher NICHT /// mehr der Standard-Watchdog, sondern nur noch auf Anforderung. /// public async Task ResetStaleInProgressTasksAsync(TimeSpan staleThreshold, CancellationToken ct = default) { var threshold = DateTimeOffset.UtcNow - staleThreshold; var staleTasks = await GetStaleTasksAsync(threshold, ct); if (staleTasks.Count == 0) return 0; var latestActivityByTaskId = await GetLatestActivityByTaskIdAsync(staleTasks.Select(task => task.Id), ct); var now = DateTimeOffset.UtcNow; var resetCount = 0; foreach (var task in staleTasks) { var currentTask = await taskRepository.GetByIdAsync(task.Id, ct); if (currentTask is null || !IsStaleInProgress(currentTask, threshold)) continue; latestActivityByTaskId.TryGetValue(currentTask.Id, out var lastActivityAt); var message = BuildActivityMessage(currentTask, staleThreshold, now, lastActivityAt); var updated = await taskRepository.TryResetStaleInProgressToBacklogAsync( currentTask.Id, threshold, now, ct); if (!updated) continue; await activityRepository.AddAsync(new ActivityEvent { Type = "task", Message = message, TaskId = task.Id }, ct); resetCount++; } if (resetCount > 0) liveUpdateService.Publish("tasks.board.snapshot", await BuildBoardSnapshotAsync(ct), "board"); return resetCount; } private async Task> GetStaleTasksAsync(DateTimeOffset threshold, CancellationToken ct) { var allTasks = await taskRepository.GetAllAsync(ct); return allTasks .Where(task => IsStaleInProgress(task, threshold)) .ToList(); } private static bool IsStaleInProgress(WorkTask task, DateTimeOffset threshold) => string.Equals(task.State, TaskStateHelper.ToStateString(TaskState.InProgress), StringComparison.OrdinalIgnoreCase) && task.UpdatedAt < threshold; private async Task> GetLatestActivityByTaskIdAsync( IEnumerable taskIds, CancellationToken ct) { var activities = await activityRepository.GetRecentForTasksAsync(taskIds, ct); return activities .Where(activity => activity.TaskId.HasValue) .GroupBy(activity => activity.TaskId!.Value) .ToDictionary(group => group.Key, group => group.Max(activity => activity.CreatedAt)); } private async Task BuildBoardSnapshotAsync(CancellationToken ct) { var allTasks = await taskRepository.GetAllAsync(ct); var activity = await activityRepository.GetRecentForTasksAsync(allTasks.Select(task => task.Id), ct); return TaskService.BuildMasterBoard(allTasks, activity); } private static string BuildActivityMessage( WorkTask task, TimeSpan staleThreshold, DateTimeOffset now, DateTimeOffset? lastActivityAt) { var staleAge = now - task.UpdatedAt; var details = new List { "reason=stale-recovery", "previous status In progress", $"stale reference {now:O}", $"stale age {FormatDuration(staleAge)}", $"threshold {FormatDuration(staleThreshold)}" }; if (lastActivityAt.HasValue) details.Add($"last activity {lastActivityAt.Value:O}"); details.Add($"last update {task.UpdatedAt:O}"); details.Add("new status Backlog"); return $"Task \"{task.Title}\" reset from In progress to Backlog by stale recovery ({string.Join("; ", details)})"; } private static string FormatDuration(TimeSpan duration) { if (duration.TotalHours >= 1) return $"{(int)duration.TotalHours}h {duration.Minutes}min"; return $"{Math.Max(0, (int)duration.TotalMinutes)}min"; } }