From c4ba3974a7c34c2c7b1db7b80a459cc941b0ef97 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 8 Nov 2026 14:59:55 -0600 Subject: [PATCH] fix(heartbeat): close run admission before a graceful shutdown closes its listener (t09) (#20102) 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 > - A graceful restart (K2) drains the control plane: it closes HTTP admission, suspends running runs, and the next boot resumes them > - Run admission stayed open while the HTTP listener closed. Heartbeat's shutdown flag was set only later, inside hot-restart preparation > - A run that finished in that window released its issue and promoted the queued comment wake. That started a new run that had no callback listener > - A remote warm continuation then failed with `local_runtime_api_listener_unavailable`, and the task needed a Retry (restart chaos t09) > - This pull request closes run admission before the listener closes, so the promoted wake stays queued for the next boot > - The benefit is that a comment sent during a deploy continues on its own after the restart ## Linked Issues or Issue Description Refs #19596 **What happened?** In restart-chaos wave w-20261108T192459-s9501 (Daytona warm, Codex C1, K2), a K2 SIGTERM started a graceful drain at 19:42:52 UTC, and the HTTP listener closed first. At 19:43:03 the running turn finished and released its issue. That promoted the queued comment wake, and `startNextQueuedRunForAgent` started a new run about 11 s into the drain. Backend setup resolved the local runtime API URL, found the listener closed, and threw `local_runtime_api_listener_unavailable`. The remote target is excluded from the closed-listener handoff, so the run failed, and the task waited for a manual Retry. **Expected behavior** No queued run starts once a graceful shutdown has begun. A wake promoted during the drain stays queued, and the next boot starts it with no Retry. **Steps to reproduce** 1. Run a remote warm task and post a comment while its turn runs, so the comment wake queues behind it. 2. Send SIGTERM to the control plane just before the turn finishes. 3. Before this change, a new run starts during the drain and fails with `local_runtime_api_listener_unavailable`. **Paperclip version or commit** master at d007afbf86 (the wave ran on c5a172a183). ## What Changed - `heartbeat/run-admission.ts` (new): a process-wide run admission gate (`closeProcessRunAdmission`, `isProcessRunAdmissionClosed`). HTTP routes build their own heartbeat instances, so the gate cannot live in one instance. - `heartbeat/queue.ts`: a new optional dependency, `isRunAdmissionClosed`, which defaults to the process-wide gate. `startNextQueuedRunForAgent`, `resumeQueuedRuns` and `resumeQueuedRunsAfterDrain` start nothing (and read nothing) once it returns true. Wakes are still recorded and promoted; only the start is held. - `heartbeat.ts`: also honors its own `shutdownInProgress` flag. - `shutdown.ts`: `closeHttpListenerForShutdown` takes `closeRunAdmission` and calls it first, before it checks or closes the listener. `index.ts` passes `closeProcessRunAdmission`. ## Verification - Red/green, new tests: - `queue.test.ts`: "starts no run once shutdown closed run admission" (for `startNextQueuedRunForAgent` and `resumeQueuedRuns`), and "honors the process-wide run admission gate in every heartbeat instance". - `shutdown.test.ts`: "closes run admission before it stops accepting requests". - `pnpm exec vitest run src/services/heartbeat/queue.test.ts src/services/heartbeat/recovery.test.ts src/__tests__/heartbeat-comment-wake-batching.test.ts src/services/native-runtime/native-startup-retry.integration.test.ts src/shutdown.test.ts` (server): 5 files, 277 tests passed. - `PAPERCLIP_TYPECHECK_LOW_MEMORY=1 pnpm typecheck` (server): passed. ## Risks - Low risk. Only queued-run starts are held during a shutdown that has already begun; wakes, drains, suspension and finalizers are unchanged. The next boot's queued-run dispatch starts held runs. - Known remaining window: a run admitted just before shutdown can still reach backend setup after the listener closed. A local runner hands off to restart recovery there. A remote warm continuation still fails, because the remote target is excluded from that handoff (#19596) and the bootstrap retry does not cover warm continuations. This change removes the t09 sequence, where the run started after shutdown had begun; it does not remove that narrower window. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use (shell, file edits, test execution, a read-only research subagent). ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- server/src/index.ts | 3 ++- server/src/services/heartbeat.ts | 2 ++ server/src/services/heartbeat/queue.test.ts | 25 +++++++++++++++++++ server/src/services/heartbeat/queue.ts | 8 ++++-- .../src/services/heartbeat/run-admission.ts | 13 ++++++++++ server/src/shutdown.test.ts | 13 ++++++++++ server/src/shutdown.ts | 2 ++ 7 files changed, 63 insertions(+), 3 deletions(-) create mode 100644 server/src/services/heartbeat/run-admission.ts diff --git a/server/src/index.ts b/server/src/index.ts index 6645c7ef067..7bfaf346c27 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,4 +1,5 @@ import type { IdleDeadlineOwners } from "./services/idle-deadline-types.js"; +import { closeProcessRunAdmission } from "./services/heartbeat/run-admission.js"; import { registerOwnershipReconciliationWork } from "./services/native-runtime/native-finalization-reconciler.js"; import { bindIdleDeadlineOwner } from "./services/idle-deadline-owner.js"; import { readIdleRetrySchedule } from "./services/heartbeat/idle-retry-schedule.js"; @@ -2044,7 +2045,7 @@ async function startServerWithDatabaseTeardown( liveEventsWebSocketServer.close(); // Finish admitted HTTP requests while their services, workers and database // are still available. Railway must switch routing before signaling A. - await closeHttpListenerForShutdown({ server, signal, timeoutMs: 30_000, log: logger, + await closeHttpListenerForShutdown({ server, signal, timeoutMs: 30_000, log: logger, closeRunAdmission: closeProcessRunAdmission, drainRequests: abort => drainHttpRequestWork(app, abort) }); const preDrain = createPreDrainShutdownBudget({ signal, log: logger }); await preDrain.run("systemd_notify", () => diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2997874085a..6fbd9ad75eb 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,5 +1,6 @@ import { registerHeldPreparationResumer } from "./interrupted-preparation-hold-settlement.js"; import { resolveProviderApiPlacement } from "./native-runtime/provider-api-placement.js"; +import { isProcessRunAdmissionClosed } from "./heartbeat/run-admission.js"; import type { StopInvocationOptions } from "./heartbeat/run-control.js"; import { isContextCompactionRun } from "./context-compaction.js"; import { isUnstartedTaskQuestionAnswer } from "./unstarted-task-question-answer.js"; @@ -1321,6 +1322,7 @@ export function heartbeatService( isHeartbeatRunTerminalStatus, liveRunExecutions, getSchedulingSuppression, + isRunAdmissionClosed: () => shutdownInProgress || isProcessRunAdmissionClosed(), resumeExecutionWaitComments, getWorktreeExecutionCutoff, resumeQueuedCommentInterrupt, diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 9d504c65874..3f288527682 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -9,6 +9,7 @@ import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { createPostgresWakeQueueAdapter, createWakeAdmissionWriter } from "../../modules/wake-queue/adapters/postgres.js"; import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; +import { closeProcessRunAdmission, resetProcessRunAdmissionForTests } from "./run-admission.js"; import { createHeartbeatQueue, DEFERRED_WAKE_CONTEXT_KEY, shouldDeferFollowupWakeForSameIssue, shouldQueueFollowupForRunningIssueWake, type HeartbeatQueueDependencies } from "./queue.js"; import { createWakeQueue } from "../../modules/wake-queue/index.js"; import { TransactionScope } from "../../modules/wake-queue/application/ports.js"; @@ -50,6 +51,30 @@ describe("heartbeat queue module boundary", () => { expect(deps.executeRun).not.toHaveBeenCalled(); }); + it.each(["startNextQueuedRunForAgent", "resumeQueuedRuns"] as const)("starts no run once shutdown closed run admission (%s)", async (operation) => { + const database = guardedDatabase(); + const deps = callbacks(database.db); + deps.isRunAdmissionClosed = () => true; + const queue = createHeartbeatQueue(database.db, deps); + await queue[operation]("agent"); + expect(database.access).not.toHaveBeenCalled(); + expect(deps.getAgent).not.toHaveBeenCalled(); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + + it("honors the process-wide run admission gate in every heartbeat instance", async () => { + const database = guardedDatabase(); + const deps = callbacks(database.db); + closeProcessRunAdmission(); + try { + await createHeartbeatQueue(database.db, deps).startNextQueuedRunForAgent("agent"); + } finally { + resetProcessRunAdmissionForTests(); + } + expect(database.access).not.toHaveBeenCalled(); + expect(deps.getAgent).not.toHaveBeenCalled(); + }); + it("tracks pending wakes across instances and clears each rejected wake", async () => { const database = guardedDatabase(); const deps = callbacks(database.db); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 0f0ebdf8f7d..5f3c2fe48d2 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -1,4 +1,5 @@ import { CONTEXT_COMPACTION_WAKE_REASON, ContextCompactionError, isContextCompactionRun, readContextCompactionRequest } from "../context-compaction.js"; +import { isProcessRunAdmissionClosed } from "./run-admission.js"; import { deliveredQueuedCommentIds } from "../queued-comment-delivery.js"; import { deferredWorkIntentWait } from "../work-intent-outbox.js"; import { createHeartbeatTimerPolicy } from "./timer-policy.js"; @@ -227,6 +228,7 @@ export interface HeartbeatQueueDependencies extends Pick string | null; filterZombieCoalesceTarget: (run: T | null, live: { has(id: string): boolean }) => T | null; getSchedulingSuppression: () => Promise<{ suppressed: boolean; reason: "worktree_instance" | "database_restore_in_progress" | "task_drain" | null }>; + isRunAdmissionClosed?: () => boolean; canStartNewExternalBotWork: (agent: typeof agents.$inferSelect) => Promise; executeRun: (runId: string) => Promise; releaseIssueExecutionAndPromote: (run: Pick, options?: { suppressImmediateRecovery?: boolean; resumeStrandedQueue?: boolean; deferredPostCommitEffects?: WakeQueuePostCommitEffect[] }) => Promise; @@ -411,6 +413,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende isHeartbeatRunTerminalStatus, liveRunExecutions, getSchedulingSuppression, + isRunAdmissionClosed = isProcessRunAdmissionClosed, resumeExecutionWaitComments, getWorktreeExecutionCutoff, resumeQueuedCommentInterrupt, @@ -1947,6 +1950,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } async function resumeQueuedRuns() { + if (isRunAdmissionClosed()) return; await resumeDeferredExecutionReceipts(); await resumeQueuedRunsAfterDrain(); } @@ -1954,7 +1958,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende // Release admission promptly without first scanning unrelated recovery work. // Every run still passes the ordinary per-agent claim and dispatch gates. async function resumeQueuedRunsAfterDrain(scope?: { companyId: string }) { - if ((await getSchedulingSuppression()).suppressed) return; + if (isRunAdmissionClosed() || (await getSchedulingSuppression()).suppressed) return; const cutoff = await getWorktreeExecutionCutoff(); const queuedRuns = await db .select({ agentId: heartbeatRuns.agentId }) @@ -2048,7 +2052,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende } async function startNextQueuedRunForAgent(agentId: string) { - if ((await getSchedulingSuppression()).suppressed) return []; + if (isRunAdmissionClosed() || (await getSchedulingSuppression()).suppressed) return []; const cutoff = await getWorktreeExecutionCutoff(); // Cancelled after the start lock is released: cancelRunInternal promotes the // agent's next queued run, which takes this same lock. diff --git a/server/src/services/heartbeat/run-admission.ts b/server/src/services/heartbeat/run-admission.ts new file mode 100644 index 00000000000..d61a0b48bd0 --- /dev/null +++ b/server/src/services/heartbeat/run-admission.ts @@ -0,0 +1,13 @@ +let runAdmissionClosed = false; + +export function closeProcessRunAdmission(): void { + runAdmissionClosed = true; +} + +export function isProcessRunAdmissionClosed(): boolean { + return runAdmissionClosed; +} + +export function resetProcessRunAdmissionForTests(): void { + runAdmissionClosed = false; +} diff --git a/server/src/shutdown.test.ts b/server/src/shutdown.test.ts index 83176e35210..728896f0535 100644 --- a/server/src/shutdown.test.ts +++ b/server/src/shutdown.test.ts @@ -281,6 +281,19 @@ describe("closeHttpListenerForShutdown", () => { expect(server.closeAllConnections).not.toHaveBeenCalled(); }); + it("closes run admission before it stops accepting requests", async () => { + const order: string[] = []; + const server = fakeServer({ listening: true, closeDelayMs: 0 }); + server.close.mockImplementation((callback?: (err?: Error) => void) => { order.push("listener"); setTimeout(() => callback?.(), 0); }); + await closeHttpListenerForShutdown({ server, signal: "SIGTERM", timeoutMs: 1_000, log: stubLogger(), + closeRunAdmission: () => order.push("run_admission") }); + expect(order).toEqual(["run_admission", "listener"]); + const stopped = vi.fn(); + await expect(closeHttpListenerForShutdown({ server: fakeServer({ listening: false }), signal: "SIGTERM", log: stubLogger(), + closeRunAdmission: stopped })).resolves.toBe("not_listening"); + expect(stopped).toHaveBeenCalledOnce(); + }); + it("closes the remaining connections when the drain outlives the grace period", async () => { vi.useFakeTimers(); try { diff --git a/server/src/shutdown.ts b/server/src/shutdown.ts index ebc9e590b8f..475df7c095b 100644 --- a/server/src/shutdown.ts +++ b/server/src/shutdown.ts @@ -127,8 +127,10 @@ export async function closeHttpListenerForShutdown(input: { signal: "SIGINT" | "SIGTERM"; timeoutMs?: number; drainRequests?: (signal: AbortSignal) => Promise; + closeRunAdmission?: () => void; log: ShutdownLogger; }): Promise<"closed" | "timed_out" | "not_listening"> { + input.closeRunAdmission?.(); if (!input.server.listening) return "not_listening"; const timeoutMs = input.timeoutMs ?? 5_000; let timer: NodeJS.Timeout | null = null;