diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 3b1d5d4578c..0f4464cfb5a 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -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", diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 477a8735782..ee6fdd737b7 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -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} diff --git a/server/src/services/legacy-execution-recovery.ts b/server/src/services/legacy-execution-recovery.ts index 0a648c20883..a3b2d4ed333 100644 --- a/server/src/services/legacy-execution-recovery.ts +++ b/server/src/services/legacy-execution-recovery.ts @@ -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) { diff --git a/server/src/services/legacy-shutdown-before-dispatch.ts b/server/src/services/legacy-shutdown-before-dispatch.ts new file mode 100644 index 00000000000..f293e9cdfff --- /dev/null +++ b/server/src/services/legacy-shutdown-before-dispatch.ts @@ -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): Promise { + 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; +}