diff --git a/docs/explore-notes/worker-task-pipeline.md b/docs/explore-notes/worker-task-pipeline.md index 743b46de..306ed028 100644 --- a/docs/explore-notes/worker-task-pipeline.md +++ b/docs/explore-notes/worker-task-pipeline.md @@ -124,7 +124,7 @@ read-only "## Reference files" section. - `TaskMergeService` — conflict resolution for worktree merges. **Hub/** -- `HubBroadcaster` — single SignalR broadcast point (TaskStarted/TaskUpdated/TaskMessage/RunCreated…). +- `HubBroadcaster` — single SignalR broadcast point (TaskStarted/TaskUpdated/TaskMessage/WorktreeUpdated…). - `WorkerHub` — SignalR hub + client methods. **Agents/** diff --git a/docs/superpowers/plans/2026-08-07-phase1-reaktivitaets-loecher.md b/docs/superpowers/plans/2026-08-07-phase1-reaktivitaets-loecher.md index 65ec1793..199d3edd 100644 --- a/docs/superpowers/plans/2026-08-07-phase1-reaktivitaets-loecher.md +++ b/docs/superpowers/plans/2026-08-07-phase1-reaktivitaets-loecher.md @@ -869,9 +869,18 @@ git commit -m "fix(worker): broadcast TaskUpdated for tasks imported from the on --- -### Task 7: `TaskUpdated` nach dem Anlegen der List-Handler-Task +### Task 7: ~~`TaskUpdated` nach dem Anlegen der List-Handler-Task~~ — ENTFÄLLT -`InteractiveLaunchSpecService` legt die Handler-Task an (`InteractiveLaunchSpecService.cs:447`) ohne Broadcast. Einzige weitere Aufrufstelle des Konstruktors ist `tests/ClaudeDo.Worker.Tests/Hub/MergeHelperTaskHubTests.cs:64`. +**Bei der Umsetzung am 2026-08-07 verworfen. Prämisse war falsch, kein Code geändert.** + +`InteractiveLaunchSpecService.CreateMergeHelperTaskAsync` (`:447`) broadcastet selbst nichts — das stimmte. Aber ihr **einziger** Produktions-Aufrufer, `WorkerHub.CreateMergeHelperTask` (`src/ClaudeDo.Worker/Hub/WorkerHub.cs:818`), sendet unmittelbar danach `Clients.All.SendAsync("TaskUpdated", taskId)`. Das kam mit Commit `c07c1f7` (2026-08-05) und ist durch `MergeHelperTaskHubTests.CreateMergeHelperTask_CreatesIdleManualTask_StampsBaseCommit_Broadcasts` abgesichert. Der UI-Pfad (`MissionControlViewModel` → `WorkerClient.CreateMergeHelperTaskAsync` → Hub) führt ausschließlich über diesen Aufrufer. + +Den Broadcast zusätzlich in den Service zu legen hätte ihn **verdoppelt**. Ihn dorthin zu *verschieben* wäre ein reiner Konsistenz-Refactor ohne Verhaltensänderung — bewusst nicht gemacht. + +Lehre für den Rest des Plans: Ein DB-Write ohne unmittelbar folgenden Broadcast ist erst dann ein Loch, wenn auch **alle Aufrufer** geprüft sind. Bei `WorktreeManager.cs:103` (Task 5) war das Loch echt, hier nicht. + +
+Ursprünglicher Task-Text (nicht umgesetzt) **Files:** - Modify: `src/ClaudeDo.Worker/Runner/InteractiveLaunchSpecService.cs` @@ -952,6 +961,8 @@ git add src/ClaudeDo.Worker/Runner/InteractiveLaunchSpecService.cs tests/ClaudeD git commit -m "fix(worker): broadcast TaskUpdated when the list-handler task is created" -- src/ClaudeDo.Worker/Runner/InteractiveLaunchSpecService.cs tests/ClaudeDo.Worker.Tests/Hub/MergeHelperTaskHubTests.cs ``` +
+ --- ### Task 8: Totes Event `RunCreated` entfernen diff --git a/docs/superpowers/specs/2026-08-07-ui-reaktivitaet-und-listen-performance-design.md b/docs/superpowers/specs/2026-08-07-ui-reaktivitaet-und-listen-performance-design.md index 01275a0e..0c2fc220 100644 --- a/docs/superpowers/specs/2026-08-07-ui-reaktivitaet-und-listen-performance-design.md +++ b/docs/superpowers/specs/2026-08-07-ui-reaktivitaet-und-listen-performance-design.md @@ -22,12 +22,14 @@ Das eigentliche Problem: **ein einziger verlorener Event ist permanent.** Der ei |---|---|---| | 1 | Blankes `catch { }` um den gesamten Delta-Pfad. Eine einzige transiente Exception (z.B. `SQLITE_BUSY`) lässt die Zeile dauerhaft auf dem alten Stand — ohne Log, ohne Retry. | `src/ClaudeDo.Ui/ViewModels/Islands/TasksIslandViewModel.cs:224` | | 2 | `QueuePicker.ClaimNextAsync` committet `status='running'` sofort. Wirft danach etwas in `RunInSlotAsync` oder im ungeschützten Setup-Block von `TaskRunner.ContinueAsync` (Zeilen 218–238 liegen außerhalb jedes `try`), fängt der Catch das ab und **loggt nur** — kein `FailAsync`, kein Broadcast. DB sagt Running, die UI erfährt es nie. | `src/ClaudeDo.Worker/Queue/QueueService.cs:349-352` | -| 3 | DB-Writes ohne Broadcast. | `src/ClaudeDo.Worker/Runner/WorktreeManager.cs:103`, `src/ClaudeDo.Worker/Online/OnlineSyncService.cs:131`, `src/ClaudeDo.Worker/Runner/InteractiveLaunchSpecService.cs:447` | +| 3 | DB-Writes ohne Broadcast. | `src/ClaudeDo.Worker/Runner/WorktreeManager.cs:103`, `src/ClaudeDo.Worker/Online/OnlineSyncService.cs:131` | + +**Korrektur (2026-08-07, bei der Umsetzung gefunden):** `InteractiveLaunchSpecService.cs:447` stand hier ursprünglich als drittes Loch. Das war falsch. Die Service-Methode broadcastet zwar selbst nicht, aber ihr einziger Produktions-Aufrufer `WorkerHub.CreateMergeHelperTask` (`src/ClaudeDo.Worker/Hub/WorkerHub.cs:818`) sendet direkt danach `TaskUpdated` — seit Commit `c07c1f7` vom 2026-08-05, abgesichert durch `MergeHelperTaskHubTests.CreateMergeHelperTask_CreatesIdleManualTask_StampsBaseCommit_Broadcasts`. Der ursprüngliche Befund hatte den DB-Write gesehen, aber den Aufrufer nicht geprüft. Ein Broadcast im Service wäre ein Duplikat gewesen. Dazu zwei kleinere Befunde: - **Race im Delta-Pfad.** `OnWorkerTaskUpdated` ist `async void` und hängt an *zwei* Events (`TaskUpdatedEvent` und `WorktreeUpdatedEvent`, `TasksIslandViewModel.cs:117-118`). Der Full-Reload-Zweig ist per `_loadCts` gegen Überholen abgesichert, der Delta-Zweig nicht — ein älterer Read kann einen neueren überschreiben. -- **Kein Busy-Timeout konfiguriert.** Die Connection-Strings beider Prozesse sind blanke `Data Source=…` (`src/ClaudeDo.App/Program.cs:100-101`, `src/ClaudeDo.Worker/Program.cs:60-61`). Ohne Timeout schlägt ein seltener `SQLITE_BUSY` sofort als Exception durch, statt kurz zu warten — das erhöht die Wahrscheinlichkeit von Loch 1. Achtung: `PRAGMA busy_timeout` ist **per Connection** und wird — anders als `journal_mode=WAL` — *nicht* in der DB-Datei persistiert. Es in `ClaudeDoDbContext.MigrateAndConfigure` zu setzen würde nur die Startup-Connection betreffen und wäre wirkungslos; es gehört in den Connection-String (`Default Timeout=`), den Microsoft.Data.Sqlite auf den Busy-Handler abbildet. +- ~~**Kein Busy-Timeout konfiguriert.**~~ **Widerlegt (2026-08-07, empirisch geprüft).** Die Vermutung war, die blanken Connection-Strings (`src/ClaudeDo.App/Program.cs:95`, `src/ClaudeDo.Worker/Program.cs:61`) ließen einen `SQLITE_BUSY` sofort durchschlagen. Das stimmt nicht: Microsoft.Data.Sqlite 8.0.11 setzt `DefaultTimeout` **von sich aus auf 30 Sekunden**, mit oder ohne das Keyword — gemessen an `SqliteConnectionStringBuilder("Data Source=x.db").DefaultTimeout` → `30`, ebenso `SqliteConnection.DefaultTimeout` und `SqliteCommand.CommandTimeout`. Ein Contention-Test (Writer hält 2s, zweiter Writer parallel) zeigt, dass der zweite wartet und nach ~2030 ms durchkommt, statt zu werfen. Der ursprünglich dafür gemachte Commit `f62dbb9` war ein No-op mit irreführendem Kommentar und wurde mit `ac58679` zurückgenommen. Die tatsächliche Absicherung gegen transiente Lesefehler leistet der Retry im Delta-Pfad, nicht ein Timeout. - **`RunCreated` ist ein totes Event.** Wird in `TaskRunner.cs:358` gesendet, hat aber keinen einzigen Abonnenten in der UI. ### Performance: der Engpass ist das Rendering, nicht die Datenbank @@ -95,11 +97,9 @@ Unabhängig von Phase 2 und 3, kann sofort starten. | Fix | Ort | |---|---| | `catch { }` ersetzen durch Log + einmaligen Retry. **Kein** Footer-Error — das ist ein Hintergrund-Refresh, keine Nutzeraktion. | `TasksIslandViewModel.cs:224` | -| `Default Timeout=30` in beide Connection-Strings (nicht als PRAGMA — siehe Analyse) | `App/Program.cs:100-101`, `Worker/Program.cs:60-61` | | Catch-Block ruft `_state.FailAsync` (das selbst broadcastet), statt nur zu loggen; `OperationCanceledException` bleibt ausgenommen | `QueueService.cs:349-352` | | `WorktreeUpdated` nach dem Insert broadcasten | `Runner/WorktreeManager.cs:103` | | `TaskUpdated` nach dem Insert broadcasten | `Online/OnlineSyncService.cs:131` | -| `TaskUpdated` nach dem Insert broadcasten | `Runner/InteractiveLaunchSpecService.cs:447` | | Monotone Sequenznummer pro TaskId im Delta-Pfad; Ergebnisse mit veralteter Sequenz verwerfen | `OnWorkerTaskUpdated` | | `RunCreated` ersatzlos entfernen (totes Event ohne Abonnent) | `HubBroadcaster`, `TaskRunner.cs:358` | diff --git a/src/ClaudeDo.Ui/ViewModels/Islands/TasksIslandViewModel.cs b/src/ClaudeDo.Ui/ViewModels/Islands/TasksIslandViewModel.cs index c4bba7fc..6ccd42f9 100644 --- a/src/ClaudeDo.Ui/ViewModels/Islands/TasksIslandViewModel.cs +++ b/src/ClaudeDo.Ui/ViewModels/Islands/TasksIslandViewModel.cs @@ -25,6 +25,10 @@ public sealed partial class TasksIslandViewModel : ViewModelBase, IDisposable // pick the flag up (see SyncInteractiveSessions). private readonly HashSet _interactiveSessionIds = new(); private static readonly TaskListFilterRegistry _filters = new(); + // Two events (TaskUpdated + WorktreeUpdated) drive the same delta refresh, so two reads for + // one task can be in flight at once. Only the newest may write to the row. + private readonly Dictionary _deltaSeq = new(); + private long _deltaCounter; public event EventHandler? SelectionChanged; public event EventHandler? FocusAddTaskRequested; @@ -161,6 +165,11 @@ public sealed partial class TasksIslandViewModel : ViewModelBase, IDisposable } private async void OnWorkerTaskUpdated(string taskId) + => await RefreshTaskFromWorkerAsync(taskId); + + // Awaitable so tests can drive it deterministically. One retry, then a full reload: + // a swallowed exception here used to leave the row on a stale status permanently. + internal async Task RefreshTaskFromWorkerAsync(string taskId) { var list = _currentList; if (list is null) return; @@ -174,54 +183,79 @@ public sealed partial class TasksIslandViewModel : ViewModelBase, IDisposable return; } + var seq = ++_deltaCounter; + _deltaSeq[taskId] = seq; + try { - await using var db = await _dbFactory.CreateDbContextAsync(); - var entity = await db.Tasks - .Include(t => t.List) - .Include(t => t.Worktree) - .FirstOrDefaultAsync(t => t.Id == taskId); - - // A parent transition (finalize/discard) broadcasts only the parent's id, but it - // changes its children's derived state — finalize flips them Draft→Planned, discard - // deletes them. The delta path below only touches the parent row and never recomputes - // the child-derived flags (ParentFinalized, HasPlanningChildren) nor drops deleted - // children, so reconcile the whole list when the updated task is (or owns) a subtree. - if (entity is not null && - (entity.PlanningPhase != PlanningPhase.None || Items.Any(r => r.ParentTaskId == entity.Id))) - { - LoadForList(list); - return; - } - - var existing = Items.FirstOrDefault(r => r.Id == taskId); - - if (entity is null) - { - if (existing is not null) Items.Remove(existing); - } - else - { - var matches = TaskMatchesList(entity, list); - if (existing is not null && matches) existing.UpdateFromEntity(entity); - else if (existing is not null) Items.Remove(existing); - else if (matches) { LoadForList(list); return; } - else return; - } - - // Keep the parent's HasQueuedSubtasks flag in sync when a child's status flips. - if (entity is not null && !string.IsNullOrEmpty(entity.ParentTaskId)) - { - var parent = Items.FirstOrDefault(r => r.Id == entity.ParentTaskId); - if (parent is not null) - parent.HasQueuedSubtasks = Items.Any(r => - r.ParentTaskId == parent.Id && (r.IsQueued || r.IsWaiting)); - } - - Regroup(); - UpdateSubtitle(); + await ApplyDeltaAsync(taskId, list, seq); } - catch { } + catch (Exception first) + { + System.Diagnostics.Debug.WriteLine( + $"TasksIsland: delta refresh for {taskId} failed ({first.Message}); retrying"); + try + { + await ApplyDeltaAsync(taskId, list, seq); + } + catch (Exception second) + { + System.Diagnostics.Debug.WriteLine( + $"TasksIsland: delta retry for {taskId} failed ({second.Message}); full reload"); + LoadForList(list); + } + } + } + + private async Task ApplyDeltaAsync(string taskId, ListNavItemViewModel list, long seq) + { + await using var db = await _dbFactory.CreateDbContextAsync(); + var entity = await db.Tasks + .Include(t => t.List) + .Include(t => t.Worktree) + .FirstOrDefaultAsync(t => t.Id == taskId); + + // A newer refresh for this task started while we were reading — its result is fresher. + if (_deltaSeq.TryGetValue(taskId, out var current) && current != seq) return; + + // A parent transition (finalize/discard) broadcasts only the parent's id, but it + // changes its children's derived state — finalize flips them Draft→Planned, discard + // deletes them. The delta path below only touches the parent row and never recomputes + // the child-derived flags (ParentFinalized, HasPlanningChildren) nor drops deleted + // children, so reconcile the whole list when the updated task is (or owns) a subtree. + if (entity is not null && + (entity.PlanningPhase != PlanningPhase.None || Items.Any(r => r.ParentTaskId == entity.Id))) + { + LoadForList(list); + return; + } + + var existing = Items.FirstOrDefault(r => r.Id == taskId); + + if (entity is null) + { + if (existing is not null) Items.Remove(existing); + } + else + { + var matches = TaskMatchesList(entity, list); + if (existing is not null && matches) existing.UpdateFromEntity(entity); + else if (existing is not null) Items.Remove(existing); + else if (matches) { LoadForList(list); return; } + else return; + } + + // Keep the parent's HasQueuedSubtasks flag in sync when a child's status flips. + if (entity is not null && !string.IsNullOrEmpty(entity.ParentTaskId)) + { + var parent = Items.FirstOrDefault(r => r.Id == entity.ParentTaskId); + if (parent is not null) + parent.HasQueuedSubtasks = Items.Any(r => + r.ParentTaskId == parent.Id && (r.IsQueued || r.IsWaiting)); + } + + Regroup(); + UpdateSubtitle(); } // NOTE: virtual:queued/virtual:running cannot be decided by a single entity — a Planning diff --git a/src/ClaudeDo.Worker/CLAUDE.md b/src/ClaudeDo.Worker/CLAUDE.md index 899d5460..e00d3147 100644 --- a/src/ClaudeDo.Worker/CLAUDE.md +++ b/src/ClaudeDo.Worker/CLAUDE.md @@ -159,7 +159,6 @@ launch specs · worktrees · agents/settings/lists · reports/notes/prep · diag - `TaskMessage` - `WorktreeUpdated` - `TaskUpdated` -- `RunCreated` - `ListUpdated` - `WorkerLog` - `PrimeFired` diff --git a/src/ClaudeDo.Worker/Hub/HubBroadcaster.cs b/src/ClaudeDo.Worker/Hub/HubBroadcaster.cs index f46dabdf..c3828c2d 100644 --- a/src/ClaudeDo.Worker/Hub/HubBroadcaster.cs +++ b/src/ClaudeDo.Worker/Hub/HubBroadcaster.cs @@ -40,9 +40,6 @@ public sealed class HubBroadcaster : IPrimeBroadcaster, IRefineBroadcaster public Task ListUpdated(string listId) => _hub.Clients.All.SendAsync("ListUpdated", listId); - public Task RunCreated(string taskId, int runNumber, bool isRetry) => - _hub.Clients.All.SendAsync("RunCreated", taskId, runNumber, isRetry); - public Task UsageUpdated(UsageSnapshotDto snapshot) => _hub.Clients.All.SendAsync("UsageUpdated", snapshot); diff --git a/src/ClaudeDo.Worker/Online/OnlineSyncService.cs b/src/ClaudeDo.Worker/Online/OnlineSyncService.cs index 3afb1e58..99e5c665 100644 --- a/src/ClaudeDo.Worker/Online/OnlineSyncService.cs +++ b/src/ClaudeDo.Worker/Online/OnlineSyncService.cs @@ -1,6 +1,7 @@ using ClaudeDo.Data; using ClaudeDo.Data.Models; using ClaudeDo.Data.Repositories; +using ClaudeDo.Worker.Hub; using ClaudeDo.Worker.Online.Interfaces; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Hosting; @@ -15,19 +16,22 @@ public sealed class OnlineSyncService : BackgroundService private readonly IOnlineAuthProvider _auth; private readonly OnlineInboxConfig _config; private readonly ILogger _logger; + private readonly HubBroadcaster _broadcaster; public OnlineSyncService( IDbContextFactory dbFactory, IOnlineInboxApi api, IOnlineAuthProvider auth, OnlineInboxConfig config, - ILogger logger) + ILogger logger, + HubBroadcaster broadcaster) { _dbFactory = dbFactory; _api = api; _auth = auth; _config = config; _logger = logger; + _broadcaster = broadcaster; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -129,6 +133,8 @@ public sealed class OnlineSyncService : BackgroundService CommitType = CommitTypeRegistry.DefaultType, }; await tasks.AddAsync(entity, ct); + // Without this the imported task only shows up after a manual reload. + await _broadcaster.TaskUpdated(entity.Id); await _api.MarkImportedAsync(remote.Id, ct); _logger.LogInformation("OnlineSyncService: imported task {Id} ('{Title}')", remote.Id, remote.Title); diff --git a/src/ClaudeDo.Worker/Queue/QueueService.cs b/src/ClaudeDo.Worker/Queue/QueueService.cs index 6167ddba..cfbfec01 100644 --- a/src/ClaudeDo.Worker/Queue/QueueService.cs +++ b/src/ClaudeDo.Worker/Queue/QueueService.cs @@ -346,9 +346,28 @@ public sealed class QueueService : BackgroundService await _runner.RunAsync(task, "queue", ct, alreadyClaimed: true); } + catch (OperationCanceledException) + { + // Cancellation is driven by the cancel path, which already wrote the terminal status. + // Marking the task Failed here would be a regression (it would stomp Cancelled). + _logger.LogInformation("Slot runner cancelled for task {TaskId}", taskId); + } catch (Exception ex) { _logger.LogError(ex, "Slot runner error for task {TaskId}", taskId); + + // The picker already committed status='running' before this ran. Without this the + // task stays Running forever and the UI never hears about it — it keeps showing the + // pre-claim status because the raw-SQL claim itself never broadcasts. + try + { + await _state.FailAsync(taskId, DateTime.UtcNow, + $"Slot runner error: {ex.Message}", CancellationToken.None); + } + catch (Exception failEx) + { + _logger.LogError(failEx, "Could not mark task {TaskId} as failed after a slot error", taskId); + } } } } diff --git a/src/ClaudeDo.Worker/Runner/TaskRunner.cs b/src/ClaudeDo.Worker/Runner/TaskRunner.cs index 423ec493..1f5b1442 100644 --- a/src/ClaudeDo.Worker/Runner/TaskRunner.cs +++ b/src/ClaudeDo.Worker/Runner/TaskRunner.cs @@ -311,6 +311,9 @@ public sealed class TaskRunner { var wtCtx = await _wtManager.CreateAsync(task, list, ct); await _broadcaster.WorkerLog($"Created worktree for \"{task.Title}\"", WorkerLogLevel.Info, DateTime.UtcNow); + // The worktrees row was just inserted; without this the UI keeps showing the task + // as having no worktree until some unrelated event happens to refresh it. + await _broadcaster.WorktreeUpdated(task.Id); return new RunDirResult(wtCtx.WorktreePath, wtCtx, null); } catch (Exception ex) @@ -355,8 +358,6 @@ public sealed class TaskRunner await taskRepo.SetLogPathAsync(taskId, logPath, ct); } - await _broadcaster.RunCreated(taskId, runNumber, isRetry); - var arguments = _argsBuilder.Build(config); await using var logWriter = new LogWriter(logPath); diff --git a/tests/ClaudeDo.Ui.Tests/ViewModels/TasksIslandDeltaResilienceTests.cs b/tests/ClaudeDo.Ui.Tests/ViewModels/TasksIslandDeltaResilienceTests.cs new file mode 100644 index 00000000..99267f15 --- /dev/null +++ b/tests/ClaudeDo.Ui.Tests/ViewModels/TasksIslandDeltaResilienceTests.cs @@ -0,0 +1,153 @@ +using ClaudeDo.Data; +using ClaudeDo.Data.Models; +using ClaudeDo.Ui.ViewModels.Islands; +using Microsoft.EntityFrameworkCore; +using TaskStatus = ClaudeDo.Data.Models.TaskStatus; + +namespace ClaudeDo.Ui.Tests.ViewModels; + +// The delta path in OnWorkerTaskUpdated used to be wrapped in a blank `catch { }`. A single +// transient DB error therefore left the row on its old status forever — the "task stuck on +// Queued although it is running" bug. It must retry, and fall back to a full reload. +public class TasksIslandDeltaResilienceTests : IDisposable +{ + private readonly string _dbPath; + + public TasksIslandDeltaResilienceTests() + { + _dbPath = Path.Combine(Path.GetTempPath(), $"claudedo_ui_delta_{Guid.NewGuid():N}.db"); + using var ctx = NewContext(); + ctx.Database.EnsureCreated(); + } + + public void Dispose() + { + try { File.Delete(_dbPath); } catch { } + try { File.Delete(_dbPath + "-wal"); } catch { } + try { File.Delete(_dbPath + "-shm"); } catch { } + } + + private ClaudeDoDbContext NewContext() + { + var opts = new DbContextOptionsBuilder() + .UseSqlite($"Data Source={_dbPath}") + .Options; + return new ClaudeDoDbContext(opts); + } + + // Throws on the first N CreateDbContext calls, then behaves normally. + private sealed class FlakyDbFactory : IDbContextFactory + { + private readonly Func _create; + private int _failuresLeft; + public int CreateCalls { get; private set; } + + public FlakyDbFactory(Func create, int failuresLeft) + { + _create = create; + _failuresLeft = failuresLeft; + } + + public ClaudeDoDbContext CreateDbContext() + { + CreateCalls++; + if (_failuresLeft > 0) + { + _failuresLeft--; + throw new InvalidOperationException("simulated transient DB failure"); + } + return _create(); + } + + public void FailNext() => _failuresLeft++; + } + + private sealed class FakeWorker : StubWorkerClient + { + } + + // A user list's nav id is prefixed — see TasksIslandRegroupTests.UserList. + private static ListNavItemViewModel UserList(string listEntityId, string name) => + new() { Id = $"user:{listEntityId}", Kind = ListKind.User, Name = name }; + + // LoadForList is void and fires a background task; this is the wait idiom the other + // TasksIsland test files use. + private static async Task LoadAndWaitAsync(TasksIslandViewModel vm, ListNavItemViewModel list) + { + vm.LoadForList(list); + var deadline = DateTime.UtcNow.AddSeconds(5); + while (DateTime.UtcNow < deadline) + { + await Task.Delay(25); + if (vm.Items.Count > 0) break; + } + await Task.Delay(50); + } + + private async Task SeedAsync() + { + await using var db = NewContext(); + db.Lists.Add(new ListEntity { Id = "L1", Name = "Work", CreatedAt = DateTime.UtcNow }); + db.Tasks.Add(new TaskEntity + { + Id = "T1", ListId = "L1", Title = "Task one", + Status = TaskStatus.Queued, CreatedAt = DateTime.UtcNow, SortOrder = 0, + }); + await db.SaveChangesAsync(); + } + + [Fact] + public async Task Delta_refresh_retries_after_a_transient_failure_and_still_applies_the_new_status() + { + await SeedAsync(); + + var flaky = new FlakyDbFactory(NewContext, failuresLeft: 0); + var vm = new TasksIslandViewModel(flaky, new FakeWorker()); + var list = UserList("L1", "Work"); + + await LoadAndWaitAsync(vm, list); + Assert.Equal(TaskStatus.Queued, vm.Items.Single(r => r.Id == "T1").Status); + + // Worker flips the task to Running. + await using (var db = NewContext()) + { + var t = await db.Tasks.FirstAsync(x => x.Id == "T1"); + t.Status = TaskStatus.Running; + await db.SaveChangesAsync(); + } + + // The next delta read fails once; the retry must still land the new status. + flaky.FailNext(); + await vm.RefreshTaskFromWorkerAsync("T1"); + + Assert.Equal(TaskStatus.Running, vm.Items.Single(r => r.Id == "T1").Status); + } + + [Fact] + public async Task A_stale_delta_result_does_not_overwrite_a_newer_one() + { + await SeedAsync(); + + var factory = new FlakyDbFactory(NewContext, failuresLeft: 0); + var vm = new TasksIslandViewModel(factory, new FakeWorker()); + var list = UserList("L1", "Work"); + await LoadAndWaitAsync(vm, list); + + // Start refresh #1 while the DB still says Queued, but do not await it yet. + var first = vm.RefreshTaskFromWorkerAsync("T1"); + + await using (var db = NewContext()) + { + var t = await db.Tasks.FirstAsync(x => x.Id == "T1"); + t.Status = TaskStatus.Running; + await db.SaveChangesAsync(); + } + + // Refresh #2 sees Running and must win, regardless of completion order. + var second = vm.RefreshTaskFromWorkerAsync("T1"); + + await Task.WhenAll(first, second); + + Assert.Equal(TaskStatus.Running, vm.Items.Single(r => r.Id == "T1").Status); + } +} diff --git a/tests/ClaudeDo.Worker.Tests/Online/OnlineSyncServiceTests.cs b/tests/ClaudeDo.Worker.Tests/Online/OnlineSyncServiceTests.cs index 1edbe5c0..ca2bce98 100644 --- a/tests/ClaudeDo.Worker.Tests/Online/OnlineSyncServiceTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Online/OnlineSyncServiceTests.cs @@ -1,6 +1,7 @@ using ClaudeDo.Data; using ClaudeDo.Data.Models; using ClaudeDo.Data.Repositories; +using ClaudeDo.Worker.Hub; using ClaudeDo.Worker.Online; using ClaudeDo.Worker.Tests.Infrastructure; using Microsoft.Extensions.Logging.Abstractions; @@ -55,7 +56,8 @@ public sealed class OnlineSyncServiceTests : IDisposable } } - private OnlineSyncService BuildService(FakeApi api, string? token = "test-token", bool enabled = true) + private OnlineSyncService BuildService( + FakeApi api, string? token = "test-token", bool enabled = true, HubBroadcaster? broadcaster = null) { var config = new OnlineInboxConfig { Enabled = enabled, PollIntervalSeconds = 60 }; var auth = new StaticTokenAuthProvider(token); @@ -64,7 +66,8 @@ public sealed class OnlineSyncServiceTests : IDisposable api, auth, config, - NullLogger.Instance); + NullLogger.Instance, + broadcaster ?? new HubBroadcaster(new CapturingHubContext())); } private async Task<(string ListId, ClaudeDoDbContext Ctx, TaskRepository Tasks, ListRepository Lists)> SeedAsync() @@ -103,6 +106,26 @@ public sealed class OnlineSyncServiceTests : IDisposable Assert.Contains(remoteId, api.MarkedImported); } + [Fact] + public async Task Tick_Imports_RemoteTask_BroadcastsTaskUpdated() + { + var (listId, ctx, _, _) = await SeedAsync(); + using var _ = ctx; + + var remoteId = Guid.NewGuid().ToString(); + var api = new FakeApi + { + UnimportedTasks = [new RemoteTask(remoteId, listId, "From Web", "desc", DateTimeOffset.UtcNow)], + }; + var hubContext = new CapturingHubContext(); + var svc = BuildService(api, broadcaster: new HubBroadcaster(hubContext)); + + await svc.TickAsync(CancellationToken.None); + + Assert.Contains(hubContext.Proxy.Calls, + c => c.Method == "TaskUpdated" && (string)c.Args[0]! == remoteId); + } + [Fact] public async Task Tick_UnknownList_Skips_And_DoesNotMark() { diff --git a/tests/ClaudeDo.Worker.Tests/Runner/QueueClaimTaskUpdatedBroadcastTests.cs b/tests/ClaudeDo.Worker.Tests/Runner/QueueClaimTaskUpdatedBroadcastTests.cs index ecee41dd..83b4e465 100644 --- a/tests/ClaudeDo.Worker.Tests/Runner/QueueClaimTaskUpdatedBroadcastTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Runner/QueueClaimTaskUpdatedBroadcastTests.cs @@ -121,4 +121,52 @@ public sealed class QueueClaimTaskUpdatedBroadcastTests : IDisposable releaseProcess.TrySetResult(); await runTask; } + + [Fact] + public async Task Creating_a_worktree_broadcasts_WorktreeUpdated() + { + string listId = Guid.NewGuid().ToString(), taskId = Guid.NewGuid().ToString(); + var repoDir = Path.Combine(_tempDir, "repo"); + Directory.CreateDirectory(repoDir); + + // A real git repo — Worker.Tests run real git by design. + await RunGitAsync(repoDir, "init"); + await RunGitAsync(repoDir, "config user.email t@t.t"); + await RunGitAsync(repoDir, "config user.name t"); + await File.WriteAllTextAsync(Path.Combine(repoDir, "a.txt"), "hi"); + await RunGitAsync(repoDir, "add a.txt"); + await RunGitAsync(repoDir, "commit -m init"); + + using (var ctx = _db.CreateContext()) + { + ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", WorkingDir = repoDir, CreatedAt = DateTime.UtcNow }); + ctx.Tasks.Add(new TaskEntity + { + Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Running, + StartedAt = DateTime.UtcNow, CreatedAt = DateTime.UtcNow, + }); + await ctx.SaveChangesAsync(); + } + + var fake = new FakeClaudeProcess((_, _, _, _, _) => + Task.FromResult(new RunResult { ExitCode = 0, ResultMarkdown = "ok" })); + var runner = BuildRunner(fake); + + using (var ctx = _db.CreateContext()) + await runner.RunAsync((await new TaskRepository(ctx).GetByIdAsync(taskId))!, "queue", + CancellationToken.None, alreadyClaimed: true); + + Assert.Contains(_hubContext.Proxy.Calls, + c => c.Method == "WorktreeUpdated" && (string)c.Args[0]! == taskId); + } + + private static async Task RunGitAsync(string dir, string args) + { + var psi = new System.Diagnostics.ProcessStartInfo("git", args) + { + WorkingDirectory = dir, RedirectStandardOutput = true, RedirectStandardError = true, + }; + using var p = System.Diagnostics.Process.Start(psi)!; + await p.WaitForExitAsync(); + } } diff --git a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs new file mode 100644 index 00000000..bf604e78 --- /dev/null +++ b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs @@ -0,0 +1,223 @@ +using ClaudeDo.Data; +using ClaudeDo.Data.Git; +using ClaudeDo.Data.Models; +using ClaudeDo.Data.Repositories; +using ClaudeDo.Worker.Config; +using ClaudeDo.Worker.Hub; +using ClaudeDo.Worker.Queue; +using ClaudeDo.Worker.Runner; +using ClaudeDo.Worker.Tests.Infrastructure; +using ClaudeDo.Worker.Usage; +using Microsoft.Data.Sqlite; +using Microsoft.Extensions.Logging.Abstractions; +using TaskStatus = ClaudeDo.Data.Models.TaskStatus; + +namespace ClaudeDo.Worker.Tests.Services; + +// The queue picker's raw-SQL claim commits status='running' before the runner starts. If +// anything then throws before the runner's own terminal-status write, the task used to stay +// Running forever with the UI never notified (RunInSlotAsync's catch only logged the error). +// It must now mark the task Failed for a real exception (which broadcasts TaskUpdated), but +// must NOT do so for a cancellation — the cancel path already wrote the terminal status. +// +// These drive the real QueueService end to end (StartAsync + the waker), not just the +// FailAsync contract, so they actually exercise the fixed catch block. +public sealed class QueueServiceSlotFailureTests : IDisposable +{ + private readonly DbFixture _db = new(); + private readonly string _tempDir; + private readonly WorkerConfig _cfg; + + public QueueServiceSlotFailureTests() + { + _tempDir = Path.Combine(Path.GetTempPath(), $"claudedo_slotfail_{Guid.NewGuid():N}"); + Directory.CreateDirectory(_tempDir); + _cfg = new WorkerConfig + { + SandboxRoot = Path.Combine(_tempDir, "sandbox"), + LogRoot = Path.Combine(_tempDir, "logs"), + QueueBackstopIntervalMs = 50, // fast for tests + }; + } + + public void Dispose() + { + _db.Dispose(); + try { Directory.Delete(_tempDir, true); } catch { } + } + + // Mirrors QueueServiceTests.CreateService but takes the picker as a parameter so each test + // can engineer the exact failure path it needs to exercise. + // Build() wires its own CapturingHubContext internally and hands it back as .Hub — the + // broadcaster inside TaskStateService (and therefore FailAsync's TaskUpdated) uses that + // exact instance, so everything else here must share it too rather than constructing a + // second CapturingHubContext that would silently miss FailAsync's broadcast. + private (QueueService service, CapturingHubContext hub, QueueWaker waker) CreateService(IQueuePicker picker) + { + var dbFactory = _db.CreateFactory(); + var built = TaskStateServiceBuilder.Build(dbFactory); + var broadcaster = new HubBroadcaster(built.Hub); + var wtManager = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger.Instance); + var argsBuilder = new ClaudeArgsBuilder(); + var runner = new TaskRunner(new FakeClaudeProcess(), dbFactory, broadcaster, wtManager, argsBuilder, _cfg, + NullLogger.Instance, built.State, new TaskRunTokenRegistry(), new AttachmentStore(), + new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); + var waker = new QueueWaker(); + var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, built.RunCancels); + var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, waker, picker, + overrideSlot, built.State, built.RunCancels, new FakeUsageGate(), new UsageState(), broadcaster); + return (service, built.Hub, waker); + } + + private async Task SeedListAsync() + { + var listId = Guid.NewGuid().ToString(); + using var ctx = _db.CreateContext(); + ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", CreatedAt = DateTime.UtcNow }); + await ctx.SaveChangesAsync(); + return listId; + } + + // Directly rewrites the task's list_id via a raw connection with FK enforcement off, + // reproducing "the list vanished between the queue claim and the run" without going + // through EF's foreign-key-checked connections (which would reject the write). + private void OrphanTaskListId(string taskId) + { + using var conn = new SqliteConnection($"Data Source={_db.DbPath}"); + conn.Open(); + using (var pragmaCmd = conn.CreateCommand()) + { + pragmaCmd.CommandText = "PRAGMA foreign_keys=OFF;"; + pragmaCmd.ExecuteNonQuery(); + } + using var cmd = conn.CreateCommand(); + cmd.CommandText = "UPDATE tasks SET list_id = 'orphaned-missing-list' WHERE id = $id;"; + cmd.Parameters.AddWithValue("$id", taskId); + cmd.ExecuteNonQuery(); + } + + [Fact] + public async Task A_throwing_slot_run_marks_the_task_Failed_and_broadcasts_TaskUpdated() + { + var listId = await SeedListAsync(); + var taskId = Guid.NewGuid().ToString(); + + using (var ctx = _db.CreateContext()) + { + ctx.Tasks.Add(new TaskEntity + { + Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Queued, + ReviewFeedback = "please fix", CreatedAt = DateTime.UtcNow, + }); + await ctx.SaveChangesAsync(); + + // A prior run with a session id routes RunInSlotAsync into TaskRunner.ContinueAsync + // instead of RunAsync. + await new TaskRunRepository(ctx).AddAsync(new TaskRunEntity + { + Id = Guid.NewGuid().ToString(), TaskId = taskId, RunNumber = 1, IsRetry = false, + Prompt = "original", SessionId = "sess-1", StartedAt = DateTime.UtcNow.AddMinutes(-5), + }); + } + + // ContinueAsync's setup block reads the list *before* its own try/catch starts + // (TaskRunner.cs, ContinueAsync ~line 232-234) and throws InvalidOperationException + // ("List not found.") straight past TaskRunner's own protection. That's the exact gap + // QueueService.RunInSlotAsync's own catch now has to cover. + OrphanTaskListId(taskId); + + var (service, hub, waker) = CreateService(new QueuePicker(_db.CreateFactory())); + + using var cts = new CancellationTokenSource(); + await service.StartAsync(cts.Token); + waker.Wake(); + + // FailAsync (TaskStateService.cs:236-249) commits the DB status flip via + // ExecuteUpdateAsync *before* it calls the broadcaster's TaskUpdated — so a poll that + // breaks the instant it observes Status==Failed can race ahead of the broadcast still + // landing in hub.Proxy.Calls. Wait for both signals together so the assertions below + // never sample a genuinely-not-yet-broadcast window as a failure. + TaskEntity? reloaded = null; + var deadline = DateTime.UtcNow.AddSeconds(10); + while (DateTime.UtcNow < deadline) + { + using var verify = _db.CreateContext(); + reloaded = await new TaskRepository(verify).GetByIdAsync(taskId); + var broadcastSeen = hub.Proxy.Calls.Any( + c => c.Method == "TaskUpdated" && (string)c.Args[0]! == taskId); + if (reloaded!.Status == TaskStatus.Failed && broadcastSeen) break; + await Task.Delay(25); + } + cts.Cancel(); + + Assert.Equal(TaskStatus.Failed, reloaded!.Status); + Assert.Contains(hub.Proxy.Calls, + c => c.Method == "TaskUpdated" && (string)c.Args[0]! == taskId); + } + + // A fake IQueuePicker that performs the real atomic claim (so the DB row transitions + // Queued->Running exactly like production) and then, synchronously before returning, + // cancels the token QueueService's per-slot CTS is linked from. By the time + // QueueService.ExecuteAsync creates that linked CTS and dispatches RunInSlotAsync, the + // token is already cancelled — deterministic, no timing race required. + private sealed class ClaimThenCancelPicker : IQueuePicker + { + private readonly IQueuePicker _inner; + private readonly CancellationTokenSource _cancelAfterClaim; + + public ClaimThenCancelPicker(IQueuePicker inner, CancellationTokenSource cancelAfterClaim) + { + _inner = inner; + _cancelAfterClaim = cancelAfterClaim; + } + + public async Task ClaimNextAsync(DateTime now, CancellationToken ct) + { + var claimed = await _inner.ClaimNextAsync(now, ct); + if (claimed is not null) _cancelAfterClaim.Cancel(); + return claimed; + } + } + + [Fact] + public async Task A_cancelled_slot_run_does_not_mark_the_task_Failed() + { + var listId = await SeedListAsync(); + var taskId = Guid.NewGuid().ToString(); + + using (var ctx = _db.CreateContext()) + { + ctx.Tasks.Add(new TaskEntity + { + Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Queued, + CreatedAt = DateTime.UtcNow, + }); + await ctx.SaveChangesAsync(); + } + + var outerCts = new CancellationTokenSource(); + var realPicker = new QueuePicker(_db.CreateFactory()); + var picker = new ClaimThenCancelPicker(realPicker, outerCts); + var (service, hub, waker) = CreateService(picker); + + await service.StartAsync(outerCts.Token); + waker.Wake(); + + // Wait for the slot to be claimed and then released again (RunInSlotAsync's + // ContinueWith removes it once the catch block — ours or a stray one — finishes). + var deadline = DateTime.UtcNow.AddSeconds(10); + while (service.GetActive().Any(a => a.taskId == taskId) && DateTime.UtcNow < deadline) + await Task.Delay(25); + await Task.Delay(100); // let the fire-and-forget continuation fully settle + + TaskEntity? reloaded; + using (var verify = _db.CreateContext()) + reloaded = await new TaskRepository(verify).GetByIdAsync(taskId); + + // The picker's atomic claim already flipped it to Running; the cancelled slot run must + // leave it there rather than flipping it to Failed. + Assert.Equal(TaskStatus.Running, reloaded!.Status); + Assert.DoesNotContain(hub.Proxy.Calls, + c => c.Method == "TaskUpdated" && (string)c.Args[0]! == taskId); + } +}