Merge branch 'claudedo/9c0bff67147148f3beeee180e70b5503'

This commit is contained in:
mika kuns
2026-08-06 13:46:52 +02:00
11 changed files with 312 additions and 44 deletions
+14 -3
View File
@@ -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.
@@ -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}");
@@ -10,8 +10,23 @@ namespace ClaudeDo.Worker.Queue;
public sealed class RunCancellationRegistry
{
private readonly ConcurrentDictionary<string, CancellationTokenSource> _running = new(StringComparer.Ordinal);
private readonly ILogger<RunCancellationRegistry> _logger;
public void Register(string taskId, CancellationTokenSource cts) => _running[taskId] = cts;
public RunCancellationRegistry(ILogger<RunCancellationRegistry> logger) => _logger = logger;
/// Registers the CTS driving <paramref name="taskId"/>'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 <paramref name="cts"/> is still the registered
/// one — a re-run may already have registered a newer CTS under the same task id.
+26 -22
View File
@@ -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).
@@ -74,7 +74,7 @@ public sealed class AddSubtaskToolTests : IDisposable
NullLogger<TaskRunner>.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<RunCancellationRegistry>.Instance);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, runCancels);
var queue = new QueueService(dbFactory, runner, cfg, NullLogger<QueueService>.Instance, waker, picker, overrideSlot, state, runCancels,
new FakeUsageGate(), new UsageState(), broadcaster);
+1 -1
View File
@@ -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<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
var runCancels = new RunCancellationRegistry();
var runCancels = new RunCancellationRegistry(NullLogger<RunCancellationRegistry>.Instance);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, runCancels);
return new QueueService(dbFactory, runner, cfg, NullLogger<QueueService>.Instance,
new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state, runCancels,
@@ -160,7 +160,7 @@ public sealed class ExternalMcpServiceTests : IDisposable
NullLogger<TaskRunner>.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<RunCancellationRegistry>.Instance);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, runCancels);
return new QueueService(dbFactory, runner, cfg, NullLogger<QueueService>.Instance, waker, picker, overrideSlot, state, runCancels,
new FakeUsageGate(), new UsageState(), broadcaster);
@@ -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<RunCancellationRegistry>.Instance);
TaskStateService? state = null;
var chain = new PlanningChainCoordinator(dbFactory, () => state!);
@@ -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<WorktreeManager>.Instance);
var runner = new TaskRunner(new FakeClaudeProcess(), dbFactory, new HubBroadcaster(new CapturingHubContext()), wt,
new ClaudeArgsBuilder(), _cfg, NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(),
new AttachmentStore(), new FakeSessionSkillSeeder(), new FakeTranscriptUsageReader());
var runCancels = new RunCancellationRegistry(NullLogger<RunCancellationRegistry>.Instance);
return new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.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<InvalidOperationException>(() => 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);
}
}
@@ -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<RunCancellationRegistry>.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<RunCancellationRegistry>.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<RunCancellationRegistry>.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<RunCancellationRegistry>.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<RunCancellationRegistry>.Instance);
var cts = new CancellationTokenSource();
sut.Register("t1", cts);
cts.Dispose();
@@ -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<WorktreeManager>.Instance);
var runner = new TaskRunner(fake, dbFactory, new HubBroadcaster(new CapturingHubContext()), wt,
new ClaudeArgsBuilder(), _cfg, NullLogger<TaskRunner>.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));
}
}