fix(worker): kill cancelled runs' processes and make MCP approve actually merge

- CancelAsync now signals the running Claude process of the cancelled task and
  its cascaded children via the new RunCancellationRegistry (queue + override
  slots register their CTS there) instead of only flipping DB state.
- external MCP review_task 'approve' now mirrors the hub's ApproveReview:
  unit merge for parents, ApproveAndMergeAsync for childless tasks, optional
  targetBranch; ReviewTaskResult carries mergeStatus/conflicts.
This commit is contained in:
mika kuns
2026-07-23 20:24:36 +02:00
parent 451afc80f8
commit fee69998f8
15 changed files with 330 additions and 46 deletions
+2 -2
View File
@@ -7,7 +7,7 @@ ASP.NET Core hosted service that executes tasks via Claude CLI in isolated envir
```
Worker/
State/ — TaskStateService + TransitionResult (sole owner of Status/PlanningPhase/BlockedBy writes)
Queue/ — IQueueWaker, IQueuePicker, QueueService (BackgroundService), OverrideSlotService
Queue/ — IQueueWaker, IQueuePicker, QueueService (BackgroundService), OverrideSlotService, RunCancellationRegistry (taskId → running-run CTS; lets TaskStateService.CancelAsync kill the process of a cancelled task/child without a DI cycle)
Lifecycle/ — StaleTaskRecovery, TaskResetService, TaskMergeService, ClaudeCliPreflight, OrphanRecovery, PlanningLineageRecovery, AttachmentOrphanRecovery (startup sweep: deletes any `attachments/<taskId>/` dirs whose task no longer exists)
Worktrees/ — WorktreeMaintenanceService
Agents/ — AgentFileService, DefaultAgentSeeder
@@ -33,7 +33,7 @@ Interfaces (e.g. `IQueueWaker`, `IPrimeClock`, `ITaskStateService`) live in an `
- **OverrideSlotService** — owns `RunNow` / `ContinueTask`; goes through `TaskStateService.StartRunningAsync` (caller-driven, serialized by slot lock).
- **StaleTaskRecovery** — startup-only service; calls `TaskStateService.RecoverStaleRunningAsync` to flip orphaned `Running` rows to `Failed`.
- **External/*** — always-on MCP tools for general Claude sessions, scoped to *starting* and *observing* sessions (no worktree/merge, multi-turn, planning, or app-settings writes). Auth via optional `X-ClaudeDo-Key` header. Registered explicitly in `Program.cs`'s external app via `.WithTools<T>()`. Every optional/filter parameter across these tools must carry a C# default value (e.g. `string? status = null`) — the MCP schema only marks a parameter optional when it has one; nullability alone doesn't do it (`ExternalMcpToolSchemaTests` guards this by reflection). `ExternalMcpExceptionFilter.Wrap` is registered as a call-tool filter so `InvalidOperationException`/`ArgumentException` messages survive as `McpException` — otherwise the SDK's own catch-all replaces any non-`McpException` with a generic "An error occurred invoking 'X'." Organized by concern:
- `ExternalMcpService` — task CRUD + execution: `ListTaskLists`, `ListTasks`, `GetTask`, `AddTask`, `AddSubtask`, `UpdateTask`, `UpdateTaskStatus` (`Idle` / `Queued`), `GetTaskStatusValues`, `ReviewTask` (`approve` / `reject_rerun` / `reject_park` / `cancel` for a WaitingForReview task), `RunTaskNow`, `ContinueTask`, `CancelTask`, `DeleteTask`; worktree/git: `GetTaskWorktree`, `GetTaskDiff`, `MergeTask`, `ListWorktrees`, `CleanupTaskWorktree`
- `ExternalMcpService` — task CRUD + execution: `ListTaskLists`, `ListTasks`, `GetTask`, `AddTask`, `AddSubtask`, `UpdateTask`, `UpdateTaskStatus` (`Idle` / `Queued`), `GetTaskStatusValues`, `ReviewTask` (`approve` / `reject_rerun` / `reject_park` / `cancel` for a WaitingForReview task; approve is review+merge exactly like the hub's `ApproveReview` — unit merge for parents, worktree merge into optional `targetBranch` for childless tasks, conflicts reported in the `ReviewTaskResult`), `RunTaskNow`, `ContinueTask`, `CancelTask`, `DeleteTask`; worktree/git: `GetTaskWorktree`, `GetTaskDiff`, `MergeTask`, `ListWorktrees`, `CleanupTaskWorktree`
- `BatchMcpTools` — best-effort batch variants that loop the `ExternalMcpService` single-entity methods (sequential — the scoped DbContext is not thread-safe; merge/review stay single-task): `BatchGetTasks`, `BatchAddTasks`, `BatchUpdateTaskStatus`, `BatchCancelTasks`, `BatchDeleteTasks`, `BatchSetMyDay`, `BatchCleanupTaskWorktrees`. Every tool returns a per-item result array ({ id/index, ok, error?, … }) — a failing item never aborts the rest — and rejects batches over 100 items.
- `ListMcpTools``CreateList`, `UpdateList`, `DeleteList`
- `ConfigMcpTools``GetListConfig`, `SetListConfig`, `GetTaskConfig`, `SetTaskConfig`
+59 -14
View File
@@ -6,6 +6,7 @@ using ClaudeDo.Data.Models;
using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Lifecycle;
using ClaudeDo.Worker.Planning;
using ClaudeDo.Worker.Queue;
using ClaudeDo.Worker.State;
using ClaudeDo.Worker.Worktrees;
@@ -18,6 +19,7 @@ namespace ClaudeDo.Worker.External;
public sealed record TaskListDto(string Id, string Name, string? WorkingDir);
public sealed record DeleteTaskResult(bool Deleted, string Id);
public sealed record CancelTaskResult(bool Cancelled, string Id);
public sealed record ReviewTaskResult(TaskDto Task, string? MergeStatus, IReadOnlyList<string> MergeConflicts, string? MergeMessage);
public sealed record StatusValueDto(string Status, string Meaning);
public sealed record TaskDto(
@@ -72,6 +74,7 @@ public sealed class ExternalMcpService
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly WorktreeMaintenanceService _maintenance;
private readonly TaskMergeService _merge;
private readonly PlanningMergeOrchestrator _planningMerge;
public ExternalMcpService(
TaskRepository tasks,
@@ -82,7 +85,8 @@ public sealed class ExternalMcpService
GitService git,
IDbContextFactory<ClaudeDoDbContext> dbFactory,
WorktreeMaintenanceService maintenance,
TaskMergeService merge)
TaskMergeService merge,
PlanningMergeOrchestrator planningMerge)
{
_tasks = tasks;
_lists = lists;
@@ -93,6 +97,7 @@ public sealed class ExternalMcpService
_dbFactory = dbFactory;
_maintenance = maintenance;
_merge = merge;
_planningMerge = planningMerge;
}
[McpServerTool, Description("List all task lists available in ClaudeDo.")]
@@ -292,34 +297,74 @@ public sealed class ExternalMcpService
[McpServerTool, Description(
"Review a task that is WaitingForReview. " +
"decision='approve' → Done. " +
"decision='approve' → review+merge, exactly like the UI's Approve: a childless task merges its worktree into " +
"targetBranch (default: the repo's current branch) then goes Done; a task with children drives the unit merge " +
"(parent worktree if active + each Done child in order); a task without an active worktree approves straight to Done. " +
"mergeStatus 'conflict' means the merge stopped on conflicts (files listed) — resolve them in the ClaudeDo UI. " +
"decision='reject_rerun' → Queued and re-runs, resuming the agent's session with your feedback as the next turn (feedback is required). " +
"decision='reject_park' → Idle for manual editing (feedback ignored). " +
"decision='cancel' → Cancelled. " +
"Fails if the task is not currently WaitingForReview (except cancel, which also works while Running/Queued).")]
public async Task<TaskDto> ReviewTask(
public async Task<ReviewTaskResult> ReviewTask(
string taskId,
string decision,
string? feedback = null,
string? targetBranch = null,
CancellationToken cancellationToken = default)
{
_ = await _tasks.GetByIdAsync(taskId, cancellationToken)
?? throw new InvalidOperationException($"Task {taskId} not found.");
TransitionResult result = decision.Trim().ToLowerInvariant() switch
string? mergeStatus = null;
IReadOnlyList<string> mergeConflicts = Array.Empty<string>();
string? mergeMessage = null;
if (decision.Trim().ToLowerInvariant() == "approve")
{
"approve" => await _state.ApproveReviewAsync(taskId, cancellationToken),
"reject_rerun" => await _state.RejectToQueueAsync(taskId, feedback ?? "", cancellationToken),
"reject_park" => await _state.RejectToIdleAsync(taskId, cancellationToken),
"cancel" => await _state.CancelAsync(taskId, DateTime.UtcNow, cancellationToken),
_ => throw new InvalidOperationException(
$"Unknown decision '{decision}'. Use approve, reject_rerun, reject_park, or cancel."),
};
// Approve is the single review+merge action — mirror the hub's ApproveReview
// routing instead of only flipping the status (which left branches unmerged).
bool hasChildren;
await using (var ctx = await _dbFactory.CreateDbContextAsync(cancellationToken))
hasChildren = await ctx.Tasks.AnyAsync(t => t.ParentTaskId == taskId, cancellationToken);
if (!result.Ok)
throw new InvalidOperationException(result.Reason ?? "Review action failed.");
if (hasChildren)
{
await _planningMerge.StartAsync(taskId, targetBranch ?? "", cancellationToken);
var parentDone = (await _tasks.GetByIdAsync(taskId, cancellationToken))!.Status == TaskStatus.Done;
mergeStatus = parentDone ? TaskMergeService.StatusMerged : TaskMergeService.StatusConflict;
if (!parentDone)
mergeMessage = "unit merge paused on a conflict — resolve and continue it in the ClaudeDo UI";
}
else
{
var r = await _merge.ApproveAndMergeAsync(taskId, targetBranch ?? "", cancellationToken);
if (r.Status == TaskMergeService.StatusBlocked)
throw new InvalidOperationException(r.ErrorMessage ?? "approve failed");
mergeStatus = r.Status;
mergeConflicts = r.ConflictFiles;
mergeMessage = r.Status == TaskMergeService.StatusConflict
? "merge conflict — the task stays WaitingForReview; resolve it in the ClaudeDo UI"
: r.ErrorMessage;
}
}
else
{
TransitionResult result = decision.Trim().ToLowerInvariant() switch
{
"reject_rerun" => await _state.RejectToQueueAsync(taskId, feedback ?? "", cancellationToken),
"reject_park" => await _state.RejectToIdleAsync(taskId, cancellationToken),
"cancel" => await _state.CancelAsync(taskId, DateTime.UtcNow, cancellationToken),
_ => throw new InvalidOperationException(
$"Unknown decision '{decision}'. Use approve, reject_rerun, reject_park, or cancel."),
};
return ToDto((await _tasks.GetByIdAsync(taskId, cancellationToken))!);
if (!result.Ok)
throw new InvalidOperationException(result.Reason ?? "Review action failed.");
}
return new ReviewTaskResult(
ToDto((await _tasks.GetByIdAsync(taskId, cancellationToken))!),
mergeStatus, mergeConflicts, mergeMessage);
}
[McpServerTool, Description("Immediately run a task in the override execution slot (bypasses the agent queue).")]
+3
View File
@@ -92,6 +92,7 @@ builder.Services.AddSingleton<PlanningChainCoordinator>();
builder.Services.AddSingleton<QueueWaker>();
builder.Services.AddSingleton<IQueueWaker>(sp => sp.GetRequiredService<QueueWaker>());
builder.Services.AddSingleton<IQueuePicker, QueuePicker>();
builder.Services.AddSingleton<RunCancellationRegistry>();
builder.Services.AddSingleton<Func<ITaskStateService>>(sp => () => sp.GetRequiredService<ITaskStateService>());
builder.Services.AddSingleton<ITaskStateService>(sp => new TaskStateService(
@@ -99,6 +100,7 @@ builder.Services.AddSingleton<ITaskStateService>(sp => new TaskStateService(
sp.GetRequiredService<HubBroadcaster>(),
sp.GetRequiredService<IQueueWaker>(),
sp.GetRequiredService<PlanningChainCoordinator>(),
sp.GetRequiredService<RunCancellationRegistry>(),
sp.GetRequiredService<ILogger<TaskStateService>>()));
// Agent file management.
@@ -269,6 +271,7 @@ if (cfg.ExternalMcpPort > 0)
externalBuilder.Services.AddSingleton(app.Services.GetRequiredService<GitService>());
externalBuilder.Services.AddSingleton(app.Services.GetRequiredService<WorktreeMaintenanceService>());
externalBuilder.Services.AddSingleton(app.Services.GetRequiredService<TaskMergeService>());
externalBuilder.Services.AddSingleton(app.Services.GetRequiredService<PlanningMergeOrchestrator>());
externalBuilder.Services.AddScoped<ExternalMcpService>();
externalBuilder.Services.AddScoped<BatchMcpTools>();
externalBuilder.Services.AddScoped<ListMcpTools>();
@@ -10,6 +10,7 @@ public sealed class OverrideSlotService
private readonly IDbContextFactory<ClaudeDoDbContext> _dbFactory;
private readonly TaskRunner _runner;
private readonly ILogger<OverrideSlotService> _logger;
private readonly RunCancellationRegistry _runCancels;
private readonly object _lock = new();
private volatile QueueSlotState? _slot;
@@ -17,11 +18,13 @@ public sealed class OverrideSlotService
public OverrideSlotService(
IDbContextFactory<ClaudeDoDbContext> dbFactory,
TaskRunner runner,
ILogger<OverrideSlotService> logger)
ILogger<OverrideSlotService> logger,
RunCancellationRegistry runCancels)
{
_dbFactory = dbFactory;
_runner = runner;
_logger = logger;
_runCancels = runCancels;
}
public QueueSlotState? CurrentSlot => _slot;
@@ -66,12 +69,14 @@ public sealed class OverrideSlotService
var cts = new CancellationTokenSource();
_slot = new QueueSlotState { TaskId = taskId, StartedAt = DateTime.UtcNow, Cts = cts };
_runCancels.Register(taskId, cts);
_ = work(cts.Token).ContinueWith(t =>
{
if (t.IsFaulted)
_logger.LogError(t.Exception, faultMessage, taskId);
lock (_lock) { _slot = null; }
_runCancels.Unregister(taskId, cts);
cts.Dispose();
}, TaskScheduler.Default);
}
+6 -1
View File
@@ -19,6 +19,7 @@ public sealed class QueueService : BackgroundService
private readonly IQueuePicker _picker;
private readonly OverrideSlotService _override;
private readonly ITaskStateService _state;
private readonly RunCancellationRegistry _runCancels;
private readonly object _lock = new();
private readonly Dictionary<string, QueueSlotState> _queueSlots = new();
@@ -31,7 +32,8 @@ public sealed class QueueService : BackgroundService
QueueWaker waker,
IQueuePicker picker,
OverrideSlotService overrideSlot,
ITaskStateService state)
ITaskStateService state,
RunCancellationRegistry runCancels)
{
_dbFactory = dbFactory;
_runner = runner;
@@ -41,6 +43,7 @@ public sealed class QueueService : BackgroundService
_picker = picker;
_override = overrideSlot;
_state = state;
_runCancels = runCancels;
}
public IReadOnlyList<(string slot, string taskId, DateTime startedAt)> GetActive()
@@ -125,12 +128,14 @@ public sealed class QueueService : BackgroundService
{
var cts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
_queueSlots[task.Id] = new QueueSlotState { TaskId = task.Id, StartedAt = DateTime.UtcNow, Cts = cts };
_runCancels.Register(task.Id, cts);
_ = RunInSlotAsync(task.Id, cts.Token).ContinueWith(t =>
{
if (t.IsFaulted)
_logger.LogError(t.Exception, "RunInSlotAsync failed for task {TaskId} in queue slot", task.Id);
lock (_lock) { _queueSlots.Remove(task.Id); }
_runCancels.Unregister(task.Id, cts);
cts.Dispose();
_waker.Wake(); // Check for next task immediately.
}, TaskScheduler.Default);
@@ -0,0 +1,35 @@
using System.Collections.Concurrent;
namespace ClaudeDo.Worker.Queue;
/// Maps a running task id to the CancellationTokenSource driving its Claude process.
/// QueueService and OverrideSlotService register their slots here so components that
/// cancel tasks DB-side (TaskStateService.CancelAsync and its child cascade) can also
/// stop the process — without depending on the queue services, which would create a
/// DI cycle (QueueService → TaskRunner → ITaskStateService).
public sealed class RunCancellationRegistry
{
private readonly ConcurrentDictionary<string, CancellationTokenSource> _running = new(StringComparer.Ordinal);
public void Register(string taskId, CancellationTokenSource cts) => _running[taskId] = cts;
/// 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.
public void Unregister(string taskId, CancellationTokenSource cts) =>
_running.TryRemove(new KeyValuePair<string, CancellationTokenSource>(taskId, cts));
public bool TryCancel(string taskId)
{
if (!_running.TryGetValue(taskId, out var cts)) return false;
try
{
cts.Cancel();
return true;
}
catch (ObjectDisposedException)
{
// Slot cleanup raced us; the run is already finished.
return false;
}
}
}
@@ -15,6 +15,7 @@ public sealed class TaskStateService : ITaskStateService
private readonly HubBroadcaster _broadcaster;
private readonly IQueueWaker _waker;
private readonly PlanningChainCoordinator _chain;
private readonly RunCancellationRegistry _runCancels;
private readonly ILogger<TaskStateService> _logger;
public TaskStateService(
@@ -22,12 +23,14 @@ public sealed class TaskStateService : ITaskStateService
HubBroadcaster broadcaster,
IQueueWaker waker,
PlanningChainCoordinator chain,
RunCancellationRegistry runCancels,
ILogger<TaskStateService> logger)
{
_dbFactory = dbFactory;
_broadcaster = broadcaster;
_waker = waker;
_chain = chain;
_runCancels = runCancels;
_logger = logger;
}
@@ -255,6 +258,13 @@ public sealed class TaskStateService : ITaskStateService
.SetProperty(t => t.BlockedByTaskId, (string?)null), ct);
}
// Also stop the actual Claude processes: the DB flip above doesn't reach a run
// already executing in a queue/override slot — without this it would keep going
// and only fail its final (now-invalid) transition.
_runCancels.TryCancel(taskId);
foreach (var childId in cancelledChildIds)
_runCancels.TryCancel(childId);
await OnChildTerminalAsync(taskId, TaskStatus.Cancelled);
await _broadcaster.TaskUpdated(taskId);
foreach (var childId in cancelledChildIds)
@@ -4,6 +4,7 @@ using ClaudeDo.Data.Repositories;
using ClaudeDo.Worker.External;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Lifecycle;
using ClaudeDo.Worker.Planning;
using ClaudeDo.Worker.Queue;
using ClaudeDo.Worker.Runner;
using ClaudeDo.Worker.Tests.Infrastructure;
@@ -72,14 +73,18 @@ public sealed class AddSubtaskToolTests : IDisposable
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder());
var waker = new ClaudeDo.Worker.Queue.QueueWaker();
var picker = new ClaudeDo.Worker.Queue.QueuePicker(dbFactory);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance);
var queue = new QueueService(dbFactory, runner, cfg, NullLogger<QueueService>.Instance, waker, picker, overrideSlot, state);
var runCancels = new RunCancellationRegistry();
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);
var maintenance = new WorktreeMaintenanceService(dbFactory, git, NullLogger<WorktreeMaintenanceService>.Instance);
var merge = new TaskMergeService(dbFactory, git, broadcaster, TaskStateServiceBuilder.Build(dbFactory).State, NullLogger<TaskMergeService>.Instance);
var merge = new TaskMergeService(dbFactory, git, broadcaster, state, NullLogger<TaskMergeService>.Instance);
var aggregator = new PlanningAggregator(dbFactory, git, NullLogger<PlanningAggregator>.Instance);
var planningMerge = new PlanningMergeOrchestrator(
dbFactory, merge, aggregator, broadcaster, git, state, NullLogger<PlanningMergeOrchestrator>.Instance);
return new ExternalMcpService(
_tasks, _lists, queue, broadcaster,
state,
git, dbFactory, maintenance, merge);
git, dbFactory, maintenance, merge, planningMerge);
}
[Fact]
+11 -5
View File
@@ -6,6 +6,7 @@ using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.External;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Lifecycle;
using ClaudeDo.Worker.Planning;
using ClaudeDo.Worker.Queue;
using ClaudeDo.Worker.Runner;
using ClaudeDo.Worker.Tests.Infrastructure;
@@ -64,11 +65,15 @@ public sealed class BatchMcpToolsTests : IDisposable
var git = new GitService();
var factory = _db.CreateFactory();
var maintenance = new WorktreeMaintenanceService(factory, git, NullLogger<WorktreeMaintenanceService>.Instance);
var merge = new TaskMergeService(factory, git, _broadcaster, TaskStateServiceBuilder.Build(factory).State, NullLogger<TaskMergeService>.Instance);
var state = TaskStateServiceBuilder.Build(factory).State;
var merge = new TaskMergeService(factory, git, _broadcaster, state, NullLogger<TaskMergeService>.Instance);
var aggregator = new PlanningAggregator(factory, git, NullLogger<PlanningAggregator>.Instance);
var planningMerge = new PlanningMergeOrchestrator(
factory, merge, aggregator, _broadcaster, git, state, NullLogger<PlanningMergeOrchestrator>.Instance);
var svc = new ExternalMcpService(
_tasks, _lists, CreateQueue(), _broadcaster,
TaskStateServiceBuilder.Build(factory).State,
git, factory, maintenance, merge);
state,
git, factory, maintenance, merge, planningMerge);
return new BatchMcpTools(svc);
}
@@ -88,9 +93,10 @@ 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());
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance);
var runCancels = new RunCancellationRegistry();
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);
new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state, runCancels);
}
[Fact]
@@ -6,6 +6,7 @@ using ClaudeDo.Worker.Config;
using ClaudeDo.Worker.External;
using ClaudeDo.Worker.Hub;
using ClaudeDo.Worker.Lifecycle;
using ClaudeDo.Worker.Planning;
using ClaudeDo.Worker.Queue;
using ClaudeDo.Worker.Runner;
using ClaudeDo.Worker.Tests.Infrastructure;
@@ -126,11 +127,15 @@ public sealed class ExternalMcpServiceTests : IDisposable
var git = new GitService();
var factory = _db.CreateFactory();
var maintenance = new WorktreeMaintenanceService(factory, git, NullLogger<WorktreeMaintenanceService>.Instance);
var merge = new TaskMergeService(factory, git, _broadcaster, TaskStateServiceBuilder.Build(factory).State, NullLogger<TaskMergeService>.Instance);
var state = TaskStateServiceBuilder.Build(factory).State;
var merge = new TaskMergeService(factory, git, _broadcaster, state, NullLogger<TaskMergeService>.Instance);
var aggregator = new PlanningAggregator(factory, git, NullLogger<PlanningAggregator>.Instance);
var planningMerge = new PlanningMergeOrchestrator(
factory, merge, aggregator, _broadcaster, git, state, NullLogger<PlanningMergeOrchestrator>.Instance);
return new ExternalMcpService(
_tasks, _lists, queue, _broadcaster,
TaskStateServiceBuilder.Build(factory).State,
git, factory, maintenance, merge);
state,
git, factory, maintenance, merge, planningMerge);
}
private QueueService CreateQueue()
@@ -154,8 +159,9 @@ public sealed class ExternalMcpServiceTests : IDisposable
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder());
var waker = new ClaudeDo.Worker.Queue.QueueWaker();
var picker = new ClaudeDo.Worker.Queue.QueuePicker(dbFactory);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance);
return new QueueService(dbFactory, runner, cfg, NullLogger<QueueService>.Instance, waker, picker, overrideSlot, state);
var runCancels = new RunCancellationRegistry();
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, runCancels);
return new QueueService(dbFactory, runner, cfg, NullLogger<QueueService>.Instance, waker, picker, overrideSlot, state, runCancels);
}
[Fact]
@@ -210,9 +216,10 @@ public sealed class ExternalMcpServiceTests : IDisposable
var task = await SeedTaskAsync(listId, status: TaskStatus.WaitingForReview);
var sut = BuildSut(CreateQueue());
var dto = await sut.ReviewTask(task.Id, "approve", null, CancellationToken.None);
var result = await sut.ReviewTask(task.Id, "approve", null, null, CancellationToken.None);
Assert.Equal("Done", dto.Status);
Assert.Equal("Done", result.Task.Status);
Assert.Equal(TaskMergeService.StatusMerged, result.MergeStatus);
}
[Fact]
@@ -223,7 +230,7 @@ public sealed class ExternalMcpServiceTests : IDisposable
var sut = BuildSut(CreateQueue());
await Assert.ThrowsAsync<InvalidOperationException>(() =>
sut.ReviewTask(task.Id, "reject_rerun", null, CancellationToken.None));
sut.ReviewTask(task.Id, "reject_rerun", null, null, CancellationToken.None));
}
[Fact]
@@ -233,9 +240,9 @@ public sealed class ExternalMcpServiceTests : IDisposable
var task = await SeedTaskAsync(listId, status: TaskStatus.WaitingForReview);
var sut = BuildSut(CreateQueue());
var dto = await sut.ReviewTask(task.Id, "reject_rerun", "fix it", CancellationToken.None);
var result = await sut.ReviewTask(task.Id, "reject_rerun", "fix it", null, CancellationToken.None);
Assert.Equal("Queued", dto.Status);
Assert.Equal("Queued", result.Task.Status);
var loaded = await new TaskRepository(_db.CreateContext()).GetByIdAsync(task.Id);
Assert.Equal("fix it", loaded!.ReviewFeedback);
}
@@ -248,7 +255,79 @@ public sealed class ExternalMcpServiceTests : IDisposable
var sut = BuildSut(CreateQueue());
await Assert.ThrowsAsync<InvalidOperationException>(() =>
sut.ReviewTask(task.Id, "bogus", null, CancellationToken.None));
sut.ReviewTask(task.Id, "bogus", null, null, CancellationToken.None));
}
[Fact]
public async Task ReviewTask_Approve_WithActiveWorktree_MergesIntoTargetBranch()
{
if (!GitAvailable) return;
var (task, list, wt) = await SeedWorktreeAsync(status: TaskStatus.WaitingForReview);
File.WriteAllText(Path.Combine(wt.WorktreePath, "feature.txt"), "x\n");
GitRepoFixture.RunGit(wt.WorktreePath, "add", "feature.txt");
GitRepoFixture.RunGit(wt.WorktreePath, "commit", "-m", "add feature");
using (var ctx = _db.CreateContext())
{
var row = ctx.Worktrees.Single(w => w.TaskId == task.Id);
row.HeadCommit = GitRepoFixture.RunGit(wt.WorktreePath, "rev-parse", "HEAD").Trim();
ctx.SaveChanges();
}
var sut = BuildSut(CreateQueue());
var result = await sut.ReviewTask(task.Id, "approve", null, null, CancellationToken.None);
Assert.Equal(TaskMergeService.StatusMerged, result.MergeStatus);
Assert.Equal("Done", result.Task.Status);
Assert.True(File.Exists(Path.Combine(list.WorkingDir!, "feature.txt")));
using var verify = _db.CreateContext();
Assert.Equal(WorktreeState.Merged, verify.Worktrees.Single(w => w.TaskId == task.Id).State);
}
[Fact]
public async Task ReviewTask_Approve_ParentWithChildren_RunsUnitMerge()
{
if (!GitAvailable) return;
var repo = new GitRepoFixture();
_repos.Add(repo);
GitRepoFixture.RunGit(repo.RepoDir, "branch", "-m", "main");
var listId = Guid.NewGuid().ToString();
var parentId = Guid.NewGuid().ToString();
var childId = 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 = parentId, ListId = listId, Title = "plan", CreatedAt = DateTime.UtcNow,
Status = TaskStatus.WaitingForReview, PlanningPhase = PlanningPhase.Finalized });
ctx.Tasks.Add(new TaskEntity { Id = childId, ListId = listId, Title = "child", CreatedAt = DateTime.UtcNow,
ParentTaskId = parentId, Status = TaskStatus.Done, SortOrder = 1 });
var wtPath = Path.Combine(Path.GetTempPath(), $"wt_{Guid.NewGuid():N}");
_worktreeCleanups.Add((repo.RepoDir, wtPath));
var branch = $"claudedo/{childId[..8]}";
GitRepoFixture.RunGit(repo.RepoDir, "worktree", "add", "-b", branch, wtPath, repo.BaseCommit);
File.WriteAllText(Path.Combine(wtPath, "child.txt"), "c\n");
GitRepoFixture.RunGit(wtPath, "add", "child.txt");
GitRepoFixture.RunGit(wtPath, "commit", "-m", "add child.txt");
ctx.Worktrees.Add(new WorktreeEntity
{
TaskId = childId, Path = wtPath, BranchName = branch,
BaseCommit = repo.BaseCommit,
HeadCommit = GitRepoFixture.RunGit(wtPath, "rev-parse", "HEAD").Trim(),
State = WorktreeState.Active, CreatedAt = DateTime.UtcNow,
});
await ctx.SaveChangesAsync();
}
var sut = BuildSut(CreateQueue());
var result = await sut.ReviewTask(parentId, "approve", null, "main", CancellationToken.None);
Assert.Equal(TaskMergeService.StatusMerged, result.MergeStatus);
Assert.Equal("Done", result.Task.Status);
Assert.True(File.Exists(Path.Combine(repo.RepoDir, "child.txt")));
using var verify = _db.CreateContext();
Assert.Equal(WorktreeState.Merged, verify.Worktrees.Single(w => w.TaskId == childId).State);
}
[Fact]
@@ -17,13 +17,15 @@ public static class TaskStateServiceBuilder
PlanningChainCoordinator Chain,
CapturingHubContext Hub,
Func<int> WakeCount,
CountingQueueWaker Waker);
CountingQueueWaker Waker,
RunCancellationRegistry RunCancels);
public static Built Build(IDbContextFactory<ClaudeDoDbContext> dbFactory)
{
var hub = new CapturingHubContext();
var broadcaster = new HubBroadcaster(hub);
var waker = new CountingQueueWaker();
var runCancels = new RunCancellationRegistry();
TaskStateService? state = null;
var chain = new PlanningChainCoordinator(dbFactory, () => state!);
@@ -32,9 +34,10 @@ public static class TaskStateServiceBuilder
broadcaster,
waker,
chain,
runCancels,
NullLogger<TaskStateService>.Instance);
return new Built(state, chain, hub, () => waker.Count, waker);
return new Built(state, chain, hub, () => waker.Count, waker, runCancels);
}
}
@@ -0,0 +1,51 @@
using ClaudeDo.Worker.Queue;
namespace ClaudeDo.Worker.Tests.Queue;
public sealed class RunCancellationRegistryTests
{
[Fact]
public void TryCancel_RegisteredTask_CancelsAndReturnsTrue()
{
var sut = new RunCancellationRegistry();
using var cts = new CancellationTokenSource();
sut.Register("t1", cts);
Assert.True(sut.TryCancel("t1"));
Assert.True(cts.IsCancellationRequested);
}
[Fact]
public void TryCancel_UnknownTask_ReturnsFalse()
{
var sut = new RunCancellationRegistry();
Assert.False(sut.TryCancel("nope"));
}
[Fact]
public void Unregister_RemovesOnlyTheGivenRegistration()
{
var sut = new RunCancellationRegistry();
using var stale = new CancellationTokenSource();
using var current = 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
Assert.True(sut.TryCancel("t1"));
Assert.True(current.IsCancellationRequested);
Assert.False(stale.IsCancellationRequested);
}
[Fact]
public void TryCancel_DisposedCts_ReturnsFalse()
{
var sut = new RunCancellationRegistry();
var cts = new CancellationTokenSource();
sut.Register("t1", cts);
cts.Dispose();
Assert.False(sut.TryCancel("t1"));
}
}
@@ -52,13 +52,14 @@ public sealed class QueueServiceSlotGuardTests : IDisposable
var dbFactory = _db.CreateFactory();
var wtManager = new WorktreeManager(new ClaudeDo.Data.Git.GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
var argsBuilder = new ClaudeArgsBuilder();
var state = TaskStateServiceBuilder.Build(dbFactory).State;
var built = TaskStateServiceBuilder.Build(dbFactory);
var state = built.State;
var runner = new TaskRunner(fake, dbFactory, broadcaster, wtManager, argsBuilder, _cfg,
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder());
_waker = new QueueWaker();
var picker = new QueuePicker(dbFactory);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance);
var service = new QueueService(dbFactory, runner, _cfg, NullLogger<QueueService>.Instance, _waker, picker, overrideSlot, state);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, built.RunCancels);
var service = new QueueService(dbFactory, runner, _cfg, NullLogger<QueueService>.Instance, _waker, picker, overrideSlot, state, built.RunCancels);
return (service, fake);
}
@@ -53,13 +53,14 @@ public sealed class QueueServiceTests : IDisposable
var dbFactory = _db.CreateFactory();
var wtManager = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger<WorktreeManager>.Instance);
var argsBuilder = new ClaudeArgsBuilder();
var state = TaskStateServiceBuilder.Build(dbFactory).State;
var built = TaskStateServiceBuilder.Build(dbFactory);
var state = built.State;
var runner = new TaskRunner(fake, dbFactory, broadcaster, wtManager, argsBuilder, _cfg,
NullLogger<TaskRunner>.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder());
_waker = new QueueWaker();
var picker = new QueuePicker(dbFactory);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance);
var service = new QueueService(dbFactory, runner, _cfg, NullLogger<QueueService>.Instance, _waker, picker, overrideSlot, state);
var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger<OverrideSlotService>.Instance, built.RunCancels);
var service = new QueueService(dbFactory, runner, _cfg, NullLogger<QueueService>.Instance, _waker, picker, overrideSlot, state, built.RunCancels);
return (service, fake);
}
@@ -214,6 +214,41 @@ public sealed class WaitingForChildrenLifecycleTests : IDisposable
Assert.Null(queuedChild.BlockedByTaskId);
}
[Fact]
public async Task Cancelling_WaitingForChildren_parent_stops_running_child_process()
{
using (var ctx = _db.CreateContext())
{
ctx.Lists.Add(new ListEntity { Id = "l1", Name = "L", CreatedAt = DateTime.UtcNow });
ctx.Tasks.Add(new TaskEntity { Id = "par", ListId = "l1", Title = "Parent",
Status = TaskStatus.WaitingForChildren, CreatedAt = DateTime.UtcNow });
ctx.Tasks.Add(new TaskEntity { Id = "c_running", ListId = "l1", Title = "Running child",
Status = TaskStatus.Running, ParentTaskId = "par", CreatedAt = DateTime.UtcNow });
await ctx.SaveChangesAsync();
}
using var childCts = new CancellationTokenSource();
_built.RunCancels.Register("c_running", childCts);
var result = await _built.State.CancelAsync("par", DateTime.UtcNow, default);
Assert.True(result.Ok);
Assert.True(childCts.IsCancellationRequested);
}
[Fact]
public async Task Cancelling_running_task_signals_its_registered_run()
{
var id = await SeedRunningStandaloneAsync();
using var cts = new CancellationTokenSource();
_built.RunCancels.Register(id, cts);
var result = await _built.State.CancelAsync(id, DateTime.UtcNow, default);
Assert.True(result.Ok);
Assert.True(cts.IsCancellationRequested);
}
// ─── FinalizePlanningAsync ────────────────────────────────────────────
private async Task<string> SeedActivePlanningParentAsync(string id = "par")