diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d0dffae1581..35a8bda18e8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3298,6 +3298,18 @@ results, intentional stops, and historical unmarked failures keep their existing terminal behavior. This continuation uses the already charged restart attempt and does not reset the provider-attempt budget. +The same continuation applies when the runner is lost without a server restart and +the run resumes itself after the retry delay (a local same-run exact-checkpoint +resume). Three durable records must hold. First, the latest process record of the +run is a verified stop of the previous runner (`native.local_process_stopped`) and +that process is gone. Second, the resume claimed a provider attempt within the +three-attempt budget from a `retryable_failure` coordinator. Third, that failure +saved `recoveryMode: "exact_checkpoint_resume"`, `recoverable`, +`providerSessionEstablished` and `checkpointExists`, with no replacement recorded, +and the runner restores that same checkpointed conversation. A remote run, a +bootstrap retry, an ambiguous state, a missing or superseded stop record, an +exhausted budget, or a replacement session keeps the existing failure. + A restart can also land after the agent proposed its result but before the provider turn ended. When the restored runner proves that turn ended with the result Paperclip accepted, it records a reconciliation and the run terminal, diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 9b9d6c170ac..0dfda06c42f 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -220,7 +220,8 @@ function canonicalTestJson(value: unknown): string { const RESTART_LOST_PROCESS_GUIDANCE = "may no longer be running"; function expectRestartContinuationGuidance(text: string) { const events = String(JSON.parse(text).events); - expect(events).toContain("This is a resumed session after a Paperclip control-plane restart."); + expect(events).toContain("This is a resumed session. The process running the previous turn was lost"); + expect(events).not.toContain("control-plane restart"); expect(events).toContain("Shell commands and background processes started in earlier turns " + RESTART_LOST_PROCESS_GUIDANCE); expect(events).toContain("check the workspace for the outputs or artifacts"); expect(events).toContain("If the expected output is missing, re-run the command."); diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 178c28ec56b..7b9d0971362 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -2731,10 +2731,10 @@ export async function executeNativeSession( message: { role: "user" as const, text: JSON.stringify({ schema: "paperclip.native-continuation.v1", events: [ - "This is a resumed session after a Paperclip control-plane restart. Paperclip restored this same conversation and workspace; the restart interrupted the previous turn.", + "This is a resumed session. The process running the previous turn was lost, for example in a Paperclip restart. Paperclip restored this same conversation and workspace; that loss interrupted the previous turn.", "Continue the current user request from the recorded progress and preserved workspace.", ...(steeredMessages.length > 0 ? [ - "During the interrupted turn the user sent the messages in steeredMessages. The restart happened before you read them. Handle each one in this turn, oldest first, together with the current request.", + "During the interrupted turn the user sent the messages in steeredMessages. The interruption happened before you read them. Handle each one in this turn, oldest first, together with the current request.", ] : []), "Shell commands and background processes started in earlier turns may no longer be running, and their tool sessions may be gone. Do not block only because such a process or its tool session cannot be found or inspected.", "Reconcile any unfinished tool or command before proceeding: check the workspace for the outputs or artifacts it should have produced. If the expected output is present, use it. If the expected output is missing, re-run the command.", diff --git a/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts b/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts new file mode 100644 index 00000000000..1d02ca75eec --- /dev/null +++ b/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts @@ -0,0 +1,105 @@ +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; + +const provider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ + ...(await importOriginal()), + executeNativeSession: provider.execute, +})); + +import { executePaperclipNativeSession } from "./native-session-executor.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; + +const EXITED_PID = 4_194_304; + +describe("a scheduled same-run resume reaching the provider runtime", () => { + let temporary: Awaited>; + let db: ReturnType; + const companyId = randomUUID(); + const agentId = randomUUID(); + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-scheduled-resume-continuation-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Scheduled resume", issuePrefix: "NSR" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native agent", status: "active", + adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "test-model" } }); + }); + afterAll(async () => { await temporary?.cleanup(); }); + + const exactResume = { recoveryMode: "exact_checkpoint_resume", recoverable: true, providerSessionEstablished: true, + checkpointExists: true, originalFailureCode: "native_session_recovery_failed" }; + + async function resumeOnce(options: { failureDetail: Record; stopRecorded: boolean }) { + const issueId = randomUUID(); + const runId = randomUUID(); + const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Resume the lost turn", + status: "in_progress", workMode: "standard", assigneeAgentId: agentId }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", + runtimeMode: "native", nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", + contextSnapshot: { issueId } }).returning(); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const native = await prepareNativeHeartbeatRun({ db, run: run!, issue: issue!, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId }, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: issueId, title: issue!.title, description: null, prompt: "Continue the work.", workMode: "standard" }, + workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, + interactionResponses: [], credentialBindings: [], + }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, + failureCode: "native_session_interrupted", failureDetail: options.failureDetail }); + const identity = { processPid: EXITED_PID, processGroupId: EXITED_PID, processStartedAt: new Date().toISOString() }; + const events = [ + { eventType: "native.process_start_requested", payload: { providerAttempt: 1 } }, + { eventType: "native.process_identity_recorded", payload: identity }, + ...(options.stopRecorded ? [{ eventType: "native.local_process_stopped", payload: identity }] : []), + ]; + const [last] = await db.select({ seq: sql`coalesce(max(${heartbeatRunEvents.seq}), 0)::int` }) + .from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); + let seq = last?.seq ?? 0; + for (const event of events) { + seq += 1; + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq, eventType: event.eventType, + stream: "system", level: "info", payload: event.payload }); + } + const passed: Array = []; + provider.execute.mockReset().mockImplementation(async (input: { resumeInterruptedTurn?: boolean }) => { + passed.push(input.resumeInterruptedTurn); + throw new Error("scheduled_resume_fixture_stop"); + }); + await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, backend: { + descriptor: async () => { throw new Error("Unexpected provider call"); }, + openSession: async () => { throw new Error("Unexpected provider call"); }, + } }).catch(() => undefined); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + return { passed, coordinator: coordinator! }; + } + + it("lets the runtime continue a lost turn after a verified stop and an admitted attempt", async () => { + const { passed, coordinator } = await resumeOnce({ failureDetail: exactResume, stopRecorded: true }); + expect(passed).toEqual([true]); + expect(coordinator.attempt).toBe(2); + }); + + it.each([ + ["no stop of the previous runner was recorded", { failureDetail: exactResume, stopRecorded: false }], + ["the saved failure is a bootstrap retry", { failureDetail: { ...exactResume, recoveryMode: "bootstrap_retry" }, stopRecorded: true }], + ["the saved failure has no checkpoint", { failureDetail: { ...exactResume, checkpointExists: false }, stopRecorded: true }], + ])("keeps the lost turn failing when %s", async (_name, options) => { + const { passed } = await resumeOnce(options); + expect(passed).toEqual([false]); + }); +}); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index ecd8920ca7c..535f9636339 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -119,6 +119,7 @@ import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js"; import { uploadVerifiedProviderPack } from "./provider-pack-upload.js"; import { selectRemotePiCompanion } from "./remote-pi-companion.js"; +import { scheduledResumeContinuationAllowed } from "./scheduled-resume-continuation.js"; import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PRIOR_OWNER_FENCE_REFUSED, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { connectorToolDefinitionsForSkills } from "../connector-runtime.js"; @@ -10063,6 +10064,7 @@ async function executePaperclipNativeSessionAttempt( let confirmedLeaseDeadline = 0; let claimedStartupContext: NativeExecutionContext | undefined; let freshClaimAttempt: number | undefined; + let scheduledResumeContinuation = false; try { attempt = await trace.measure( "native.coordinator.claim", @@ -10212,6 +10214,15 @@ async function executePaperclipNativeSessionAttempt( ); if (nextAttempt > 3) throw new Error("native_session_retry_exhausted"); + scheduledResumeContinuation = (await scheduledResumeContinuationAllowed(tx as unknown as Db, { + companyId: input.execution.binding.companyId, + runId: input.execution.binding.runId, + restartRecovery: Boolean(recovering), + targetKind: input.runnerExecutionTarget?.kind === "remote" ? "remote" : "local", + coordinatorPhase: coordinator.phase, + failureDetail: coordinator.failureDetail, + admittedAttempt: nextAttempt, + })).allowed; const nextControllerGeneration = recovering ? recovering.controllerGeneration : coordinator.controllerBootId === controller.bootId @@ -11302,7 +11313,7 @@ async function executePaperclipNativeSessionAttempt( trace.activate(runnerSessionStartupScope); const runExecution = () => executeNativeSession({ - resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner", + resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner" || scheduledResumeContinuation, getFreshSessionHandoff: input.getFreshSessionHandoff, // A failed seal commits nothing: the messages stay queued for the // turn boundary instead of failing this run before its turn. diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.test.ts b/server/src/services/native-runtime/scheduled-resume-continuation.test.ts new file mode 100644 index 00000000000..b48033c0eaf --- /dev/null +++ b/server/src/services/native-runtime/scheduled-resume-continuation.test.ts @@ -0,0 +1,140 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; +import { + decideScheduledResumeContinuation, + scheduledResumeContinuationAllowed, + type ScheduledResumeContinuationEvidence, +} from "./scheduled-resume-continuation.js"; + +const proven: ScheduledResumeContinuationEvidence = { + restartRecovery: false, + targetKind: "local", + coordinatorPhase: "retryable_failure", + failureDetail: { + recoveryMode: "exact_checkpoint_resume", + recoverable: true, + providerSessionEstablished: true, + checkpointExists: true, + originalFailureCode: "native_session_recovery_failed", + }, + admittedAttempt: 2, + localStop: { processPid: 81495 }, +}; + +describe("same-run scheduled resume continuation of a turn lost on restore", () => { + it("allows the continuation when the stop, the admitted attempt and the exact checkpoint are all recorded", () => { + expect(decideScheduledResumeContinuation(proven)).toEqual({ allowed: true }); + expect(decideScheduledResumeContinuation({ ...proven, admittedAttempt: 3 })).toEqual({ allowed: true }); + }); + + const detail = (patch: Record) => ({ ...proven, failureDetail: { ...proven.failureDetail, ...patch } }); + it.each<[string, ScheduledResumeContinuationEvidence, string]>([ + ["a restart recovery claim owns its own rule", { ...proven, restartRecovery: true }, "restart_recovery_claim"], + ["the run executes on a remote target", { ...proven, targetKind: "remote" }, "remote_target"], + ["the coordinator is not a settled same-run retry", { ...proven, coordinatorPhase: "observed" }, "not_a_scheduled_same_run_resume"], + ["a fresh run has no coordinator phase", { ...proven, coordinatorPhase: null }, "not_a_scheduled_same_run_resume"], + ["the retry is a bootstrap retry", detail({ recoveryMode: "bootstrap_retry" }), "not_an_exact_checkpoint_resume"], + ["the state is ambiguous", detail({ recoveryMode: "ambiguous_state" }), "not_an_exact_checkpoint_resume"], + ["no failure detail was saved", { ...proven, failureDetail: null }, "not_an_exact_checkpoint_resume"], + ["the failure is not recoverable", detail({ recoverable: false }), "failure_not_recoverable"], + ["no provider session was established", detail({ providerSessionEstablished: false }), "provider_session_not_established"], + ["no checkpoint exists", detail({ checkpointExists: false }), "checkpoint_missing"], + ["a replacement run was created", detail({ successorRunId: "run-2" }), "replacement_recorded"], + ["a replacement was denied", detail({ replacementDenied: "uncertain_provider_action" }), "replacement_recorded"], + ["the attempt exceeds the budget", { ...proven, admittedAttempt: 4 }, "attempt_not_admitted"], + ["no attempt was admitted", { ...proven, admittedAttempt: 0 }, "attempt_not_admitted"], + ["the previous runner has no stop receipt", { ...proven, localStop: null }, "local_stop_unverified"], + ])("refuses when %s", (_name, evidence, reason) => { + expect(decideScheduledResumeContinuation(evidence)).toEqual({ allowed: false, reason }); + }); +}); + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; + +describeEmbeddedPostgres("the durable stop receipt behind a scheduled resume continuation", () => { + let db!: ReturnType; + let stopDatabase: (() => Promise) | undefined; + let companyId!: string; + let agentId!: string; + let deadPid!: number; + + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("paperclip-scheduled-resume-continuation-"); + stopDatabase = started.cleanup; + db = createDb(started.connectionString); + companyId = randomUUID(); + agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Resume", issuePrefix: "SRC", + requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", role: "engineer", status: "active", + adapterType: "paperclip_runner", adapterConfig: { provider: "codex" }, runtimeConfig: {}, permissions: {} }); + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await once(child, "exit"); + deadPid = child.pid!; + }, 120_000); + + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); + await stopDatabase?.(); + }); + + async function seedRun(events: Array<{ eventType: string; payload: Record }>) { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "failed", runtimeMode: "native", + invocationSource: "assignment", triggerDetail: "system", contextSnapshot: {} }); + let seq = 0; + for (const event of events) { + seq += 1; + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq, eventType: event.eventType, + stream: "system", level: "info", payload: event.payload }); + } + return runId; + } + + const identity = () => ({ processPid: deadPid, processGroupId: deadPid, processStartedAt: "2026-11-10T18:56:54.000Z" }); + const decide = (runId: string, patch: Partial = {}) => { + const { localStop: _localStop, ...evidence } = { ...proven, ...patch }; + return scheduledResumeContinuationAllowed(db, { companyId, runId, ...evidence }); + }; + + it("allows the continuation after the previous runner's verified stop", async () => { + const runId = await seedRun([ + { eventType: "native.process_start_requested", payload: { providerAttempt: 1 } }, + { eventType: "native.process_identity_recorded", payload: identity() }, + { eventType: "native.local_process_stopped", payload: identity() }, + ]); + expect(await decide(runId)).toEqual({ allowed: true }); + }); + + it.each<[string, () => Array<{ eventType: string; payload: Record }>]>([ + ["no stop was recorded", () => [ + { eventType: "native.process_identity_recorded", payload: identity() }, + ]], + ["a newer launch follows the stop", () => [ + { eventType: "native.process_identity_recorded", payload: identity() }, + { eventType: "native.local_process_stopped", payload: identity() }, + { eventType: "native.process_start_requested", payload: { providerAttempt: 2 } }, + ]], + ["the stopped process is still alive", () => [ + { eventType: "native.local_process_stopped", + payload: { ...identity(), processPid: process.pid, processGroupId: process.pid } }, + ]], + ["the run never had a process", () => []], + ])("refuses when %s", async (_name, events) => { + const runId = await seedRun(events()); + expect(await decide(runId)).toEqual({ allowed: false, reason: "local_stop_unverified" }); + }); + + it("never reads a stop receipt for a restart claim or a remote target", async () => { + const runId = await seedRun([{ eventType: "native.local_process_stopped", payload: identity() }]); + expect(await decide(runId, { restartRecovery: true })).toEqual({ allowed: false, reason: "restart_recovery_claim" }); + expect(await decide(runId, { targetKind: "remote" })).toEqual({ allowed: false, reason: "remote_target" }); + }); +}); diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.ts b/server/src/services/native-runtime/scheduled-resume-continuation.ts new file mode 100644 index 00000000000..4f315655909 --- /dev/null +++ b/server/src/services/native-runtime/scheduled-resume-continuation.ts @@ -0,0 +1,57 @@ +import type { Db } from "@paperclipai/db"; +import { readNativeLocalProcessStop } from "../native-local-process-stop.js"; +import { NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT } from "./native-restart-recovery.js"; + +export type ScheduledResumeContinuationRefusal = + | "restart_recovery_claim" + | "remote_target" + | "not_a_scheduled_same_run_resume" + | "not_an_exact_checkpoint_resume" + | "failure_not_recoverable" + | "provider_session_not_established" + | "checkpoint_missing" + | "replacement_recorded" + | "attempt_not_admitted" + | "local_stop_unverified"; + +export interface ScheduledResumeContinuationEvidence { + restartRecovery: boolean; + targetKind: "local" | "remote"; + coordinatorPhase: string | null; + failureDetail: Record | null | undefined; + admittedAttempt: number; + localStop: { processPid: number } | null; +} + +export type ScheduledResumeContinuationDecision = + | { allowed: true } + | { allowed: false; reason: ScheduledResumeContinuationRefusal }; + +export function decideScheduledResumeContinuation( + evidence: ScheduledResumeContinuationEvidence, +): ScheduledResumeContinuationDecision { + const refuse = (reason: ScheduledResumeContinuationRefusal) => ({ allowed: false as const, reason }); + if (evidence.restartRecovery) return refuse("restart_recovery_claim"); + if (evidence.targetKind !== "local") return refuse("remote_target"); + if (evidence.coordinatorPhase !== "retryable_failure") return refuse("not_a_scheduled_same_run_resume"); + const detail = evidence.failureDetail ?? {}; + if (detail.recoveryMode !== "exact_checkpoint_resume") return refuse("not_an_exact_checkpoint_resume"); + if (detail.recoverable !== true) return refuse("failure_not_recoverable"); + if (detail.providerSessionEstablished !== true) return refuse("provider_session_not_established"); + if (detail.checkpointExists !== true) return refuse("checkpoint_missing"); + if (detail.successorRunId || detail.replacementDenied) return refuse("replacement_recorded"); + if (!Number.isSafeInteger(evidence.admittedAttempt) || evidence.admittedAttempt < 1 + || evidence.admittedAttempt > NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT) return refuse("attempt_not_admitted"); + if (!evidence.localStop) return refuse("local_stop_unverified"); + return { allowed: true }; +} + +export async function scheduledResumeContinuationAllowed( + db: Db, + input: Omit & { companyId: string; runId: string }, +): Promise { + const { companyId, runId, ...evidence } = input; + const localStop = evidence.restartRecovery || evidence.targetKind !== "local" + ? null : await readNativeLocalProcessStop(db, companyId, runId); + return decideScheduledResumeContinuation({ ...evidence, localStop }); +}