diff --git a/docs/explore-notes/worker-task-pipeline.md b/docs/explore-notes/worker-task-pipeline.md index 3e49c709..ffde4eb2 100644 --- a/docs/explore-notes/worker-task-pipeline.md +++ b/docs/explore-notes/worker-task-pipeline.md @@ -1,6 +1,6 @@ > **Explore-note — verify before trusting.** Distilled map of a subsystem, not authoritative. -> Last verified against commit `58741c2` (2026-08-06). -> Drift check: `git log --oneline 58741c2..HEAD -- src/ClaudeDo.Worker` +> Last verified against commit `cc90600` (2026-08-06). +> Drift check: `git log --oneline cc90600..HEAD -- src/ClaudeDo.Worker` > Stable structure only (no line numbers). See docs/explore-notes/README.md. # Worker: Task Execution Pipeline @@ -28,9 +28,11 @@ How a task moves Queued → Running → terminal, across `src/ClaudeDo.Worker` 5. **Run Preparation** — `TaskRunner.RunAsync()` (Runner/TaskRunner.cs) - Loads task, list config, subtasks, attachments from the DB. + - `StartRunningAsync()` (only if not pre-claimed): atomic claim to Running, **before any + resource is created**. A rejected claim (task already Running) bails out immediately — + no worktree, no MCP token file. Broadcasts TaskStarted. - `PrepareRunDirectoryAsync()`: worktree (via WorktreeManager) if the list has a WorkingDir, else sandbox. Generates a per-run MCP token, writes MCP config to disk. - - `StartRunningAsync()` (only if not pre-claimed): atomic Queued → Running. Broadcasts TaskStarted. 6. **Claude Execution** — `TaskRunner.RunOnceAsync()` (Runner/TaskRunner.cs) - Creates a TaskRunEntity, points the task at the run's log path. @@ -152,6 +154,15 @@ Program.cs (DI setup) - **Slot limit** — respects MaxParallelExecutions; a backstop timer wakes even if a Wake() is missed. - **Pre-claimed tasks** — the dispatcher pre-claims via the picker; the override slot (RunNow/ContinueTask) must call StartRunningAsync if a task is not pre-claimed. +- **Claim before create** — `TaskRunner.RunAsync`'s unclaimed path calls `StartRunningAsync` + *before* `PrepareRunDirectoryAsync`. RunNow racing the picker for the same Queued row used to + create the worktree first and only claim afterwards, so the losing dispatch could hit + WorktreeManager's branch-collision self-heal and force-remove the winner's live worktree + mid-run. `OverrideSlotService.RunNow` also fast-rejects a task already Running in the DB + (defense in depth; the picker's atomic SQL claim is the real arbiter either way). + `RunCancellationRegistry.Register` refuses (and logs) a second registration for the same task + id instead of silently overwriting the first, so a losing dispatch's cleanup can't unregister + the winner's CTS out from under it. - **Terminal writes** — use `CancellationToken.None`; a task is never left Running after crash/cancel. - **Per-run MCP tokens** — each run gets a unique token scoping tool access; unregistered on end. - **Auto-retry** — one automatic retry if a session exists and the first run failed. diff --git a/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs b/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs index 4f06d8e5..57454661 100644 --- a/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs +++ b/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs @@ -33,9 +33,15 @@ public sealed class OverrideSlotService { using (var context = _dbFactory.CreateDbContext()) { - var exists = await new TaskRepository(context).GetByIdAsync(taskId); - if (exists is null) + var task = await new TaskRepository(context).GetByIdAsync(taskId); + if (task is null) throw new KeyNotFoundException($"Task '{taskId}' not found."); + + // Fast-fail precheck only — not the race guard. A task sitting Queued still + // passes here and can race the queue picker's atomic claim; TaskRunner.RunAsync + // resolves that by claiming Running before creating any resources. + if (task.Status == Data.Models.TaskStatus.Running) + throw new InvalidOperationException("task is already running"); } StartInSlot(taskId, ct => RunInSlotAsync(taskId, ct), "RunInSlotAsync failed for task {TaskId}"); diff --git a/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs b/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs index 462395c8..210585cf 100644 --- a/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs +++ b/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs @@ -10,8 +10,23 @@ namespace ClaudeDo.Worker.Queue; public sealed class RunCancellationRegistry { private readonly ConcurrentDictionary _running = new(StringComparer.Ordinal); + private readonly ILogger _logger; - public void Register(string taskId, CancellationTokenSource cts) => _running[taskId] = cts; + public RunCancellationRegistry(ILogger logger) => _logger = logger; + + /// Registers the CTS driving 's run. Refuses (and logs) instead of + /// silently overwriting when a CTS is already registered for this task id — an overwrite + /// would mean a double-dispatch is in flight, and the caller that lost the race would then + /// unregister the winner's CTS out from under it, leaving CancelAsync unable to reach the + /// still-running process. + public bool Register(string taskId, CancellationTokenSource cts) + { + if (_running.TryAdd(taskId, cts)) return true; + + _logger.LogWarning( + "Task {TaskId} already has a registered run; refusing to overwrite (double-dispatch?)", taskId); + return false; + } /// Removes the registration only if is still the registered /// one — a re-run may already have registered a newer CTS under the same task id. diff --git a/src/ClaudeDo.Worker/Runner/TaskRunner.cs b/src/ClaudeDo.Worker/Runner/TaskRunner.cs index 386d591f..423ec493 100644 --- a/src/ClaudeDo.Worker/Runner/TaskRunner.cs +++ b/src/ClaudeDo.Worker/Runner/TaskRunner.cs @@ -87,6 +87,32 @@ public sealed class TaskRunner attachmentPaths = attachments.Select(a => Path.Combine(_attachments.TaskDir(task.Id), a.FileName)).ToList(); } + var now = DateTime.UtcNow; + // Claim Running before creating any resources (worktree, MCP token file, ...): + // the queue picker claims Queued→Running atomically (incl. StartedAt) before + // dispatching, so only unclaimed dispatches (override slot) need to claim here. + // Claiming first means a losing double-dispatch (RunNow racing the picker for the + // same row) bails out immediately instead of creating a worktree the winner then + // has to self-heal past. + if (!alreadyClaimed) + { + var startResult = await _state.StartRunningAsync(task.Id, now, ct); + if (!startResult.Ok) + { + _logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", task.Id, startResult.Reason); + return; + } + } + else + { + // Queue-claimed dispatches skip StartRunningAsync (the atomic SQL claim in + // QueuePicker already flipped the row to Running), so it never broadcasts + // TaskUpdated for this transition. Send it here so the task-list badge flips + // live instead of staying on "Queued" until the run finishes. + await _broadcaster.TaskUpdated(task.Id); + } + await _broadcaster.TaskStarted(slot, task.Id, now); + // Determine working directory: worktree or sandbox. var prep = await PrepareRunDirectoryAsync(task, list, ct); if (prep.FailureReason is not null) @@ -117,28 +143,6 @@ public sealed class TaskRunner : "mcp__claudedo_run__AskUser", }; - var now = DateTime.UtcNow; - // The queue picker claims Queued→Running atomically (incl. StartedAt) before - // dispatching; only unclaimed dispatches (override slot) claim here. - if (!alreadyClaimed) - { - var startResult = await _state.StartRunningAsync(task.Id, now, ct); - if (!startResult.Ok) - { - _logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", task.Id, startResult.Reason); - return; - } - } - else - { - // Queue-claimed dispatches skip StartRunningAsync (the atomic SQL claim in - // QueuePicker already flipped the row to Running), so it never broadcasts - // TaskUpdated for this transition. Send it here so the task-list badge flips - // live instead of staying on "Queued" until the run finishes. - await _broadcaster.TaskUpdated(task.Id); - } - await _broadcaster.TaskStarted(slot, task.Id, now); - await _skillSeeder.SeedAsync(runDir, resolvedConfig.SkillNames, wtCtx is not null, ct); // Build prompt: title + description + only the OPEN sub-tasks (resolved ones are dropped). diff --git a/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs b/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs index c8866d34..07374156 100644 --- a/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs @@ -74,7 +74,7 @@ public sealed class AddSubtaskToolTests : IDisposable NullLogger.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); var waker = new ClaudeDo.Worker.Queue.QueueWaker(); var picker = new ClaudeDo.Worker.Queue.QueuePicker(dbFactory); - var runCancels = new RunCancellationRegistry(); + var runCancels = new RunCancellationRegistry(NullLogger.Instance); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); var queue = new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels, new FakeUsageGate(), new UsageState(), broadcaster); diff --git a/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs b/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs index 745b8a18..361220fd 100644 --- a/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs @@ -94,7 +94,7 @@ public sealed class BatchMcpToolsTests : IDisposable var state = TaskStateServiceBuilder.Build(dbFactory).State; var runner = new TaskRunner(new FakeClaudeProcess(), dbFactory, broadcaster, wtManager, new ClaudeArgsBuilder(), cfg, NullLogger.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); - var runCancels = new RunCancellationRegistry(); + var runCancels = new RunCancellationRegistry(NullLogger.Instance); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state, runCancels, diff --git a/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs b/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs index 08c8d6ca..e6d306d3 100644 --- a/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs @@ -160,7 +160,7 @@ public sealed class ExternalMcpServiceTests : IDisposable NullLogger.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); var waker = new ClaudeDo.Worker.Queue.QueueWaker(); var picker = new ClaudeDo.Worker.Queue.QueuePicker(dbFactory); - var runCancels = new RunCancellationRegistry(); + var runCancels = new RunCancellationRegistry(NullLogger.Instance); var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels, new FakeUsageGate(), new UsageState(), broadcaster); diff --git a/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs b/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs index 283a193e..a6399ab0 100644 --- a/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs +++ b/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs @@ -25,7 +25,7 @@ public static class TaskStateServiceBuilder var hub = new CapturingHubContext(); var broadcaster = new HubBroadcaster(hub); var waker = new CountingQueueWaker(); - var runCancels = new RunCancellationRegistry(); + var runCancels = new RunCancellationRegistry(NullLogger.Instance); TaskStateService? state = null; var chain = new PlanningChainCoordinator(dbFactory, () => state!); diff --git a/tests/ClaudeDo.Worker.Tests/Queue/OverrideSlotServiceTests.cs b/tests/ClaudeDo.Worker.Tests/Queue/OverrideSlotServiceTests.cs new file mode 100644 index 00000000..5a437eaa --- /dev/null +++ b/tests/ClaudeDo.Worker.Tests/Queue/OverrideSlotServiceTests.cs @@ -0,0 +1,89 @@ +using ClaudeDo.Data; +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 Microsoft.Extensions.Logging.Abstractions; +using TaskStatus = ClaudeDo.Data.Models.TaskStatus; + +namespace ClaudeDo.Worker.Tests.Queue; + +/// OverrideSlotService.RunNow's own precheck — distinct from QueueService.EnsureNotInQueueSlot, +/// which only guards a task already tracked in its in-memory queue-slot dict. This covers a +/// task the DB already shows Running (e.g. the queue picker's atomic claim landed) regardless +/// of local slot bookkeeping. The real race-safety net is TaskRunner.RunAsync claiming Running +/// before creating any resources; this precheck just fails fast for the common case. +public sealed class OverrideSlotServiceTests : IDisposable +{ + private readonly DbFixture _db = new(); + private readonly string _tempDir; + private readonly WorkerConfig _cfg; + + public OverrideSlotServiceTests() + { + _tempDir = Path.Combine(Path.GetTempPath(), $"cd_override_{Guid.NewGuid():N}"); + Directory.CreateDirectory(_tempDir); + _cfg = new WorkerConfig { SandboxRoot = _tempDir, LogRoot = _tempDir }; + } + + public void Dispose() { _db.Dispose(); try { Directory.Delete(_tempDir, true); } catch { } } + + private OverrideSlotService BuildService() + { + var dbFactory = _db.CreateFactory(); + var state = TaskStateServiceBuilder.Build(dbFactory).State; + var wt = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger.Instance); + var runner = new TaskRunner(new FakeClaudeProcess(), dbFactory, new HubBroadcaster(new CapturingHubContext()), wt, + new ClaudeArgsBuilder(), _cfg, NullLogger.Instance, state, new TaskRunTokenRegistry(), + new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); + var runCancels = new RunCancellationRegistry(NullLogger.Instance); + return new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); + } + + [Fact] + public async Task RunNow_TaskAlreadyRunningInDb_ThrowsWithoutDispatching() + { + string listId = Guid.NewGuid().ToString(), taskId = Guid.NewGuid().ToString(); + using (var ctx = _db.CreateContext()) + { + ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", 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 service = BuildService(); + + var ex = await Assert.ThrowsAsync(() => service.RunNow(taskId)); + Assert.Contains("already running", ex.Message); + Assert.Null(service.CurrentSlot); + } + + [Fact] + public async Task RunNow_TaskQueued_DoesNotThrow() + { + string listId = Guid.NewGuid().ToString(), taskId = Guid.NewGuid().ToString(); + using (var ctx = _db.CreateContext()) + { + ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", CreatedAt = DateTime.UtcNow }); + ctx.Tasks.Add(new TaskEntity + { + Id = taskId, ListId = listId, Title = "T", Status = TaskStatus.Queued, + CreatedAt = DateTime.UtcNow, + }); + await ctx.SaveChangesAsync(); + } + + var service = BuildService(); + + // The dispatch itself races with FakeClaudeProcess's immediate completion in the + // background, so this only asserts the precheck doesn't reject a claimable task. + await service.RunNow(taskId); + } +} diff --git a/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs b/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs index 7b682a07..75b54a1f 100644 --- a/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs @@ -1,4 +1,5 @@ using ClaudeDo.Worker.Queue; +using Microsoft.Extensions.Logging.Abstractions; namespace ClaudeDo.Worker.Tests.Queue; @@ -7,9 +8,9 @@ public sealed class RunCancellationRegistryTests [Fact] public void TryCancel_RegisteredTask_CancelsAndReturnsTrue() { - var sut = new RunCancellationRegistry(); + var sut = new RunCancellationRegistry(NullLogger.Instance); using var cts = new CancellationTokenSource(); - sut.Register("t1", cts); + Assert.True(sut.Register("t1", cts)); Assert.True(sut.TryCancel("t1")); Assert.True(cts.IsCancellationRequested); @@ -18,30 +19,50 @@ public sealed class RunCancellationRegistryTests [Fact] public void TryCancel_UnknownTask_ReturnsFalse() { - var sut = new RunCancellationRegistry(); + var sut = new RunCancellationRegistry(NullLogger.Instance); Assert.False(sut.TryCancel("nope")); } + [Fact] + public void Register_SecondCallForSameTask_RefusesAndKeepsFirstRegistration() + { + // A double-dispatch (e.g. RunNow racing the queue picker for the same task) must not + // let the second registration silently clobber the first — that would leave CancelAsync + // pointing at the wrong (or already-finished) CTS. + var sut = new RunCancellationRegistry(NullLogger.Instance); + using var first = new CancellationTokenSource(); + using var second = new CancellationTokenSource(); + + Assert.True(sut.Register("t1", first)); + Assert.False(sut.Register("t1", second)); + + Assert.True(sut.TryCancel("t1")); + Assert.True(first.IsCancellationRequested); + Assert.False(second.IsCancellationRequested); + } + [Fact] public void Unregister_RemovesOnlyTheGivenRegistration() { - var sut = new RunCancellationRegistry(); - using var stale = new CancellationTokenSource(); - using var current = new CancellationTokenSource(); + var sut = new RunCancellationRegistry(NullLogger.Instance); + using var first = new CancellationTokenSource(); - sut.Register("t1", stale); - sut.Register("t1", current); // re-run replaced the registration - sut.Unregister("t1", stale); // late cleanup of the old slot must not evict the new one + sut.Register("t1", first); + sut.Unregister("t1", first); + + // Slot is free again once the run that held it cleans up properly, so a genuine + // sequential re-run can register a fresh CTS under the same task id. + using var second = new CancellationTokenSource(); + Assert.True(sut.Register("t1", second)); Assert.True(sut.TryCancel("t1")); - Assert.True(current.IsCancellationRequested); - Assert.False(stale.IsCancellationRequested); + Assert.True(second.IsCancellationRequested); } [Fact] public void TryCancel_DisposedCts_ReturnsFalse() { - var sut = new RunCancellationRegistry(); + var sut = new RunCancellationRegistry(NullLogger.Instance); var cts = new CancellationTokenSource(); sut.Register("t1", cts); cts.Dispose(); diff --git a/tests/ClaudeDo.Worker.Tests/Runner/ConcurrentDispatchRaceTests.cs b/tests/ClaudeDo.Worker.Tests/Runner/ConcurrentDispatchRaceTests.cs new file mode 100644 index 00000000..7bcaad4b --- /dev/null +++ b/tests/ClaudeDo.Worker.Tests/Runner/ConcurrentDispatchRaceTests.cs @@ -0,0 +1,122 @@ +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 Microsoft.Extensions.Logging.Abstractions; +using TaskStatus = ClaudeDo.Data.Models.TaskStatus; + +namespace ClaudeDo.Worker.Tests.Runner; + +/// Regression test for the RunNow/queue-picker double-dispatch race: a Queued task +/// dispatched both via the override slot (RunNow, unclaimed) and the queue picker (atomic +/// SQL Queued->Running claim) must not create its worktree twice. Before the fix, +/// TaskRunner.RunAsync created the worktree (PrepareRunDirectoryAsync) BEFORE claiming +/// Running, so the losing dispatch reached WorktreeManager's "branch already exists" +/// self-heal and force-removed the winner's live worktree out from under its still-running +/// process. The fix claims Running first and bails out before touching git when the claim +/// is rejected. +public sealed class ConcurrentDispatchRaceTests : IDisposable +{ + private readonly DbFixture _db = new(); + private readonly GitRepoFixture? _repo; + private readonly string _tempDir; + private readonly WorkerConfig _cfg; + + private static bool GitAvailable => GitRepoFixture.IsGitAvailable(); + + public ConcurrentDispatchRaceTests() + { + _tempDir = Path.Combine(Path.GetTempPath(), $"cd_race_{Guid.NewGuid():N}"); + Directory.CreateDirectory(_tempDir); + _cfg = new WorkerConfig { SandboxRoot = _tempDir, LogRoot = _tempDir }; + if (GitAvailable) _repo = new GitRepoFixture(); + } + + public void Dispose() + { + _db.Dispose(); + _repo?.Dispose(); + try { Directory.Delete(_tempDir, true); } catch { /* best effort */ } + } + + [Fact] + public async Task RunNow_RacingQueuePicker_LoserBailsBeforeTouchingWorktree() + { + if (!GitAvailable) { Assert.True(true, "git not available -- skipping"); return; } + + var dbFactory = _db.CreateFactory(); + string listId = Guid.NewGuid().ToString(), taskId = Guid.NewGuid().ToString(); + using (var ctx = _db.CreateContext()) + { + ctx.Lists.Add(new ListEntity { Id = listId, Name = "L", WorkingDir = _repo!.RepoDir, CreatedAt = DateTime.UtcNow }); + ctx.Tasks.Add(new TaskEntity + { + Id = taskId, ListId = listId, Title = "Racing task", Status = TaskStatus.Queued, + CreatedAt = DateTime.UtcNow, + }); + await ctx.SaveChangesAsync(); + } + + var processStarted = new TaskCompletionSource(); + var releaseProcess = new TaskCompletionSource(); + var fake = new FakeClaudeProcess(async (_, _, _, _, _) => + { + processStarted.TrySetResult(); + await releaseProcess.Task; + return new RunResult { ExitCode = 0, ResultMarkdown = "ok" }; + }); + + var state = TaskStateServiceBuilder.Build(dbFactory).State; + var wt = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger.Instance); + var runner = new TaskRunner(fake, dbFactory, new HubBroadcaster(new CapturingHubContext()), wt, + new ClaudeArgsBuilder(), _cfg, NullLogger.Instance, state, new TaskRunTokenRegistry(), + new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader()); + + // Winner: mirrors QueuePicker's atomic Queued->Running claim, then a queue-slot + // dispatch (alreadyClaimed: true, matching QueueService.RunInSlotAsync). + var picker = new QueuePicker(dbFactory); + var claimed = await picker.ClaimNextAsync(DateTime.UtcNow, CancellationToken.None); + Assert.NotNull(claimed); + + var winnerTask = runner.RunAsync(claimed!, "queue", CancellationToken.None, alreadyClaimed: true); + await processStarted.Task; // worktree exists, "Claude" is now running in it + + WorktreeEntity? wtRowBefore; + using (var ctx = _db.CreateContext()) + wtRowBefore = await new WorktreeRepository(ctx).GetByTaskIdAsync(taskId); + Assert.NotNull(wtRowBefore); + + var markerPath = Path.Combine(wtRowBefore!.Path, "in-progress.marker"); + File.WriteAllText(markerPath, "winner's work"); + + // Loser: a RunNow-style dispatch (unclaimed) of the SAME task arriving after the + // picker already claimed it. Before the fix this would recreate the worktree via + // WorktreeManager's self-heal, wiping the winner's live directory (and this marker + // file) out from under the still-running process. + TaskEntity taskForOverride; + using (var ctx = _db.CreateContext()) + taskForOverride = (await new TaskRepository(ctx).GetByIdAsync(taskId))!; + + await runner.RunAsync(taskForOverride, "override", CancellationToken.None); + + Assert.True(File.Exists(markerPath), "loser dispatch touched the winner's live worktree"); + Assert.Equal(1, fake.CallCount); + + releaseProcess.TrySetResult(); + await winnerTask; + + using var verify = _db.CreateContext(); + Assert.Single(await new TaskRunRepository(verify).GetByTaskIdAsync(taskId)); + + var wtRowAfter = await new WorktreeRepository(verify).GetByTaskIdAsync(taskId); + Assert.NotNull(wtRowAfter); + Assert.Equal(WorktreeState.Active, wtRowAfter!.State); + Assert.Equal(wtRowBefore.Path, wtRowAfter.Path); + Assert.True(Directory.Exists(wtRowAfter.Path)); + } +}