releases/ClaudeDo
2
0
Fork 0
Code Issues 1 Pull Requests 2 Actions Packages Projects Releases 12 Wiki Activity

fix(worker): queue dispatches skip the StartRunning re-claim

Browse Source 
The picker claims Queued->Running atomically before dispatch; the new
StartRunningAsync guard then rejected every queue-dispatched run. Add
alreadyClaimed to RunAsync/ContinueAsync (queue passes true, override
slot keeps the guard) and align the routing tests.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
mika kuns
2026-06-09 23:59:56 +02:00
parent 0cba9f9640
commit 74ca2e0dcd
3 changed files with 24 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
src/ClaudeDo.Worker/Queue/QueueService.cs
View File

@@ -191,14 +191,14 @@ public sealed class QueueService : BackgroundService
if (sessionId is not null)
{
await _runner.ContinueAsync(taskId, feedback, "queue", ct);
await _runner.ContinueAsync(taskId, feedback, "queue", ct, alreadyClaimed: true);
}
else
{
task.Description = string.IsNullOrWhiteSpace(task.Description)
? $"Reviewer feedback: {feedback}"
: $"{task.Description}\n\nReviewer feedback: {feedback}";
await _runner.RunAsync(task, "queue", ct);
await _runner.RunAsync(task, "queue", ct, alreadyClaimed: true);
}
// Clear the consumed feedback only once the run reached a successful
@@ -212,7 +212,7 @@ public sealed class QueueService : BackgroundService
return;
}
await _runner.RunAsync(task, "queue", ct);
await _runner.RunAsync(task, "queue", ct, alreadyClaimed: true);
}
catch (Exception ex)
{

29
src/ClaudeDo.Worker/Runner/TaskRunner.cs
View File

@@ -44,7 +44,7 @@ public sealed class TaskRunner
_tokens = tokens;
}
public async Task RunAsync(TaskEntity task, string slot, CancellationToken ct)
public async Task RunAsync(TaskEntity task, string slot, CancellationToken ct, bool alreadyClaimed = false)
{
string? mcpToken = null;
string? mcpConfigPath = null;
@@ -98,11 +98,16 @@ public sealed class TaskRunner
}
var now = DateTime.UtcNow;
var startResult = await _state.StartRunningAsync(task.Id, now, ct);
if (!startResult.Ok)
// The queue picker claims Queued→Running atomically (incl. StartedAt) before
// dispatching; only unclaimed dispatches (override slot) claim here.
if (!alreadyClaimed)
{
_logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", task.Id, startResult.Reason);
return;
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;
}
}
await _broadcaster.TaskStarted(slot, task.Id, now);
@@ -167,7 +172,7 @@ public sealed class TaskRunner
}
}
public async Task ContinueAsync(string taskId, string followUpPrompt, string slot, CancellationToken ct)
public async Task ContinueAsync(string taskId, string followUpPrompt, string slot, CancellationToken ct, bool alreadyClaimed = false)
{
TaskEntity task;
TaskRunEntity lastRun;
@@ -213,11 +218,15 @@ public sealed class TaskRunner
}
var now = DateTime.UtcNow;
var startResult = await _state.StartRunningAsync(taskId, now, ct);
if (!startResult.Ok)
// See RunAsync: queue dispatches arrive pre-claimed by the picker.
if (!alreadyClaimed)
{
_logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", taskId, startResult.Reason);
return;
var startResult = await _state.StartRunningAsync(taskId, now, ct);
if (!startResult.Ok)
{
_logger.LogWarning("Task {TaskId} skipped: StartRunningAsync rejected ({Reason})", taskId, startResult.Reason);
return;
}
}
await _broadcaster.TaskStarted(slot, taskId, now);

4
tests/ClaudeDo.Worker.Tests/Runner/StandaloneChildrenRoutingTests.cs
View File

@@ -47,7 +47,7 @@ public sealed class StandaloneChildrenRoutingTests : IDisposable
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry());
using (var ctx = _db.CreateContext())
await runner.RunAsync((await new TaskRepository(ctx).GetByIdAsync("p1"))!, "slot-1", default);
await runner.RunAsync((await new TaskRepository(ctx).GetByIdAsync("p1"))!, "slot-1", default, alreadyClaimed: true);
using var verify = _db.CreateContext();
var repo = new TaskRepository(verify);
@@ -74,7 +74,7 @@ public sealed class StandaloneChildrenRoutingTests : IDisposable
new ClaudeArgsBuilder(), _cfg, NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry());
using (var ctx = _db.CreateContext())
await runner.RunAsync((await new TaskRepository(ctx).GetByIdAsync("solo"))!, "slot-1", default);
await runner.RunAsync((await new TaskRepository(ctx).GetByIdAsync("solo"))!, "slot-1", default, alreadyClaimed: true);
using var verify = _db.CreateContext();
Assert.Equal(TaskStatus.WaitingForReview, (await new TaskRepository(verify).GetByIdAsync("solo"))!.Status);