fix(heartbeat): keep drain-time wakes queued and retry interrupted disposition handoffs (#14028)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The server's heartbeat scheduler queues, admits, and recovers agent runs. > - A control plane arms an instance task drain before it restarts the process, so no new run starts mid-restart. > - Two recovery paths treated that transient hold as a permanent verdict: a wake that arrived during a drain was written as `skipped` and never replayed, and a corrective disposition run that the restart interrupted counted as an exhausted attempt, so the issue went to `blocked`. > - Both outcomes leave an assigned issue with no run and no path until a person notices. > - This pull request keeps drain-time wakes in the durable queue and gives an interrupted corrective run the same bounded transient retry any interrupted run gets. > - The benefit is that a graceful restart never strands or blocks an issue by itself. ## Linked Issues or Issue Description No existing issue found. I searched open PRs that touch the task drain (#13527 adds opt-in termination of active runs; #13965 handles deferred wakes after a stale cancellation). Neither replays drain-time wakes or changes the corrective-handoff escalation. **What happened?** 1. An operator accepted a plan confirmation while the instance was in a task drain (a control plane armed it before a deploy restart). `enqueueWakeup` wrote the assignee's continuation wake with `status: "skipped"` and `heartbeatSkip.reason: "task_drain"`. The drain is process-local and cleared on the restart seconds later, but nothing replays skipped wakes. The issue stayed in `todo` with no run for 20 hours until a person retried it by hand. The stranded-issue sweeper did not help: a `todo` issue whose latest run succeeded is treated as deliberately handed back. 2. On another issue, the watchdog correctly raised `successful_run_missing_state` and queued the single corrective handoff run. A graceful server shutdown (the same deploy restart) interrupted that run with `server_shutdown_interrupted`. On the next boot, `reconcileStrandedAssignedIssues` saw a corrective run at `handoffAttempt >= maxHandoffAttempts` and escalated the issue to `blocked` with a board-owned `missing_disposition` recovery action. The agent never got to finish one corrective attempt. **Expected behavior** A task drain holds admission, not the request. A wake that arrives during a drain must run once the drain lifts or the process restarts. An interrupted corrective run is not evidence that the agent could not choose a disposition; it should be retried like any interrupted run, and escalate only when that retry budget is spent or a finished attempt still leaves no disposition. **Steps to reproduce** 1. Assign an agent an issue and `POST /api/instance/task-drain` with a Cloud control assertion (or call `startTaskDrain({})` in a test). 2. Trigger any wake for that agent (assignment, comment, or an accepted `request_confirmation`). 3. Observe the `agent_wakeup_requests` row: `status = skipped`, `reason = heartbeat.scheduling_suppressed`, `payload.heartbeatSkip.reason = task_drain`. Stop the drain or restart: no run is ever created for that wake. 4. For the second case: let a `finish_successful_run_handoff` run be interrupted by SIGTERM during a graceful shutdown, then boot. The issue moves to `blocked` on the startup sweep without any retry. **Paperclip version or commit** master at e2f1a66aa7 (2026-09-25). **Deployment mode** Managed cloud instance behind a control plane that arms task drains before deploy restarts; the code paths are the same for any operator who uses the drain endpoint. ## What Changed - `enqueueWakeup` no longer writes a wake as `skipped` when the only suppression is `task_drain`. The wake and its queued run land in the durable queue; `startNextQueuedRunForAgent` and `executeRun` already refuse to admit work while the drain is active, and the boot-time `resumeQueuedRuns` pass picks it up after the restart. `worktree_instance` and `database_restore_in_progress` suppression still write `skipped`: those holds are not transient restarts. - `reconcileStrandedAssignedIssues`: when the latest run is a corrective successful-run handoff at its attempt cap **and** that run is `interrupted`, the sweeper schedules the bounded transient retry (via `enqueueStrandedIssueRecovery` → `scheduleRecoveryRetry`) instead of escalating. The retry keeps the handoff context, so it is still the corrective run. If the retry budget is spent, or a finished attempt still leaves no disposition, escalation proceeds exactly as before. Failed corrective runs (adapter or provider failures) are unchanged: they still escalate immediately. - New sweep counter `successfulRunHandoffRetried`, included in the startup and periodic recovery log lines. - Tests: `heartbeat-task-drain-admission-release.test.ts` gains a case that a wake during a drain is queued, held while the drain is active (quiescence still reports true), and runs to completion once the drain lifts. `heartbeat-process-recovery.test.ts` gains two cases: an interrupted corrective run is retried with its handoff context and the issue stays `in_progress`; an interrupted corrective run with a spent transient budget escalates to `blocked` with the usual recovery action evidence. ## Verification ``` pnpm -r --filter './packages/**' build cd server npx vitest run src/__tests__/heartbeat-process-recovery.test.ts # 296 passed npx vitest run src/__tests__/heartbeat-task-drain-admission-release.test.ts \ src/__tests__/heartbeat-worktree-suppression.test.ts \ src/__tests__/heartbeat-scheduling-suppression.test.ts \ src/__tests__/heartbeat-task-drain.test.ts \ src/__tests__/instance-settings-routes.test.ts \ src/services/recovery/successful-run-handoff.test.ts \ src/__tests__/issue-recovery-actions.test.ts \ src/__tests__/heartbeat-comment-wake-batching.test.ts \ src/__tests__/attention-service.test.ts # 226 passed npx tsc --noEmit -p tsconfig.json # clean ``` Live reproduction of the first defect: after this change was written, a board wake issued against a draining managed instance (running current master) was again recorded as `skipped` with `heartbeatSkip.reason = task_drain`, which is the exact row the new test asserts no longer appears. ## Risks - A wake that lands during a drain now waits in the queue instead of being dropped. `getTaskDrainStatus().pendingWakes` counts only in-flight enqueue promises, so quiescence is unchanged and a control plane waiting for the drain is not held longer. If a drain is stopped without a restart, the queued run starts on the next scheduler tick. - The interrupted-handoff retry reuses the existing transient retry budget (two attempts) and the existing `hasActiveExecutionPath` skip, so a sweep cannot double-schedule. Native-runtime corrective runs still return `null` from the recovery enqueue and escalate as before. - Self-hosted instances that never call the drain endpoint see no change on the first path; the second path only changes behavior for corrective runs interrupted by a graceful shutdown. ## Model Used Claude (Anthropic) — `claude-fable-5-1`, extended thinking, tool use (Claude Code CLI). Human-directed: diagnosis of the two production incidents, the fix design, and review were steered by the maintainer. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-29 12:59:13 +01:00 · 2026-09-25 11:28:54 -07:00
1 parent bd20309323
commit 5b09d66183
@@ -5841,6 +5841,171 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
    ).toBe(true);
  });

+  it("retries an interrupted corrective successful-run handoff instead of escalating it", async () => {
+    // A graceful server shutdown (a deploy restart) interrupted the single
+    // corrective run before the agent could choose a disposition. That is
+    // not a finished attempt, so the sweeper must give it the bounded
+    // transient retry any interrupted run gets, not a board escalation.
+    const { agentId, runId, issueId } = await seedStrandedIssueFixture({
+      status: "in_progress",
+      runStatus: "failed",
+      runErrorCode: "server_shutdown_interrupted",
+      runError: "Interrupted by graceful server shutdown (SIGTERM)",
+    });
+    const sourceRunId = randomUUID();
+    await db
+      .update(heartbeatRuns)
+      .set({
+        status: "interrupted",
+        contextSnapshot: {
+          issueId,
+          taskId: issueId,
+          wakeReason: "finish_successful_run_handoff",
+          sourceRunId,
+          resumeFromRunId: sourceRunId,
+          handoffRequired: true,
+          handoffReason: "successful_run_missing_state",
+          missingDisposition: "clear_next_step",
+          handoffAttempt: 1,
+          maxHandoffAttempts: 1,
+        },
+      })
+      .where(eq(heartbeatRuns.id, runId));
+
+    const heartbeat = heartbeatService(db);
+    const result = await heartbeat.reconcileStrandedAssignedIssues();
+    expect(result.successfulRunHandoffRetried).toBe(1);
+    expect(result.successfulRunHandoffEscalated).toBe(0);
+    expect(result.escalated).toBe(0);
+    expect(result.issueIds).toEqual([issueId]);
+
+    const retryRun = await db
+      .select()
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.retryOfRunId, runId))
+      .then((rows) => rows[0] ?? null);
+    expect(retryRun).toMatchObject({
+      agentId,
+      status: "scheduled_retry",
+      scheduledRetryAttempt: 1,
+      scheduledRetryReason: "transient_failure",
+    });
+    // The retry keeps the handoff contract: it is still the corrective run,
+    // so a finished attempt that leaves no disposition still escalates.
+    expect(retryRun?.contextSnapshot).toMatchObject({
+      issueId,
+      retryOfRunId: runId,
+      handoffRequired: true,
+      handoffReason: "successful_run_missing_state",
+      missingDisposition: "clear_next_step",
+      handoffAttempt: 1,
+      maxHandoffAttempts: 1,
+    });
+
+    const sourceIssue = await db
+      .select()
+      .from(issues)
+      .where(eq(issues.id, issueId))
+      .then((rows) => rows[0] ?? null);
+    expect(sourceIssue?.status).toBe("in_progress");
+    expect(sourceIssue?.assigneeAgentId).toBe(agentId);
+    const recoveryActions = await db
+      .select()
+      .from(issueRecoveryActions)
+      .where(eq(issueRecoveryActions.sourceIssueId, issueId));
+    expect(recoveryActions).toHaveLength(0);
+    const comments = await db
+      .select()
+      .from(issueComments)
+      .where(eq(issueComments.issueId, issueId));
+    expect(comments).toHaveLength(0);
+
+    // A second sweep sees the scheduled retry as the live path and leaves
+    // the issue alone rather than escalating or double-scheduling.
+    const again = await heartbeat.reconcileStrandedAssignedIssues();
+    expect(again.successfulRunHandoffRetried).toBe(0);
+    expect(again.successfulRunHandoffEscalated).toBe(0);
+    const retryRuns = await db
+      .select({ id: heartbeatRuns.id })
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.retryOfRunId, runId));
+    expect(retryRuns).toHaveLength(1);
+  });
+
+  it("escalates an interrupted corrective successful-run handoff once its transient retry budget is spent", async () => {
+    const { companyId, agentId, runId, issueId } =
+      await seedStrandedIssueFixture({
+        status: "in_progress",
+        runStatus: "failed",
+        runErrorCode: "server_shutdown_interrupted",
+        runError: "Interrupted by graceful server shutdown (SIGTERM)",
+        // The real interrupted run carries the conversation-continuation
+        // marker the shutdown path writes; it also keeps a spent retry chain
+        // out of legacy reconciliation so the sweeper reaches the handoff
+        // branch with an exhausted budget.
+        resultJson: {
+          stopReason: "interrupted",
+          conversationContinuation: "continue_conversation_v1",
+        },
+      });
+    const sourceRunId = randomUUID();
+    await db
+      .update(heartbeatRuns)
+      .set({
+        status: "interrupted",
+        scheduledRetryAttempt: 2,
+        scheduledRetryReason: "transient_failure",
+        contextSnapshot: {
+          issueId,
+          taskId: issueId,
+          wakeReason: "transient_failure_retry",
+          sourceRunId,
+          resumeFromRunId: sourceRunId,
+          handoffRequired: true,
+          handoffReason: "successful_run_missing_state",
+          missingDisposition: "clear_next_step",
+          handoffAttempt: 1,
+          maxHandoffAttempts: 1,
+        },
+      })
+      .where(eq(heartbeatRuns.id, runId));
+
+    const heartbeat = heartbeatService(db);
+    const result = await heartbeat.reconcileStrandedAssignedIssues();
+    expect(result.successfulRunHandoffRetried).toBe(0);
+    expect(result.successfulRunHandoffEscalated).toBe(1);
+    expect(result.issueIds).toEqual([issueId]);
+    const retryRuns = await db
+      .select({ id: heartbeatRuns.id })
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.retryOfRunId, runId));
+    expect(retryRuns).toHaveLength(0);
+
+    const recoveryAction = await expectSourceScopedStrandedRecoveryAction({
+      companyId,
+      agentId,
+      issueId,
+      runId,
+      previousStatus: "in_progress",
+      retryReason: null,
+      cause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
+      kind: "missing_disposition",
+    });
+    expect(recoveryAction.evidence).toMatchObject({
+      sourceRunId,
+      missingDisposition: "clear_next_step",
+      latestRunStatus: "interrupted",
+      latestRunErrorCode: "server_shutdown_interrupted",
+      recoveryCause: SUCCESSFUL_RUN_MISSING_STATE_REASON,
+    });
+    const sourceIssue = await db
+      .select()
+      .from(issues)
+      .where(eq(issues.id, issueId))
+      .then((rows) => rows[0] ?? null);
+    expect(sourceIssue?.status).toBe("blocked");
+  });
+
  it("escalates an exhausted successful handoff run that still leaves no disposition", async () => {
    const { companyId, agentId, runId, issueId } =
      await seedStrandedIssueFixture({
@@ -75,12 +75,10 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => {
    await tempDb?.cleanup();
  });

-  async function seedQueuedRun() {
+  async function seedAgentAndIssue() {
    const companyId = randomUUID();
    const agentId = randomUUID();
    const issueId = randomUUID();
-    const runId = randomUUID();
-    const wakeupRequestId = randomUUID();

    await db.insert(companies).values({
      id: companyId,
@@ -120,6 +118,14 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => {
      responsibleUserId: "responsible-user",
    });

+    return { companyId, agentId, issueId };
+  }
+
+  async function seedQueuedRun() {
+    const { companyId, agentId, issueId } = await seedAgentAndIssue();
+    const runId = randomUUID();
+    const wakeupRequestId = randomUUID();
+
    await db.insert(agentWakeupRequests).values({
      id: wakeupRequestId,
      companyId,
@@ -219,6 +225,78 @@ describeEmbeddedPostgres("heartbeat task-drain admission release", () => {
    expect(finished?.status).toBe("succeeded");
  }, 20_000);

+  it("keeps a wake that arrives during a task drain queued and runs it once the drain lifts", async () => {
+    const { agentId, issueId } = await seedAgentAndIssue();
+    const heartbeat = heartbeatService(db);
+
+    // The drain holds ADMISSION, not the request. The wake lands in the
+    // durable queue, so the restart that clears the process-local drain
+    // resumes it instead of losing it. (Before this, the wake was written
+    // as `skipped` and the issue sat in `todo` with no run and no path.)
+    startTaskDrain({});
+    await heartbeat.wakeup(agentId, {
+      source: "assignment",
+      triggerDetail: "system",
+      reason: "issue_assigned",
+      payload: { issueId },
+      contextSnapshot: { issueId, wakeReason: "issue_assigned" },
+      requestedByActorType: "system",
+      requestedByActorId: "issue_assignment",
+    });
+    await heartbeat.drainActiveRunExecutions();
+
+    const wakeup = await db
+      .select({
+        status: agentWakeupRequests.status,
+        reason: agentWakeupRequests.reason,
+        payload: agentWakeupRequests.payload,
+      })
+      .from(agentWakeupRequests)
+      .then((rows) => rows[0] ?? null);
+    expect(wakeup).toMatchObject({ status: "queued", reason: "issue_assigned" });
+    expect(wakeup?.payload).not.toHaveProperty("heartbeatSkip");
+    const queuedRuns = await db
+      .select({ id: heartbeatRuns.id, status: heartbeatRuns.status })
+      .from(heartbeatRuns);
+    expect(queuedRuns).toHaveLength(1);
+    expect(queuedRuns[0]).toMatchObject({ status: "queued" });
+    const runId = queuedRuns[0]!.id;
+
+    // Admission stays held for as long as the drain is active, and the
+    // queued row does not count against quiescence: the restart may proceed.
+    await heartbeat.resumeQueuedRuns();
+    await heartbeat.drainActiveRunExecutions();
+    const held = await db
+      .select({ status: heartbeatRuns.status })
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, runId))
+      .then((rows) => rows[0] ?? null);
+    expect(held?.status).toBe("queued");
+    expect(getTaskDrainStatus()).toMatchObject({
+      draining: true,
+      activeRuns: 0,
+      pendingWakes: 0,
+      quiescent: true,
+    });
+
+    // The drain lifts (a restart clears it the same way): the queued wake
+    // runs to completion through the normal admission path.
+    stopTaskDrain();
+    await heartbeat.resumeQueuedRuns();
+    await heartbeat.drainActiveRunExecutions();
+    const finished = await db
+      .select({ status: heartbeatRuns.status })
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.id, runId))
+      .then((rows) => rows[0] ?? null);
+    expect(finished?.status).toBe("succeeded");
+    const completedWakeup = await db
+      .select({ status: agentWakeupRequests.status })
+      .from(agentWakeupRequests)
+      .then((rows) => rows[0] ?? null);
+    expect(completedWakeup?.status).toBe("completed");
+  }, 20_000);
+
  // Inject one rollback only after the test starts task drain. Admission also
  // uses transactions, so transaction call numbers do not identify release.
  function withFailingTransactionalUpdate(realDb: typeof db, failingTable: unknown, armed: () => boolean) {
@@ -1529,6 +1529,7 @@ async function startServerWithDatabaseTeardown(
          reconciled.dispatchRequeued > 0 ||
          reconciled.continuationRequeued > 0 ||
          reconciled.successfulRunHandoffEscalated > 0 ||
+          reconciled.successfulRunHandoffRetried > 0 ||
          reconciled.escalated > 0
        ) {
          logger.warn(
@@ -1769,6 +1770,7 @@ async function startServerWithDatabaseTeardown(
                reconciled.dispatchRequeued > 0 ||
                reconciled.continuationRequeued > 0 ||
                reconciled.successfulRunHandoffEscalated > 0 ||
+                reconciled.successfulRunHandoffRetried > 0 ||
                reconciled.escalated > 0
              ) {
                logger.warn(
@@ -26339,7 +26339,19 @@ export function heartbeatService(
    };

    const schedulingSuppression = await getSchedulingSuppression();
-    if (schedulingSuppression.suppressed) {
+    // A task drain holds ADMISSION, not the request. The drain is a
+    // process-local pre-restart hold, so a wake that arrives while it is
+    // active still names real work that must run once the process comes
+    // back: leave it in the durable queue and let the dispatch-side checks
+    // (startNextQueuedRunForAgent / executeRun) keep it from starting until
+    // the drain lifts or the restart clears it. Writing it as `skipped`
+    // here dropped the wake permanently — an accepted plan whose
+    // continuation wake landed mid-drain left its issue in `todo` with no
+    // run and no path until a person noticed.
+    if (
+      schedulingSuppression.suppressed &&
+      schedulingSuppression.reason !== "task_drain"
+    ) {
      await writeSkippedHeartbeatRequest("heartbeat.scheduling_suppressed", {
        reason: schedulingSuppression.reason,
      });
@@ -4174,6 +4174,7 @@ export function recoveryService(
      successfulContinuationObserved: 0,
      orphanBlockersAssigned: 0,
      successfulRunHandoffEscalated: 0,
+      successfulRunHandoffRetried: 0,
      reviewParticipantRequeued: 0,
      escalated: 0,
      waitingOnReviewResolved: 0,
@@ -4995,6 +4996,39 @@ export function recoveryService(
          continue;
        }

+        // An interrupted corrective run is not evidence that the agent could
+        // not choose a disposition: a graceful server shutdown (a deploy
+        // restart, a lost process) ended the attempt before the agent
+        // finished. The attempt cap counts attempts the agent got to finish,
+        // so give the interrupted run the same bounded transient retry any
+        // interrupted run gets — the retry keeps the handoff context, so it
+        // is still the corrective run — and escalate only once that retry
+        // budget is spent or a finished attempt still leaves no disposition.
+        // Native-runtime corrective runs have no process-loss retry lane
+        // (a graceful shutdown suspends their controller for reattach
+        // instead of interrupting them; other native interruptions are
+        // reconciled by their own finalizer), so the recovery enqueue
+        // returns null for them and they escalate exactly as before.
+        if (latestRun?.status === "interrupted") {
+          if (await isInvocationBudgetBlocked(issue, agentId)) {
+            result.skipped += 1;
+            continue;
+          }
+          const retried = await enqueueStrandedIssueRecovery({
+            issueId: issue.id,
+            agentId,
+            reason: "issue_continuation_needed",
+            retryReason: "issue_continuation_needed",
+            source: "issue.successful_run_handoff_interrupted_retry",
+            retryOfRunId: latestRun.id,
+          });
+          if (retried) {
+            result.successfulRunHandoffRetried += 1;
+            result.issueIds.push(issue.id);
+            continue;
+          }
+        }
+
        const updated = await escalateStrandedAssignedIssue({
          issue,
          previousStatus: "in_progress",