From fee69998f826cc6dd22fcb783321704e855fc203 Mon Sep 17 00:00:00 2001 From: mika kuns Date: Thu, 23 Jul 2026 20:24:36 +0200 Subject: [PATCH] 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. --- src/ClaudeDo.Worker/CLAUDE.md | 4 +- .../External/ExternalMcpService.cs | 73 ++++++++++--- src/ClaudeDo.Worker/Program.cs | 3 + .../Queue/OverrideSlotService.cs | 7 +- src/ClaudeDo.Worker/Queue/QueueService.cs | 7 +- .../Queue/RunCancellationRegistry.cs | 35 ++++++ src/ClaudeDo.Worker/State/TaskStateService.cs | 10 ++ .../External/AddSubtaskToolTests.cs | 13 ++- .../External/BatchMcpToolsTests.cs | 16 ++- .../External/ExternalMcpServiceTests.cs | 101 ++++++++++++++++-- .../Infrastructure/TaskStateServiceBuilder.cs | 7 +- .../Queue/RunCancellationRegistryTests.cs | 51 +++++++++ .../Services/QueueServiceSlotGuardTests.cs | 7 +- .../Services/QueueServiceTests.cs | 7 +- .../WaitingForChildrenLifecycleTests.cs | 35 ++++++ 15 files changed, 330 insertions(+), 46 deletions(-) create mode 100644 src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs create mode 100644 tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs diff --git a/src/ClaudeDo.Worker/CLAUDE.md b/src/ClaudeDo.Worker/CLAUDE.md index 9faf679e..6bf5fdd4 100644 --- a/src/ClaudeDo.Worker/CLAUDE.md +++ b/src/ClaudeDo.Worker/CLAUDE.md @@ -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//` 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()`. 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` diff --git a/src/ClaudeDo.Worker/External/ExternalMcpService.cs b/src/ClaudeDo.Worker/External/ExternalMcpService.cs index 15d18cc8..da457ba2 100644 --- a/src/ClaudeDo.Worker/External/ExternalMcpService.cs +++ b/src/ClaudeDo.Worker/External/ExternalMcpService.cs @@ -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 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 _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 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 ReviewTask( + public async Task 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 mergeConflicts = Array.Empty(); + 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).")] diff --git a/src/ClaudeDo.Worker/Program.cs b/src/ClaudeDo.Worker/Program.cs index 1fc0758e..2cb4f463 100644 --- a/src/ClaudeDo.Worker/Program.cs +++ b/src/ClaudeDo.Worker/Program.cs @@ -92,6 +92,7 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(sp => sp.GetRequiredService()); builder.Services.AddSingleton(); +builder.Services.AddSingleton(); builder.Services.AddSingleton>(sp => () => sp.GetRequiredService()); builder.Services.AddSingleton(sp => new TaskStateService( @@ -99,6 +100,7 @@ builder.Services.AddSingleton(sp => new TaskStateService( sp.GetRequiredService(), sp.GetRequiredService(), sp.GetRequiredService(), + sp.GetRequiredService(), sp.GetRequiredService>())); // Agent file management. @@ -269,6 +271,7 @@ if (cfg.ExternalMcpPort > 0) externalBuilder.Services.AddSingleton(app.Services.GetRequiredService()); externalBuilder.Services.AddSingleton(app.Services.GetRequiredService()); externalBuilder.Services.AddSingleton(app.Services.GetRequiredService()); + externalBuilder.Services.AddSingleton(app.Services.GetRequiredService()); externalBuilder.Services.AddScoped(); externalBuilder.Services.AddScoped(); externalBuilder.Services.AddScoped(); diff --git a/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs b/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs index 6c95684a..4f06d8e5 100644 --- a/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs +++ b/src/ClaudeDo.Worker/Queue/OverrideSlotService.cs @@ -10,6 +10,7 @@ public sealed class OverrideSlotService private readonly IDbContextFactory _dbFactory; private readonly TaskRunner _runner; private readonly ILogger _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 dbFactory, TaskRunner runner, - ILogger logger) + ILogger 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); } diff --git a/src/ClaudeDo.Worker/Queue/QueueService.cs b/src/ClaudeDo.Worker/Queue/QueueService.cs index 949de0e8..d3161c2f 100644 --- a/src/ClaudeDo.Worker/Queue/QueueService.cs +++ b/src/ClaudeDo.Worker/Queue/QueueService.cs @@ -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 _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); diff --git a/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs b/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs new file mode 100644 index 00000000..462395c8 --- /dev/null +++ b/src/ClaudeDo.Worker/Queue/RunCancellationRegistry.cs @@ -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 _running = new(StringComparer.Ordinal); + + public void Register(string taskId, CancellationTokenSource cts) => _running[taskId] = cts; + + /// 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. + public void Unregister(string taskId, CancellationTokenSource cts) => + _running.TryRemove(new KeyValuePair(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; + } + } +} diff --git a/src/ClaudeDo.Worker/State/TaskStateService.cs b/src/ClaudeDo.Worker/State/TaskStateService.cs index 5da658eb..22b30dbe 100644 --- a/src/ClaudeDo.Worker/State/TaskStateService.cs +++ b/src/ClaudeDo.Worker/State/TaskStateService.cs @@ -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 _logger; public TaskStateService( @@ -22,12 +23,14 @@ public sealed class TaskStateService : ITaskStateService HubBroadcaster broadcaster, IQueueWaker waker, PlanningChainCoordinator chain, + RunCancellationRegistry runCancels, ILogger 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) diff --git a/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs b/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs index d5ac186d..d9eaf7ef 100644 --- a/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/AddSubtaskToolTests.cs @@ -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.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.Instance); - var queue = new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state); + var runCancels = new RunCancellationRegistry(); + var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); + var queue = new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state, runCancels); var maintenance = new WorktreeMaintenanceService(dbFactory, git, NullLogger.Instance); - var merge = new TaskMergeService(dbFactory, git, broadcaster, TaskStateServiceBuilder.Build(dbFactory).State, NullLogger.Instance); + var merge = new TaskMergeService(dbFactory, git, broadcaster, state, NullLogger.Instance); + var aggregator = new PlanningAggregator(dbFactory, git, NullLogger.Instance); + var planningMerge = new PlanningMergeOrchestrator( + dbFactory, merge, aggregator, broadcaster, git, state, NullLogger.Instance); return new ExternalMcpService( _tasks, _lists, queue, broadcaster, state, - git, dbFactory, maintenance, merge); + git, dbFactory, maintenance, merge, planningMerge); } [Fact] diff --git a/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs b/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs index f38eb8bc..ad265331 100644 --- a/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/BatchMcpToolsTests.cs @@ -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.Instance); - var merge = new TaskMergeService(factory, git, _broadcaster, TaskStateServiceBuilder.Build(factory).State, NullLogger.Instance); + var state = TaskStateServiceBuilder.Build(factory).State; + var merge = new TaskMergeService(factory, git, _broadcaster, state, NullLogger.Instance); + var aggregator = new PlanningAggregator(factory, git, NullLogger.Instance); + var planningMerge = new PlanningMergeOrchestrator( + factory, merge, aggregator, _broadcaster, git, state, NullLogger.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.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder()); - var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance); + var runCancels = new RunCancellationRegistry(); + var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, - new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state); + new QueueWaker(), new QueuePicker(dbFactory), overrideSlot, state, runCancels); } [Fact] diff --git a/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs b/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs index 64bb2465..1d88205e 100644 --- a/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs +++ b/tests/ClaudeDo.Worker.Tests/External/ExternalMcpServiceTests.cs @@ -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.Instance); - var merge = new TaskMergeService(factory, git, _broadcaster, TaskStateServiceBuilder.Build(factory).State, NullLogger.Instance); + var state = TaskStateServiceBuilder.Build(factory).State; + var merge = new TaskMergeService(factory, git, _broadcaster, state, NullLogger.Instance); + var aggregator = new PlanningAggregator(factory, git, NullLogger.Instance); + var planningMerge = new PlanningMergeOrchestrator( + factory, merge, aggregator, _broadcaster, git, state, NullLogger.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.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.Instance); - return new QueueService(dbFactory, runner, cfg, NullLogger.Instance, waker, picker, overrideSlot, state); + var runCancels = new RunCancellationRegistry(); + var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, runCancels); + return new QueueService(dbFactory, runner, cfg, NullLogger.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(() => - 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(() => - 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] diff --git a/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs b/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs index 5529b474..283a193e 100644 --- a/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs +++ b/tests/ClaudeDo.Worker.Tests/Infrastructure/TaskStateServiceBuilder.cs @@ -17,13 +17,15 @@ public static class TaskStateServiceBuilder PlanningChainCoordinator Chain, CapturingHubContext Hub, Func WakeCount, - CountingQueueWaker Waker); + CountingQueueWaker Waker, + RunCancellationRegistry RunCancels); public static Built Build(IDbContextFactory 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.Instance); - return new Built(state, chain, hub, () => waker.Count, waker); + return new Built(state, chain, hub, () => waker.Count, waker, runCancels); } } diff --git a/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs b/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs new file mode 100644 index 00000000..7b682a07 --- /dev/null +++ b/tests/ClaudeDo.Worker.Tests/Queue/RunCancellationRegistryTests.cs @@ -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")); + } +} diff --git a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs index 98ebd6e8..a9cbc2fd 100644 --- a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceSlotGuardTests.cs @@ -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.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.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder()); _waker = new QueueWaker(); var picker = new QueuePicker(dbFactory); - var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance); - var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state); + var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, built.RunCancels); + var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state, built.RunCancels); return (service, fake); } diff --git a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs index 5898b483..e276e6a1 100644 --- a/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs +++ b/tests/ClaudeDo.Worker.Tests/Services/QueueServiceTests.cs @@ -53,13 +53,14 @@ public sealed class QueueServiceTests : IDisposable var dbFactory = _db.CreateFactory(); var wtManager = new WorktreeManager(new GitService(), dbFactory, _cfg, NullLogger.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.Instance, state, new TaskRunTokenRegistry(), new AttachmentStore(), new FakeSessionSkillSeeder()); _waker = new QueueWaker(); var picker = new QueuePicker(dbFactory); - var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance); - var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state); + var overrideSlot = new OverrideSlotService(dbFactory, runner, NullLogger.Instance, built.RunCancels); + var service = new QueueService(dbFactory, runner, _cfg, NullLogger.Instance, _waker, picker, overrideSlot, state, built.RunCancels); return (service, fake); } diff --git a/tests/ClaudeDo.Worker.Tests/WaitingForChildrenLifecycleTests.cs b/tests/ClaudeDo.Worker.Tests/WaitingForChildrenLifecycleTests.cs index 262f9447..ed4d455d 100644 --- a/tests/ClaudeDo.Worker.Tests/WaitingForChildrenLifecycleTests.cs +++ b/tests/ClaudeDo.Worker.Tests/WaitingForChildrenLifecycleTests.cs @@ -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 SeedActivePlanningParentAsync(string id = "par")