From 5b09d661831f0b2d265c696496a9528f9629bb7d Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 25 Sep 2026 11:28:54 -0700 Subject: [PATCH] fix(heartbeat): keep drain-time wakes queued and retry interrupted disposition handoffs (#14028) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 --- .../heartbeat-process-recovery.test.ts | 165 ++++++++++++++++++ ...tbeat-task-drain-admission-release.test.ts | 84 ++++++++- server/src/index.ts | 2 + server/src/services/heartbeat.ts | 14 +- server/src/services/recovery/service.ts | 34 ++++ 5 files changed, 295 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 212f19d00bd..ccf848eb184 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -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({ diff --git a/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts b/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts index 32ddc0de94b..8de38150f94 100644 --- a/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts +++ b/server/src/__tests__/heartbeat-task-drain-admission-release.test.ts @@ -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) { diff --git a/server/src/index.ts b/server/src/index.ts index ba8cde96eb9..dc47462a483 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -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( diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index abf7e8a3425..48a252089aa 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -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, }); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 19316440a2a..98be0347ae0 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -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",