fix(coordinator): stop terminal orphan re-arms

.changeset/terminal-coordinator-orphan-recovery.md

@@ -0,0 +1,5 @@
+---
+"agentweaver": patch
+---
+
+Stop terminal coordinator runs from being repeatedly recovered after a service restart.

apps/Agentweaver.Api/Coordinator/CoordinatorDispatchService.cs

@@ -442,7 +442,8 @@ await PersistStoppedCoordinatorWorkPlanStatusAsync(
                 // subtask's bounded recovery budget — reset it to Pending so the frontier redispatches
                 // a fresh child (on a fresh pod) next iteration. Only when the budget is exhausted
                 // (RecoveryAttempts >= MaxRecoveryAttempts) does the stall become a genuine terminal.
-                if (await TryRedispatchStalledSubtaskAsync(
+                if (!coordinatorStopped
+                    && await TryRedispatchStalledSubtaskAsync(
                         context, workPlanId.Value, result.SubtaskId, result.ChildRunId, statusById, seq, ct)
                         .ConfigureAwait(false))
                     continue;

apps/Agentweaver.Api/Coordinator/CoordinatorReconciler.cs

@@ -156,6 +156,13 @@ or WorkPlanStatus.AssemblyBlocked
 
             try
             {
+                // A prior process can stop after the coordinator run becomes terminal but before its
+                // work-plan status is settled. Work plans and runs use separate stores, so this check
+                // cannot be part of the EF candidate query. Active children must first be recovered by
+                // the dispatch loop; it re-observes and drains them without launching new children.
+                if (await TrySetTerminalCoordinatorWorkPlanStatusAsync(plan, ct).ConfigureAwait(false))
+                    continue;
+
                 switch (plan.Status)
                 {
                     case WorkPlanStatus.Dispatching:
@@ -484,6 +491,57 @@ private async Task<bool> TryReArmAssemblyAsync(PlanCandidate plan, CancellationT
             ProjectId: run.ProjectId);
     }
 
+    /// <summary>
+    /// Settles a candidate whose coordinator run has reached a terminal result and whose children have
+    /// all drained. The work-plan status is the durable orphan-scan cursor, so persisting this transition
+    /// makes future scans skip the run even after this pod is replaced. Returns <c>true</c> when the
+    /// candidate must not be re-armed.
+    /// </summary>
+    private async Task<bool> TrySetTerminalCoordinatorWorkPlanStatusAsync(
+        PlanCandidate plan,
+        CancellationToken ct)
+    {
+        if (string.IsNullOrWhiteSpace(plan.CoordinatorRunId)
+            || !RunId.TryParse(plan.CoordinatorRunId, out var runId))
+            return false;
+
+        var run = await _runStore.GetAsync(runId, ct).ConfigureAwait(false);
+        var terminalStatus = GetTerminalCoordinatorWorkPlanStatus(run?.Status);
+        if (terminalStatus is null)
+            return false;
+
+        using var scope = _scopeFactory.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<MemoryDbContext>();
+        var hasActiveSubtasks = await db.Subtasks
+            .AsNoTracking()
+            .AnyAsync(s => s.WorkPlanId == plan.WorkPlanId
+                        && (s.Status == SubtaskStatus.Dispatched || s.Status == SubtaskStatus.Running), ct)
+            .ConfigureAwait(false);
+        if (hasActiveSubtasks)
+            return false;
+
+        var now = DateTimeOffset.UtcNow;
+        await db.WorkPlans
+            .Where(w => w.Id == plan.WorkPlanId && w.Status == plan.Status)
+            .ExecuteUpdateAsync(s => s
+                .SetProperty(w => w.Status, terminalStatus)
+                .SetProperty(w => w.UpdatedAt, now), ct)
+            .ConfigureAwait(false);
+
+        _logger.LogInformation(
+            "Coordinator reconciler: settled stopped coordinator run {RunId} as work-plan status {Status}",
+            plan.CoordinatorRunId, terminalStatus);
+        return true;
+    }
+
+    private static string? GetTerminalCoordinatorWorkPlanStatus(RunStatus? status) => status switch
+    {
+        RunStatus.Completed or RunStatus.Merged => WorkPlanStatus.Complete,
+        RunStatus.Declined => WorkPlanStatus.AssemblyDeclined,
+        RunStatus.Failed or RunStatus.MergeFailed => WorkPlanStatus.AssemblyFailed,
+        _ => null,
+    };
+
     private async Task ResetAssemblyPlanAsync(int workPlanId, CancellationToken ct)
     {
         using var scope = _scopeFactory.CreateScope();

tests/Agentweaver.Tests/Coordinator/CoordinatorReconcilerTests.cs

@@ -221,6 +221,98 @@ public async Task Sweep_ActiveDispatch_DoesNotDoubleArm()
         dispatch.StartDispatchCalls.Should().BeEmpty("an already-active coordinator is never re-armed");
     }
 
+    [Fact]
+    public async Task Sweep_TerminalCoordinator_PersistsTerminalPlanState_AndSkipsRepeatedScans()
+    {
+        var coord = RunId.New().ToString();
+        await SeedCoordinatorRunAsync(coord, RunStatus.Completed);
+        var (planId, _) = await SeedPlanAsync(coord, new[] { (SubtaskStatus.Completed, (string?)null) });
+        var dispatch = new RecordingDispatch();
+        var reconciler = BuildReconciler(dispatch);
+
+        (await reconciler.SweepAsync(default)).Should().Be(0);
+        (await reconciler.SweepAsync(default)).Should().Be(0);
+
+        (await GetPlanStatusAsync(planId)).Should().Be(WorkPlanStatus.Complete,
+            "the terminal result is persisted instead of re-armed on every heartbeat");
+        dispatch.StartDispatchCalls.Should().BeEmpty();
+    }
+
+    [Fact]
+    public async Task Sweep_TerminalCoordinator_DurablePlanStateSurvivesPodRestart()
+    {
+        var coord = RunId.New().ToString();
+        await SeedCoordinatorRunAsync(coord, RunStatus.Failed);
+        var (planId, _) = await SeedPlanAsync(coord, new[] { (SubtaskStatus.Completed, (string?)null) });
+        var firstPodDispatch = new RecordingDispatch();
+
+        (await BuildReconciler(firstPodDispatch).SweepAsync(default)).Should().Be(0);
+
+        var restartedPodDispatch = new RecordingDispatch();
+        (await BuildReconciler(restartedPodDispatch).SweepAsync(default)).Should().Be(0);
+
+        (await GetPlanStatusAsync(planId)).Should().Be(WorkPlanStatus.AssemblyFailed);
+        firstPodDispatch.StartDispatchCalls.Should().BeEmpty();
+        restartedPodDispatch.StartDispatchCalls.Should().BeEmpty(
+            "a new pod reads the persisted terminal work-plan state, not a per-pod memory set");
+    }
+
+    [Fact]
+    public async Task Sweep_TerminalCoordinatorWithActiveChild_RearmsRecoveryUntilChildDrainsThenSettles()
+    {
+        var coord = RunId.New().ToString();
+        var child = await SeedChildRunAsync(RunStatus.AssembleReady, parentRunId: coord);
+        var (planId, subtaskIds) = await SeedPlanAsync(
+            coord, new[] { (SubtaskStatus.Running, (string?)child) });
+        _streamStore.Create(coord, "owner");
+
+        // This represents the first sweep after a pod restart: the coordinator already stopped, but
+        // its persisted child is still marked running. The real dispatch recovery loop must observe
+        // that child before the plan can become terminal.
+        await SeedCoordinatorRunAsync(coord, RunStatus.Failed);
+        var restartedPodDispatch = new RecoveringDispatch(BuildDispatch());
+        var restartedPodReconciler = BuildReconciler(restartedPodDispatch);
+
+        (await restartedPodReconciler.SweepAsync(default)).Should().Be(1);
+        await restartedPodDispatch.WaitForRecoveryAsync();
+
+        (await GetSubtaskAsync(subtaskIds[0])).Status.Should().Be(SubtaskStatus.AssembleReady,
+            "the restarted dispatch loop re-observes and drains the existing child");
+        (await GetPlanStatusAsync(planId)).Should().Be(WorkPlanStatus.AssemblyFailed,
+            "the terminal coordinator settles only after its active child drains");
+        restartedPodDispatch.StartDispatchCalls.Should().ContainSingle();
+        (await _runStore.GetRunsByParentAsync(coord)).Should().ContainSingle()
+            .Which.Id.Should().Be(RunId.Parse(child),
+                "terminal-coordinator recovery drains its existing child instead of creating a new one");
+
+        var nextPodDispatch = new RecordingDispatch();
+        (await BuildReconciler(nextPodDispatch).SweepAsync(default)).Should().Be(0);
+        nextPodDispatch.StartDispatchCalls.Should().BeEmpty(
+            "the durable terminal plan state prevents re-arming on later restarts");
+    }
+
+    [Fact]
+    public async Task Sweep_CoordinatorBecomesTerminalAfterChildDrains_SettlesPreviouslyOrphanedDispatch()
+    {
+        var coord = RunId.New().ToString();
+        await SeedCoordinatorRunAsync(coord);
+        var (planId, subtaskIds) = await SeedPlanAsync(
+            coord, new[] { (SubtaskStatus.Running, (string?)RunId.New().ToString()) });
+        var dispatch = new RecordingDispatch();
+        var reconciler = BuildReconciler(dispatch);
+
+        (await reconciler.SweepAsync(default)).Should().Be(1,
+            "a genuinely interrupted non-terminal coordinator still recovers");
+        await SetSubtaskStatusAsync(subtaskIds[0], SubtaskStatus.Completed);
+        await _runStore.UpdateStatusAsync(RunId.Parse(coord), RunStatus.Failed, DateTimeOffset.UtcNow);
+
+        (await reconciler.SweepAsync(default)).Should().Be(0);
+
+        (await GetPlanStatusAsync(planId)).Should().Be(WorkPlanStatus.AssemblyFailed);
+        dispatch.StartDispatchCalls.Should().ContainSingle(
+            "the terminal transition stops future recovery without preventing the earlier non-terminal recovery");
+    }
+
     // -----------------------------------------------------------------------
     // in_review handling: legitimate-vs-orphaned + auto-abandon escape hatch.
     // -----------------------------------------------------------------------
@@ -572,7 +664,7 @@ private CoordinatorDispatchService BuildDispatch(double stallTimeoutMinutes = 15
             runOptions: null, autopilot: null, configuration: config);
     }
 
-    private CoordinatorReconciler BuildReconciler(RecordingDispatch dispatch)
+    private CoordinatorReconciler BuildReconciler(ICoordinatorDispatch dispatch)
     {
         return new CoordinatorReconciler(
             _scopeFactory, _runStore, _streamStore, dispatch, NullLogger<CoordinatorReconciler>.Instance);
@@ -581,7 +673,10 @@ private CoordinatorReconciler BuildReconciler(RecordingDispatch dispatch)
     private static CoordinatorDispatchContext Context(string coord) =>
         new(coord, "repo", "main", "owner", null);
 
-    private async Task<string> SeedChildRunAsync(RunStatus status, DateTimeOffset? startedAt = null)
+    private async Task<string> SeedChildRunAsync(
+        RunStatus status,
+        DateTimeOffset? startedAt = null,
+        string? parentRunId = null)
     {
         var id = RunId.New();
         var run = new Run
@@ -595,7 +690,7 @@ private async Task<string> SeedChildRunAsync(RunStatus status, DateTimeOffset? s
             Status = RunStatus.InProgress,
             StartedAt = startedAt ?? DateTimeOffset.UtcNow,
             AgentName = "morpheus",
-            ParentRunId = RunId.New().ToString(),
+            ParentRunId = parentRunId ?? RunId.New().ToString(),
             SubtaskId = "0",
         };
         await _runStore.InsertAsync(run);
@@ -789,6 +884,23 @@ private sealed class RecordingDispatch : ICoordinatorDispatch
         public bool IsDispatchActive(string coordinatorRunId) => Active;
     }
 
+    private sealed class RecoveringDispatch(CoordinatorDispatchService dispatch) : ICoordinatorDispatch
+    {
+        public List<CoordinatorDispatchContext> StartDispatchCalls { get; } = [];
+        private Task? RecoveryTask { get; set; }
+
+        public void StartDispatch(CoordinatorDispatchContext context)
+        {
+            StartDispatchCalls.Add(context);
+            RecoveryTask = dispatch.RunDispatchLoopAsync(context, CancellationToken.None);
+        }
+
+        public bool IsDispatchActive(string coordinatorRunId) => false;
+
+        public Task WaitForRecoveryAsync() =>
+            RecoveryTask ?? throw new InvalidOperationException("Recovery was not started.");
+    }
+
     private sealed class TestHostApplicationLifetime : IHostApplicationLifetime
     {
         public CancellationToken ApplicationStarted => CancellationToken.None;