From 389c9126c8537739f3aae79ee10f87437ea96a8e Mon Sep 17 00:00:00 2001 From: mika kuns Date: Thu, 20 Aug 2026 15:00:14 +0200 Subject: [PATCH] =?UTF-8?q?fix(worker):=20Stuck-Running-Fenster=20in=20Con?= =?UTF-8?q?tinue-=20und=20Stop-Pfad=20schlie=C3=9Fen?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TaskRunner.ContinueAsync: Claim, SeedAsync und SetupMcpConfigAsync liefen vor dem try/catch (anders als RunAsync) - warf einer davon nach dem Running-Claim, propagierte die Exception ungefangen bis zu OverrideSlotService.RunContinueInSlotAsync, das nur loggt. Task blieb Running. Fix: derselbe Aufbau wie RunAsync (Claim+Seed+MCP-Setup im try, MarkFailed im catch, mcpToken/mcpConfigPath vor dem try auf null). SetupMcpConfigAsync bekommt zusätzlich einen onTokenRegistered-Callback, damit die äußere mcpToken-Variable den Token sofort nach dem Register sieht - sonst hätte ein Fehler zwischen Register und Rückgabe (z.B. File.WriteAllTextAsync) den Token in der TaskRunTokenRegistry geleakt (betrifft denselben Aufruf in RunAsync mit, daher dort ebenfalls verdrahtet - RunAsync-Struktur selbst unverändert). QueueService.RunInSlotAsync: der Stop-Button (CancelTask) cancelt die Slot-CTS direkt ohne TaskStateService-Schreibzugriff. Traf das die Pre-Dispatch-DB-Reads, loggte der OCE-Catch nur und die vom Picker bereits auf Running geclaimte Task blieb dort für immer hängen. Fix: Status im Catch neu lesen und nur wenn er noch Running ist über TaskStateService.CancelAsync auf Cancelled setzen - ein Ursprung, der bereits selbst einen Terminalstatus geschrieben hat (z.B. CancelReview), wird nicht überschrieben. Kommentar korrigiert. TDD: neue Tests in ContinueAsyncExceptionTests (Seed-/MCP-Setup-Fehler nach Claim -> Failed, kein Token-Leak) und QueueServiceSlotFailureTests (Stop während Pre-Dispatch -> Cancelled statt Running; ein bereits terminal geschriebener Status wird nicht gestompt) vorher rot, jetzt grün. Worker.Tests: 1213/1213 grün, Worker baut in Release. --- src/ClaudeDo.Worker/Queue/QueueService.cs | 27 +++- .../Runner/TaskRunTokenRegistry.cs | 5 + src/ClaudeDo.Worker/Runner/TaskRunner.cs | 67 +++++----- .../Infrastructure/FakeSessionSkillSeeder.cs | 5 + .../Runner/ContinueAsyncExceptionTests.cs | 96 ++++++++++++++- .../Services/QueueServiceSlotFailureTests.cs | 115 +++++++++++++++--- 6 files changed, 266 insertions(+), 49 deletions(-) diff --git a/src/ClaudeDo.Worker/Queue/QueueService.cs b/src/ClaudeDo.Worker/Queue/QueueService.cs index cfbfec01..8d993b6e 100644 --- a/src/ClaudeDo.Worker/Queue/QueueService.cs +++ b/src/ClaudeDo.Worker/Queue/QueueService.cs @@ -348,9 +348,32 @@ public sealed class QueueService : BackgroundService } 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). + // Most cancellation origins (TaskStateService.CancelAsync, the runner's own + // OCE handling) already write a terminal status before the CTS is actually + // cancelled, so by the time we get here the task has already left Running. + // A plain Stop (QueueService.CancelTask cancels the slot's CTS directly, with no + // TaskStateService write) landing during the pre-dispatch reads above is the one + // gap nothing else covers — the picker's claim already committed status='running' + // and nothing since has moved it off that. Only close that specific gap: write + // Cancelled when (and only when) the task is still Running, so an origin that + // already wrote its own terminal status is never stomped. _logger.LogInformation("Slot runner cancelled for task {TaskId}", taskId); + try + { + TaskStatus current; + using (var context = _dbFactory.CreateDbContext()) + current = await context.Tasks.AsNoTracking() + .Where(t => t.Id == taskId) + .Select(t => t.Status) + .FirstAsync(CancellationToken.None); + + if (current == TaskStatus.Running) + await _state.CancelAsync(taskId, DateTime.UtcNow, CancellationToken.None); + } + catch (Exception cancelEx) + { + _logger.LogError(cancelEx, "Could not finalize cancelled task {TaskId} after slot cancellation", taskId); + } } catch (Exception ex) { diff --git a/src/ClaudeDo.Worker/Runner/TaskRunTokenRegistry.cs b/src/ClaudeDo.Worker/Runner/TaskRunTokenRegistry.cs index a2228164..84d9c6d9 100644 --- a/src/ClaudeDo.Worker/Runner/TaskRunTokenRegistry.cs +++ b/src/ClaudeDo.Worker/Runner/TaskRunTokenRegistry.cs @@ -17,6 +17,11 @@ public sealed class TaskRunTokenRegistry return false; } public void Unregister(string token) => _tokenToTaskId.TryRemove(token, out _); + + // Test-only leak check: a run must never leave its token registered after it ends, + // no matter which step of setup failed. + public bool HasAnyTokenFor(string taskId) => _tokenToTaskId.Values.Contains(taskId); + public static string GenerateToken() { var bytes = RandomNumberGenerator.GetBytes(32); diff --git a/src/ClaudeDo.Worker/Runner/TaskRunner.cs b/src/ClaudeDo.Worker/Runner/TaskRunner.cs index 4c38d745..9f602601 100644 --- a/src/ClaudeDo.Worker/Runner/TaskRunner.cs +++ b/src/ClaudeDo.Worker/Runner/TaskRunner.cs @@ -124,7 +124,7 @@ public sealed class TaskRunner var runDir = prep.RunDir!; var resolvedConfig = await ResolveConfigAsync(task, list, listConfig, null, ct); - (mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct); + (mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct, token => mcpToken = token); await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct); @@ -237,32 +237,35 @@ public sealed class TaskRunner runDir = Path.Combine(_cfg.SandboxRoot, taskId); } - var now = DateTime.UtcNow; - // See RunAsync: queue dispatches arrive pre-claimed by the picker. - if (!alreadyClaimed) - { - var startResult = await _state.StartRunningAsync(taskId, now, ct); - if (!startResult.Ok) - { - _logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", taskId, startResult.Reason); - return; - } - } - else - { - // Queue-claimed dispatches skip StartRunningAsync, so broadcast TaskUpdated here - // (see RunAsync for the full rationale). - await _broadcaster.TaskUpdated(taskId); - } - await _broadcaster.TaskStarted(slot, taskId, now); - - await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct); - - var (mcpToken, mcpConfigPath, mcpConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct); - resolvedConfig = mcpConfig; - + string? mcpToken = null; + string? mcpConfigPath = null; try { + var now = DateTime.UtcNow; + // See RunAsync: queue dispatches arrive pre-claimed by the picker. Claim, skill + // seeding, and MCP setup all live inside this try (mirroring RunAsync) so a failure + // in any of them still lands the task in Failed instead of stuck Running. + if (!alreadyClaimed) + { + var startResult = await _state.StartRunningAsync(taskId, now, ct); + if (!startResult.Ok) + { + _logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", taskId, startResult.Reason); + return; + } + } + else + { + // Queue-claimed dispatches skip StartRunningAsync, so broadcast TaskUpdated here + // (see RunAsync for the full rationale). + await _broadcaster.TaskUpdated(taskId); + } + await _broadcaster.TaskStarted(slot, taskId, now); + + await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct); + + (mcpToken, mcpConfigPath, resolvedConfig) = await SetupMcpConfigAsync(task, resolvedConfig, ct, token => mcpToken = token); + var nextRunNumber = lastRun.RunNumber + 1; var result = await RunOnceAsync(taskId, task.Number, task.Title, slot, runDir, resolvedConfig, nextRunNumber, false, followUpPrompt, ct); @@ -290,8 +293,12 @@ public sealed class TaskRunner } finally { - _tokens.Unregister(mcpToken); - try { File.Delete(mcpConfigPath); } catch { /* best effort */ } + if (mcpToken is not null) + { + _tokens.Unregister(mcpToken); + if (mcpConfigPath is not null) + try { File.Delete(mcpConfigPath); } catch { /* best effort */ } + } } } @@ -300,11 +307,15 @@ public sealed class TaskRunner // SuggestImprovement for filing out-of-scope follow-ups. Used by both a fresh run and a // --resume continuation, which needs the same wiring or the resumed session loses every // mcp__claudedo_run__* tool. + // onTokenRegistered fires the instant the token is registered, before any I/O that could + // still fail — callers assign it straight into their outer (pre-try) mcpToken variable so a + // later throw in this method still leaves the finally block able to unregister it. private async Task<(string McpToken, string McpConfigPath, ClaudeRunConfig Config)> SetupMcpConfigAsync( - TaskEntity task, ClaudeRunConfig resolvedConfig, CancellationToken ct) + TaskEntity task, ClaudeRunConfig resolvedConfig, CancellationToken ct, Action? onTokenRegistered = null) { var mcpToken = TaskRunTokenRegistry.GenerateToken(); _tokens.Register(mcpToken, task.Id); + onTokenRegistered?.Invoke(mcpToken); Directory.CreateDirectory(_cfg.LogRoot); var mcpConfigPath = Path.Combine(_cfg.LogRoot, $"{task.Id}_mcp.json"); await File.WriteAllTextAsync(mcpConfigPath, BuildRunMcpConfigJson(mcpToken), ct); diff --git a/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeSessionSkillSeeder.cs b/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeSessionSkillSeeder.cs index c4a2d352..af42eaed 100644 --- a/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeSessionSkillSeeder.cs +++ b/tests/ClaudeDo.Worker.Tests/Infrastructure/FakeSessionSkillSeeder.cs @@ -9,10 +9,15 @@ public sealed class FakeSessionSkillSeeder : ISessionSkillSeeder public readonly record struct Call(string WorkingDir, IReadOnlyList SkillNames, bool IsWorktree); public readonly ConcurrentQueue Calls = new(); + // Lets a test simulate a seeding failure (e.g. a broken skill install) to verify callers + // don't leave a task stuck Running when SeedAsync throws. + public Exception? ThrowOnSeed; + public Task SeedAsync(string workingDir, IReadOnlyList skillNames, bool isWorktree, CancellationToken ct) { Interlocked.Increment(ref CallCount); Calls.Enqueue(new Call(workingDir, skillNames, isWorktree)); + if (ThrowOnSeed is not null) throw ThrowOnSeed; return Task.CompletedTask; } } diff --git a/tests/ClaudeDo.Worker.Tests/Runner/ContinueAsyncExceptionTests.cs b/tests/ClaudeDo.Worker.Tests/Runner/ContinueAsyncExceptionTests.cs index 70afcd75..f3083744 100644 --- a/tests/ClaudeDo.Worker.Tests/Runner/ContinueAsyncExceptionTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Runner/ContinueAsyncExceptionTests.cs @@ -4,6 +4,7 @@ using ClaudeDo.Data.Repositories; using ClaudeDo.Worker.Config; using ClaudeDo.Worker.Hub; using ClaudeDo.Worker.Runner; +using ClaudeDo.Worker.Skills; using ClaudeDo.Worker.Tests.Infrastructure; using Microsoft.Extensions.Logging.Abstractions; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -29,14 +30,101 @@ public sealed class ContinueAsyncExceptionTests : IDisposable public void Dispose() { _db.Dispose(); try { Directory.Delete(_tempDir, true); } catch { } } - private TaskRunner BuildRunner(IClaudeProcess claude, ClaudeDoDbContext ctx) + private TaskRunner BuildRunner( + IClaudeProcess claude, ClaudeDoDbContext ctx, + ISessionSkillSeeder? skillSeeder = null, TaskRunTokenRegistry? tokens = null, WorkerConfig? cfg = null) { var dbFactory = _db.CreateFactory(); var broadcaster = new HubBroadcaster(new CapturingHubContext()); var state = TaskStateServiceBuilder.Build(dbFactory).State; - var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger.Instance); - return new TaskRunner(claude, dbFactory, broadcaster, wt, new ClaudeArgsBuilder(), _cfg, - NullLogger.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); + var effectiveCfg = cfg ?? _cfg; + var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, effectiveCfg, NullLogger.Instance); + return new TaskRunner(claude, dbFactory, broadcaster, wt, new ClaudeArgsBuilder(), effectiveCfg, + NullLogger.Instance, state, tokens ?? new TaskRunTokenRegistry(), new AttachmentStore(), + skillSeeder ?? new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); + } + + private async Task SeedContinuableTaskAsync(string sessionId) + { + string listId, taskId; + using var ctx = _db.CreateContext(); + listId = Guid.NewGuid().ToString(); + ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", WorkingDir = null, CreatedAt = DateTime.UtcNow }); + + taskId = Guid.NewGuid().ToString(); + ctx.Tasks.Add(new TaskEntity + { + Id = taskId, + ListId = listId, + Title = "Continue me", + Status = TaskStatus.WaitingForReview, + CreatedAt = DateTime.UtcNow, + }); + await ctx.SaveChangesAsync(); + + await new TaskRunRepository(ctx).AddAsync(new TaskRunEntity + { + Id = Guid.NewGuid().ToString(), + TaskId = taskId, + RunNumber = 1, + IsRetry = false, + Prompt = "original prompt", + SessionId = sessionId, + StartedAt = DateTime.UtcNow.AddMinutes(-5), + FinishedAt = DateTime.UtcNow.AddMinutes(-1), + ExitCode = 0, + ResultMarkdown = "first result", + }); + return taskId; + } + + // Bug: SeedAsync used to run after the Running claim but before ContinueAsync's own + // try/catch started, so a throw here propagated straight out of ContinueAsync to the + // caller (OverrideSlotService.RunContinueInSlotAsync), which only logs — leaving the task + // stuck Running forever. + [Fact] + public async Task ContinueAsync_SkillSeederThrowsAfterClaim_MarksTaskFailed_NotStuckRunning() + { + var taskId = await SeedContinuableTaskAsync("sess-seed-throws"); + var throwingSeeder = new FakeSessionSkillSeeder { ThrowOnSeed = new InvalidOperationException("skill seed exploded") }; + + using var ctx2 = _db.CreateContext(); + var runner = BuildRunner(new FakeClaudeProcess(), ctx2, skillSeeder: throwingSeeder); + + await runner.ContinueAsync(taskId, "please continue", "slot-1", CancellationToken.None); + + using var verify = _db.CreateContext(); + var task = await new TaskRepository(verify).GetByIdAsync(taskId); + Assert.NotNull(task); + Assert.Equal(TaskStatus.Failed, task.Status); + } + + // Bug: SetupMcpConfigAsync registers the per-run MCP token *before* it can fail (writing + // the config file to disk). When it threw after Register but before the try/catch existed, + // the token was never unregistered — a leak in TaskRunTokenRegistry on top of the stuck + // Running task. + [Fact] + public async Task ContinueAsync_McpSetupThrowsAfterTokenRegistered_MarksTaskFailed_AndUnregistersToken() + { + var taskId = await SeedContinuableTaskAsync("sess-mcp-throws"); + + // SetupMcpConfigAsync calls Directory.CreateDirectory(cfg.LogRoot) right after + // registering the token; pointing LogRoot at an existing file makes that call throw. + var badLogRoot = Path.Combine(_tempDir, "logroot_is_a_file"); + await File.WriteAllTextAsync(badLogRoot, "not a directory"); + var cfg = new WorkerConfig { SandboxRoot = _tempDir, LogRoot = badLogRoot }; + var tokens = new TaskRunTokenRegistry(); + + using var ctx2 = _db.CreateContext(); + var runner = BuildRunner(new FakeClaudeProcess(), ctx2, tokens: tokens, cfg: cfg); + + await runner.ContinueAsync(taskId, "please continue", "slot-1", CancellationToken.None); + + using var verify = _db.CreateContext(); + var task = await new TaskRepository(verify).GetByIdAsync(taskId); + Assert.NotNull(task); + Assert.Equal(TaskStatus.Failed, task.Status); + Assert.False(tokens.HasAnyTokenFor(taskId)); } [Fact] diff --git a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs index bf604e78..3a913bf4 100644 --- a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotFailureTests.cs @@ -9,6 +9,7 @@ using ClaudeDo.Worker.Runner; using ClaudeDo.Worker.Tests.Infrastructure; using ClaudeDo.Worker.Usage; using Microsoft.Data.Sqlite; +using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging.Abstractions; using TaskStatus = ClaudeDo.Data.Models.TaskStatus; @@ -180,7 +181,7 @@ public sealed class QueueServiceSlotFailureTests : IDisposable } [Fact] - public async Task A_cancelled_slot_run_does_not_mark_the_task_Failed() + public async Task A_cancelled_slot_run_is_marked_Cancelled_not_left_Running() { var listId = await SeedListAsync(); var taskId = Guid.NewGuid().ToString(); @@ -203,21 +204,105 @@ public sealed class QueueServiceSlotFailureTests : IDisposable 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 + var reloaded = await PollUntilLeftQueuedAsync(taskId); - 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, + // This reproduces a plain Stop (QueueService.CancelTask cancels the slot's CTS + // directly, without going through TaskStateService) landing during the pre-dispatch DB + // reads — before TaskRunner's own claim/try-catch ever starts. Nothing else writes a + // terminal status for that window, so the OCE catch itself must close it. + Assert.Equal(TaskStatus.Cancelled, reloaded!.Status); + Assert.Contains(hub.Proxy.Calls, c => c.Method == "TaskUpdated" && (string)c.Args[0]! == taskId); } + + // A_cancelled_slot_run_is_marked_Cancelled_not_left_Running above reproduces a plain Stop. + // TaskStateService.CancelAsync (the hub's CancelReview action) is a different origin: it + // writes Cancelled *before* it cancels the run's CTS via RunCancellationRegistry, so by the + // time RunInSlotAsync's OCE catch runs, the task has already left Running. The catch must + // recognize that and not blindly stomp whatever terminal status is already there. + [Fact] + public async Task A_cancelled_slot_run_does_not_stomp_a_status_already_written_terminal() + { + 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 ClaimMarkFailedThenCancelPicker(realPicker, outerCts, _db); + var (service, hub, waker) = CreateService(picker); + + await service.StartAsync(outerCts.Token); + waker.Wake(); + + var reloaded = await PollUntilLeftQueuedAsync(taskId); + + // Some other terminal write (simulating CancelAsync/FailAsync having already run) must + // survive — the OCE catch must not overwrite it with Cancelled just because it observed + // a cancellation. + Assert.Equal(TaskStatus.Failed, reloaded!.Status); + } + + // Polls until the task leaves its initial Queued status (the claim + dispatch + OCE-catch + // chain settling), or gives up at the deadline; then waits a further grace period so the + // fire-and-forget continuation (queue-slot cleanup, any terminal-status write) fully lands + // before the caller reads the final state. + private async Task PollUntilLeftQueuedAsync(string taskId) + { + 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); + if (reloaded!.Status != TaskStatus.Queued) break; + await Task.Delay(25); + } + await Task.Delay(150); // let the fire-and-forget continuation fully settle + + using var final = _db.CreateContext(); + return await new TaskRepository(final).GetByIdAsync(taskId); + } + + // Simulates a terminal status already having been written (by TaskStateService, from some + // other origin) between the picker's claim and RunInSlotAsync's OCE catch running. + private sealed class ClaimMarkFailedThenCancelPicker : IQueuePicker + { + private readonly IQueuePicker _inner; + private readonly CancellationTokenSource _cancelAfterClaim; + private readonly DbFixture _db; + + public ClaimMarkFailedThenCancelPicker(IQueuePicker inner, CancellationTokenSource cancelAfterClaim, DbFixture db) + { + _inner = inner; + _cancelAfterClaim = cancelAfterClaim; + _db = db; + } + + public async Task ClaimNextAsync(DateTime now, CancellationToken ct) + { + var claimed = await _inner.ClaimNextAsync(now, ct); + if (claimed is not null) + { + using (var ctx = _db.CreateContext()) + { + await ctx.Tasks.Where(t => t.Id == claimed.Id) + .ExecuteUpdateAsync(s => s + .SetProperty(t => t.Status, TaskStatus.Failed) + .SetProperty(t => t.FinishedAt, DateTime.UtcNow)); + } + _cancelAfterClaim.Cancel(); + } + return claimed; + } + } }