mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(heartbeat): resume legacy runs a graceful shutdown stopped before dispatch (#20461)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work
> - The heartbeat service runs agent turns and recovers them after a
server restart
> - A graceful restart can stop a legacy run (for example a `process`
adapter agent) while the run is still in setup, before the adapter
starts
> - That run was parked with `legacy_execution_requires_reconciliation`,
so automatic recovery stopped
> - A `process` or `http` adapter cannot continue that hold through
Retry, so the accepted Retry waited forever and the task stayed blocked
> - This pull request proves "no adapter work started" for that case and
resumes the run automatically
> - The benefit is that a restart during setup no longer strands the
task, and Retry starts a run for holds that an older server saved
## Linked Issues or Issue Description
**What happened?**
The restart-chaos harness ran workload Q3 (a board approves an agent's
app action late) with a `process` adapter agent. A graceful restart
(fault K2) fired right after the approval. The agent's follow-up run was
interrupted with `server_shutdown_interrupted` 0.2 s after it started,
while it was still in setup. Automatic recovery stopped with
`legacy_execution_requires_reconciliation`. The board clicked Retry. The
server accepted the Retry (HTTP 202), but no run started. The task
showed "Retry is waiting for execution checks" (`retryWait.reason =
execution_recovery`) forever.
**Expected behavior**
A run that a graceful shutdown stops before any adapter work must resume
automatically. An accepted Retry must start a run.
**Steps to reproduce**
1. Run a `process` adapter agent on a task.
2. Send SIGTERM to the server while the run is still in setup (before
the adapter starts).
3. Restart the server. The task is blocked with
`legacy_execution_requires_reconciliation`.
4. Click Retry. The request is accepted, but no run starts.
**Paperclip version or commit**
master at 2c2186706e
**Additional context**
The graceful-shutdown drain marks a legacy run `interrupted` without any
bootstrap evidence. The legacy classifier then treats the run as
"outcome unknown" and creates a no-replay hold. Only conversation
adapters can continue that hold through Retry
(`admitExplicitNativeContinuation` returns null for `process`/`http`
adapters), so the Retry intent stays deferred with no owner. The
hard-restart reaper already uses the durable stage proof
(`executionStage = 'preparing'`, no process identity) to retry such
runs. The graceful path did not.
## What Changed
- New `server/src/services/legacy-shutdown-before-dispatch.ts`: one SQL
proof that a legacy run interrupted by graceful shutdown never reached
the adapter. The proof requires: `executionStage = 'preparing'` with a
controller boot ID, no `budgetDispatch.handedOffAt` handoff receipt, no
process identity, no execution-start event, and no native or
workspace-recovery obligation.
- `terminalizeLegacyExecution` applies that proof in the same
transaction that commits the shutdown interruption, under the run lock.
A matching run gets `executionRecovery: { kind: "bootstrap",
providerWorkStarted: false }`. The existing process-loss retry then
schedules it in the interrupted-preparation lane (no delay, no failure
budget spent), and no hold is created.
- `executionBlockerPredicate` ignores a
`legacy_execution_requires_reconciliation` hold whose source run carries
the same proof. This releases holds that an older server saved for the
same case, so Retry starts a run (upgrade-safe for old rows).
- Tests in `server/src/__tests__/heartbeat-process-recovery.test.ts`.
## Verification
- Red before, green after:
- `resumes a legacy run that a graceful shutdown stopped before adapter
dispatch (stage=preparing)`
- `lets Retry start a run for a pre-dispatch shutdown hold saved by an
older server`
- Control (passes before and after): `... (stage=dispatching)` keeps the
reconciliation hold and schedules no retry.
- `cd server && npx vitest run
src/__tests__/heartbeat-process-recovery.test.ts
src/services/heartbeat/recovery.test.ts
src/__tests__/heartbeat-retry-scheduling.test.ts
src/__tests__/interrupted-preparation-hold-settle.test.ts
src/__tests__/interrupted-preparation-restore-retry.test.ts` (609
passed).
- `npx tsc --noEmit -p server` passes.
- Live proof: restart-chaos `--env local --adapters codex --workloads Q3
--faults K2 --matrix --concurrency 2 --seed 109`. Results are added in a
comment.
## Risks
- The proof is narrow. A run that reached `dispatching`, recorded a
handoff, a process identity or any execution event keeps its
reconciliation hold. Rows from a controller without a lease (no boot ID)
keep their hold.
- The stage proof relies on the `dispatching` stage write that the lease
renewal makes before adapter admission. That write and the `preparing`
claim were added in the same change, so old rows with `preparing` are
reliable.
- Not changed: a `process` or `http` run that really started and has
unknown outcomes still keeps its hold. Retry for that hold is still
accepted but cannot continue it. That is a separate product decision.
## Model Used
- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use and code execution.
## 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 <noreply@anthropic.com>
This commit is contained in:
1 parent
693de27c90
commit
233f615644
4 files changed
+178
No files matched your search
@@ -4890,6 +4890,94 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: "still preparing", resumes: true },
|
||||
{ name: "admitted to dispatch", stage: "dispatching", resumes: false },
|
||||
{ name: "with a dispatch handoff receipt", handoff: true, resumes: false },
|
||||
{ name: "with a process identity", processStarted: true, resumes: false },
|
||||
{ name: "with an adapter.invoke event", invoked: true, resumes: false },
|
||||
{ name: "with a workspace restore failure", restoreFailure: true, resumes: false },
|
||||
] as Array<{ name: string; resumes: boolean; stage?: string; handoff?: boolean; processStarted?: boolean;
|
||||
invoked?: boolean; restoreFailure?: boolean }>)("resumes a legacy run that a graceful shutdown stopped before adapter dispatch ($name)", async (variant) => {
|
||||
const { resumes } = variant;
|
||||
const { agentId, runId, issueId } = await seedRunFixture({ adapterType: "process", agentStatus: "running" });
|
||||
await db.update(heartbeatRuns).set({
|
||||
controllerBootId: controllerLeases.legacyControllerBootId,
|
||||
controllerLeaseExpiresAt: new Date(Date.now() + 60_000),
|
||||
executionStage: variant.stage ?? "preparing",
|
||||
startedAt: new Date(),
|
||||
...(variant.handoff ? { usageJson: { budgetDispatch: { reservationId: randomUUID(), recoveryLeaseOwner: null,
|
||||
admittedAt: new Date().toISOString(), handedOffAt: new Date().toISOString() } } } : {}),
|
||||
...(variant.processStarted ? { processStartedAt: new Date() } : {}),
|
||||
...(variant.restoreFailure ? { resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } } : {}),
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
// Setup had not reached the adapter: no adapter.invoke was recorded.
|
||||
if (!variant.invoked) await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
||||
|
||||
const result = await heartbeatService(db).drainRunningRunsForShutdown("SIGTERM");
|
||||
|
||||
expect(result.interruptedRunIds).toEqual([runId]);
|
||||
const [interrupted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
expect(interrupted).toMatchObject({ status: "interrupted", errorCode: "server_shutdown_interrupted" });
|
||||
if (resumes) {
|
||||
expect(interrupted!.resultJson?.executionRecovery).toEqual({ kind: "bootstrap", providerWorkStarted: false });
|
||||
expect(result.retryRunIds).toHaveLength(1);
|
||||
expect(await selectRecoveryActionsForIssue(issueId)).toEqual([]);
|
||||
expect(await getExecutionBlocker(db, interrupted!.companyId, issueId)).toBeNull();
|
||||
expect(await selectRecoveryRunsByAgent(agentId)).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: result.retryRunIds[0], retryOfRunId: runId, status: "scheduled_retry" }),
|
||||
]));
|
||||
} else {
|
||||
expect(interrupted!.resultJson?.executionRecovery).toBeUndefined();
|
||||
expect(result.retryRunIds).toEqual([]);
|
||||
expect(await selectRecoveryActionsForIssue(issueId)).toEqual([
|
||||
expect.objectContaining({ cause: "legacy_execution_requires_reconciliation" }),
|
||||
]);
|
||||
expect(await getExecutionBlocker(db, interrupted!.companyId, issueId)).toMatchObject({ runId });
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ name: "before dispatch", startedWork: false },
|
||||
{ name: "after an adapter.invoke event", startedWork: true },
|
||||
])("lets Retry start a run only for a pre-dispatch shutdown hold saved by an older server ($name)", async ({ startedWork }) => {
|
||||
const { companyId, agentId, runId, issueId } = await seedRunFixture({ adapterType: "process", agentStatus: "idle" });
|
||||
// The older server committed the interruption and its no-replay hold
|
||||
// without the bootstrap certificate.
|
||||
await db.update(heartbeatRuns).set({
|
||||
status: "interrupted", errorCode: "server_shutdown_interrupted", error: "Interrupted by graceful server shutdown (SIGTERM)",
|
||||
finishedAt: new Date(), startedAt: new Date(Date.now() - 1_000),
|
||||
controllerBootId: randomUUID(), controllerLeaseExpiresAt: null, executionStage: "preparing",
|
||||
resultJson: { stopReason: "interrupted", timeoutFired: false, timeoutSource: "default", timeoutConfigured: false, effectiveTimeoutSec: 0 },
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
if (!startedWork) await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId));
|
||||
await db.update(issues).set({ status: "blocked", executionRunId: null, checkoutRunId: null }).where(eq(issues.id, issueId));
|
||||
await db.insert(issueRecoveryActions).values({
|
||||
companyId, sourceIssueId: issueId, kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: agentId,
|
||||
cause: "legacy_execution_requires_reconciliation", fingerprint: `legacy-execution:${runId}`, status: "active",
|
||||
evidence: { runId, originalFailureCode: "server_shutdown_interrupted", adapterRecovery: "unsupported_or_unknown", attempt: 1 },
|
||||
nextAction: "Inspect the stopped provider and recorded actions, then reconcile their outcomes before continuing.",
|
||||
attemptCount: 1, maxAttempts: 3,
|
||||
});
|
||||
|
||||
if (startedWork) {
|
||||
// The adapter may have acted: the hold stays and no replay is admitted.
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toMatchObject({ runId });
|
||||
return;
|
||||
}
|
||||
expect(await getExecutionBlocker(db, companyId, issueId)).toBeNull();
|
||||
const heartbeat = heartbeatService(db);
|
||||
const receipt = await requestTaskRetry({ db, companyId, issueId, actorId: "responsible-user",
|
||||
idempotencyKey: randomUUID(), observedRunId: runId });
|
||||
expect(receipt.accepted).toBe(true);
|
||||
await heartbeat.resumeTaskRetry(companyId, receipt.requestId!);
|
||||
const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!));
|
||||
expect(wake?.payload?.executionWait).toBeUndefined();
|
||||
const runs = await selectRecoveryRunsByAgent(agentId);
|
||||
expect(runs.filter(run => run.id !== runId)).toHaveLength(1);
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
});
|
||||
|
||||
it("suspends native Paperclip Runner ownership on graceful restart without cancelling or creating a retry run", async () => {
|
||||
const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({
|
||||
adapterType: "paperclip_runner",
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { isPreDispatchReviewWait, neverStartedDispatchCancellationPredicate } from "./pre-dispatch-review-wait.js";
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
import { preDispatchExecutionWaitRecoveryActionPredicate } from "./pre-dispatch-execution-wait.js";
|
||||
import { legacyShutdownBeforeDispatchCondition } from "./legacy-shutdown-before-dispatch.js";
|
||||
import { and, desc, eq, gt, inArray, not, or, sql, type SQL } from "drizzle-orm";
|
||||
import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js";
|
||||
import { z } from "zod";
|
||||
@@ -115,6 +116,17 @@ export function executionBlockerPredicate() {
|
||||
return and(
|
||||
not(conversationRecoveryActionPredicate()!),
|
||||
not(preDispatchExecutionWaitRecoveryActionPredicate()),
|
||||
// Holds saved before shutdown certified pre-dispatch interruptions carry
|
||||
// the same row proof; they have no unknown outcome to reconcile.
|
||||
sql`not (${issueRecoveryActions.cause} = 'legacy_execution_requires_reconciliation' and exists (
|
||||
select 1 from ${heartbeatRuns}
|
||||
where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId}
|
||||
and ${heartbeatRuns.id} = ${canonicalExecutionUuid(sql`coalesce(${issueRecoveryActions.evidence}->>'runId', ${issueRecoveryActions.evidence}->>'sourceRunId')`)}
|
||||
and ${issueRecoveryActions.sourceIssueId} = ${canonicalExecutionUuid(sql`${heartbeatRuns.contextSnapshot}->>'issueId'`)}
|
||||
and ${legacyShutdownBeforeDispatchCondition()}
|
||||
and coalesce(${issueRecoveryActions.evidence}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive'
|
||||
and ${issueRecoveryActions.evidence}->'workspaceRestoreRecovery' is null
|
||||
))`,
|
||||
sql`not (${issueRecoveryActions.cause} = 'legacy_execution_requires_reconciliation' and exists (
|
||||
select 1 from ${heartbeatRuns}
|
||||
where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId}
|
||||
|
||||
@@ -31,6 +31,7 @@ import { isSupersededConversationRun } from "./agent-conversations.js";
|
||||
import { issueService } from "./issues.js";
|
||||
import { isInterruptedNativePreparation, isInterruptedNativePreparationCandidate } from "./cancelled-native-startup.js";
|
||||
import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js";
|
||||
import { recordLegacyShutdownBeforeDispatch, SERVER_SHUTDOWN_INTERRUPTED_CODE } from "./legacy-shutdown-before-dispatch.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
export const LEGACY_RECOVERY_CAUSE = "legacy_execution_requires_reconciliation";
|
||||
@@ -169,6 +170,12 @@ export async function terminalizeLegacyExecution(input: {
|
||||
)
|
||||
.returning();
|
||||
if (!updated) return null;
|
||||
// A graceful shutdown that stopped setup before adapter admission did no
|
||||
// provider work: certify it here, under the run lock, so the process-loss
|
||||
// retry resumes it instead of parking it for reconciliation.
|
||||
if (!input.recordRestoreFailureOnly && current.status === "running" && updated.status === "interrupted"
|
||||
&& updated.errorCode === SERVER_SHUTDOWN_INTERRUPTED_CODE)
|
||||
updated = await recordLegacyShutdownBeforeDispatch(tx as unknown as Db, updated) ?? updated;
|
||||
const retainedLeaseIds = await preserveLegacyWorkspaceRestoreSources(tx as unknown as Db,
|
||||
input.recordRestoreFailureOnly ? { ...updated, status: "failed" } : updated, input.workspaceRestoreSource);
|
||||
if (retainedLeaseIds.length) {
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq, isNotNull, isNull, sql } from "drizzle-orm";
|
||||
import { heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { executionStartedEventPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
|
||||
export const SERVER_SHUTDOWN_INTERRUPTED_CODE = "server_shutdown_interrupted";
|
||||
|
||||
/**
|
||||
* A graceful shutdown stopped a legacy run while it was still preparing.
|
||||
*
|
||||
* The proof is every clause together; none is redundant:
|
||||
* - `budgetDispatch.handedOffAt`: the dispatch gate
|
||||
* (`dispatchResolvedInteractionIfCurrent`) writes it under the run lock
|
||||
* before it calls the adapter. It is the dispatch linearization point.
|
||||
* - `executionStage`: every legacy dispatch first renews its controller lease
|
||||
* with stage `dispatching` (`authorizeFailedChatRetryExecution("dispatching")`
|
||||
* in `dispatchResolvedInteractionContinuationAfterAdmission`).
|
||||
* - No process identity and no execution-start event (`adapter.invoke`,
|
||||
* process start/identity, session or provider events).
|
||||
* The gate and the renewal both require `status = 'running'`, so once the
|
||||
* shutdown commits the interruption these records can no longer change.
|
||||
*
|
||||
* Rows from a controller without a lease (no boot ID) or with any native,
|
||||
* remote-workspace or file-recovery obligation stay with their own recovery.
|
||||
*/
|
||||
export function legacyShutdownBeforeDispatchCondition() {
|
||||
return and(
|
||||
eq(heartbeatRuns.runtimeMode, "legacy"),
|
||||
eq(heartbeatRuns.status, "interrupted"),
|
||||
eq(heartbeatRuns.errorCode, SERVER_SHUTDOWN_INTERRUPTED_CODE),
|
||||
isNotNull(heartbeatRuns.finishedAt),
|
||||
isNotNull(heartbeatRuns.controllerBootId),
|
||||
eq(heartbeatRuns.executionStage, "preparing"),
|
||||
isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), isNull(heartbeatRuns.processStartedAt),
|
||||
isNull(heartbeatRuns.nativeIssueId), isNull(heartbeatRuns.nativeSessionId), isNull(heartbeatRuns.sessionIdAfter),
|
||||
sql`${heartbeatRuns.usageJson}#>'{budgetDispatch,handedOffAt}' is null`,
|
||||
sql`coalesce(${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}', 'false'::jsonb) = 'false'::jsonb`,
|
||||
sql`${heartbeatRuns.resultJson}->'workspaceRestoreFailure' is null`,
|
||||
sql`${heartbeatRuns.resultJson}->'workspaceRestoreRecovery' is null`,
|
||||
sql`not (coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) ?| array[
|
||||
'nativeExecutionInput', 'nativeWorkspaceSync', 'remoteWorkspace', 'nativeWorkspaceFinalizationOwner', 'sessionCheckpoint'])`,
|
||||
sql`coalesce(${heartbeatRuns.runnerProfileJson}#>>'{adapterDispatch,adapterType}', '') <> 'paperclip_runner'`,
|
||||
sql`not exists (select 1 from ${nativeRunFinalizations}
|
||||
where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id})`,
|
||||
sql`not exists (select 1 from ${heartbeatRunEvents}
|
||||
where ${and(eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId),
|
||||
eq(heartbeatRunEvents.runId, heartbeatRuns.id), executionStartedEventPredicate())})`,
|
||||
)!;
|
||||
}
|
||||
|
||||
/**
|
||||
* Record the bootstrap certificate on a run the shutdown just interrupted
|
||||
* before dispatch. Call it in the transaction that committed the interruption,
|
||||
* so the row lock fences a concurrent stage or handoff write. Returns the
|
||||
* stamped row, or null when the run does not carry that proof.
|
||||
*/
|
||||
export async function recordLegacyShutdownBeforeDispatch(tx: Db, run: Pick<Run, "id" | "companyId">): Promise<Run | null> {
|
||||
const [stamped] = await tx.update(heartbeatRuns).set({
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
|
||||
|| jsonb_build_object('executionRecovery', jsonb_build_object('kind', 'bootstrap', 'providerWorkStarted', false))`,
|
||||
executionStatusDeliveryId: randomUUID(),
|
||||
updatedAt: new Date(),
|
||||
}).where(and(
|
||||
eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId),
|
||||
sql`${heartbeatRuns.resultJson}->'executionRecovery' is null`,
|
||||
legacyShutdownBeforeDispatchCondition(),
|
||||
)).returning();
|
||||
return stamped ?? null;
|
||||
}
|
||||
Reference in new issue
Block a user