diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 7101d22b1e3..67d40550fab 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3349,9 +3349,14 @@ 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. +and the runner restores that same checkpointed conversation. The resume reconciler +clears the saved failure when it claims the run. So it first appends a +`same_run_resume` entry to `recovery_history` that copies those fields, together +with its controller boot. The executor then reads that entry. It must be the last +entry, and it must name the executor's boot. A remote run, a bootstrap retry, an +ambiguous state, a missing or superseded stop record, an exhausted budget, a +replacement session, or a claim entry from another boot or followed by another +entry 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 diff --git a/server/src/services/native-runtime/native-exact-resume-failure.test.ts b/server/src/services/native-runtime/native-exact-resume-failure.test.ts index 50405fe9157..4994667ab0f 100644 --- a/server/src/services/native-runtime/native-exact-resume-failure.test.ts +++ b/server/src/services/native-runtime/native-exact-resume-failure.test.ts @@ -50,7 +50,18 @@ describe("interruptedExactResumeCause", () => { }); }); + it("finds the interruption behind the resume reconciler's claim entry for the same attempt", () => { + const resumeClaim = { at: "2099-01-01T00:01:00.000Z", disposition: "same_run_resume", controllerBootId: "boot-1", + providerAttempt: 1, evidence: { recoveryMode: "exact_checkpoint_resume" } }; + expect(interruptedExactResumeCause({ ...claimed, recoveryHistory: [interruption, resumeClaim] })).toEqual({ + originalFailureCode: "native_session_interrupted", + originalMessage: "PRP command turn.start failed: classification=session_resume_required", + interruptedAttempt: 1, + }); + }); + it.each([ + ["a resume claim entry for a later provider attempt", { disposition: "same_run_resume", providerAttempt: 2 }], ["a wait for a later provider attempt", { disposition: "blocked", reason: "held", providerAttempt: 2 }], ["a later recovery entry that is not a wait", { disposition: "resume_dead_runner", reason: "restart", providerAttempt: 1 }], ])("does not reach past %s", (_name, later) => { diff --git a/server/src/services/native-runtime/native-exact-resume-failure.ts b/server/src/services/native-runtime/native-exact-resume-failure.ts index 21c150d416c..5fc73e95008 100644 --- a/server/src/services/native-runtime/native-exact-resume-failure.ts +++ b/server/src/services/native-runtime/native-exact-resume-failure.ts @@ -16,7 +16,7 @@ const FAILURE_DETAIL_MAX_CHARS = 2_000; const REFUSAL_MAX_CHARS = 480; const REFUSAL_SEPARATOR = "; exact checkpoint resume refused: "; /** Restart-recovery waits that keep the interrupted provider attempt. */ -const RECOVERY_WAIT_DISPOSITIONS: ReadonlySet = new Set(["blocked", "awaiting_evidence"]); +const RECOVERY_WAIT_DISPOSITIONS: ReadonlySet = new Set(["blocked", "awaiting_evidence", "same_run_resume"]); /** * Cuts to at most `max` UTF-16 code units without splitting a surrogate pair. diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 97782172df6..c22b0389015 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,3 +1,5 @@ +import { sameRunResumeClaimEntry } from "./scheduled-resume-continuation.js"; +import { appendBoundedRecoveryHistory } from "./native-restart-recovery.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { currentControllerProcessStamp } from "../controller-process-identity.js"; import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js"; @@ -309,15 +311,16 @@ export async function claimNativeSessionResumptions(input: { const leaseExpiresAt = new Date(now.getTime() + 20 * 60_000); await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); + const stamp = currentControllerProcessStamp(); + const resumeClaim = sameRunResumeClaimEntry({ at: now, controllerBootId: stamp.bootId, + providerAttempt: row.coordinator.attempt, failureDetail: row.coordinator.failureDetail }); const updated = await tx.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner, leaseExpiresAt, - ...(() => { - const stamp = currentControllerProcessStamp(); - return { controllerBootId: stamp.bootId, controllerPid: stamp.pid, controllerHostHash: stamp.hostHash, - controllerProcessStartedAt: stamp.processStartedAt ? new Date(stamp.processStartedAt) : null }; - })(), + controllerBootId: stamp.bootId, controllerPid: stamp.pid, controllerHostHash: stamp.hostHash, + controllerProcessStartedAt: stamp.processStartedAt ? new Date(stamp.processStartedAt) : null, + recoveryHistory: appendBoundedRecoveryHistory(resumeClaim), failureCode: null, failureDetail: null, nextAttemptAt: null, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 7c825b3d318..ec0bce99dbd 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -963,7 +963,7 @@ function historyEntry(input: { } satisfies Record; } -function appendBoundedRecoveryHistory(entry: Record) { +export function appendBoundedRecoveryHistory(entry: Record) { const encoded = JSON.stringify(entry); return sql`( select coalesce(jsonb_agg(item order by ordinal), '[]'::jsonb) 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 index 1d02ca75eec..22c96d48e3f 100644 --- 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 @@ -14,6 +14,7 @@ vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ import { executePaperclipNativeSession } from "./native-session-executor.js"; import { buildNativeCompletionContract } from "./completion-contracts.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; const EXITED_PID = 4_194_304; @@ -35,7 +36,7 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { 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 }) { + async function resumeOnce(options: { failureDetail: Record; stopRecorded: boolean; viaReconciler?: boolean }) { const issueId = randomUUID(); const runId = randomUUID(); const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Resume the lost turn", @@ -58,7 +59,12 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, interactionResponses: [], credentialBindings: [], }; - await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution, + sessionCheckpoint: { backendKind: "runner", driverKind: "codex_app_server", sessionId: native.normalizedSessionId, + identity: { companyId, issueId, runId, agentId, sessionId: native.normalizedSessionId }, + providerSessionId: "provider-thread", cursor: "1", activeTurnId: "lost-turn", pendingRuntimeRequests: [], lineage: [] } }, + ...(options.viaReconciler ? { status: "failed" as const, nativePhase: "retryable_failure" } : {}) }) + .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() }; @@ -80,7 +86,13 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { passed.push(input.resumeInterruptedTurn); throw new Error("scheduled_resume_fixture_stop"); }); - await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, backend: { + let leaseOwner: string | undefined; + if (options.viaReconciler) { + const claims = await claimNativeSessionResumptions({ db, runnerInstanceId: native.runnerInstanceId, runIds: [runId] }); + expect(claims).toEqual([expect.objectContaining({ runId })]); + leaseOwner = claims[0]!.leaseOwner; + } + await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, leaseOwner, backend: { descriptor: async () => { throw new Error("Unexpected provider call"); }, openSession: async () => { throw new Error("Unexpected provider call"); }, } }).catch(() => undefined); @@ -88,12 +100,17 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { 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 }); + it.each([false, true])("lets the runtime continue a lost turn after a verified stop and an admitted attempt (claimed by the resume reconciler: %s)", async (viaReconciler) => { + const { passed, coordinator } = await resumeOnce({ failureDetail: exactResume, stopRecorded: true, viaReconciler }); expect(passed).toEqual([true]); expect(coordinator.attempt).toBe(2); }); + it("keeps the lost turn failing when the reconciler claimed a resume without a stop receipt", async () => { + const { passed } = await resumeOnce({ failureDetail: exactResume, stopRecorded: false, viaReconciler: true }); + expect(passed).toEqual([false]); + }); + 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 }], diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index b8373d661ee..dafc06da83e 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10371,6 +10371,9 @@ async function executePaperclipNativeSessionAttempt( targetKind: input.runnerExecutionTarget?.kind === "remote" ? "remote" : "local", coordinatorPhase: coordinator.phase, failureDetail: coordinator.failureDetail, + resumeClaim: Array.isArray(coordinator.recoveryHistory) + ? (coordinator.recoveryHistory.at(-1) as Record | undefined) ?? null : null, + controllerBootId: controller.bootId, admittedAttempt: nextAttempt, })).allowed; const nextControllerGeneration = recovering diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.test.ts b/server/src/services/native-runtime/scheduled-resume-continuation.test.ts index b48033c0eaf..502c3e2657e 100644 --- a/server/src/services/native-runtime/scheduled-resume-continuation.test.ts +++ b/server/src/services/native-runtime/scheduled-resume-continuation.test.ts @@ -8,6 +8,7 @@ import { } from "../../__tests__/helpers/embedded-postgres.js"; import { decideScheduledResumeContinuation, + sameRunResumeClaimEntry, scheduledResumeContinuationAllowed, type ScheduledResumeContinuationEvidence, } from "./scheduled-resume-continuation.js"; @@ -53,6 +54,27 @@ describe("same-run scheduled resume continuation of a turn lost on restore", () ])("refuses when %s", (_name, evidence, reason) => { expect(decideScheduledResumeContinuation(evidence)).toEqual({ allowed: false, reason }); }); + + const resumeClaim = (patch: Record = {}, detail: Record = {}) => + ({ ...sameRunResumeClaimEntry({ at: new Date(), controllerBootId: "boot-1", providerAttempt: 1, + failureDetail: { ...proven.failureDetail, ...detail } }), ...patch }); + const claimed = (claim: Record | null, patch: Partial = {}) => + ({ ...proven, coordinatorPhase: "observed", failureDetail: null, resumeClaim: claim, controllerBootId: "boot-1", ...patch }); + + it("reads the evidence the resume reconciler recorded when it claimed the run", () => { + expect(decideScheduledResumeContinuation(claimed(resumeClaim()))).toEqual({ allowed: true }); + }); + + it.each<[string, ScheduledResumeContinuationEvidence, string]>([ + ["the last history entry is not the resume claim", claimed(resumeClaim({ disposition: "retryable_failure" })), "not_a_scheduled_same_run_resume"], + ["no history entry exists", claimed(null), "not_a_scheduled_same_run_resume"], + ["another boot recorded the resume claim", claimed(resumeClaim(), { controllerBootId: "boot-2" }), "not_a_scheduled_same_run_resume"], + ["the claimed failure was a bootstrap retry", claimed(resumeClaim({}, { recoveryMode: "bootstrap_retry" })), "not_an_exact_checkpoint_resume"], + ["the claimed failure had no checkpoint", claimed(resumeClaim({}, { checkpointExists: false })), "checkpoint_missing"], + ["the claimed failure recorded a replacement", claimed(resumeClaim({}, { successorRunId: "run-2" })), "replacement_recorded"], + ])("refuses a reconciler-claimed resume when %s", (_name, evidence, reason) => { + expect(decideScheduledResumeContinuation(evidence)).toEqual({ allowed: false, reason }); + }); }); const support = await getEmbeddedPostgresTestSupport(); diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.ts b/server/src/services/native-runtime/scheduled-resume-continuation.ts index 4f315655909..215e540754a 100644 --- a/server/src/services/native-runtime/scheduled-resume-continuation.ts +++ b/server/src/services/native-runtime/scheduled-resume-continuation.ts @@ -2,6 +2,28 @@ 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 const SAME_RUN_RESUME_DISPOSITION = "same_run_resume"; + +export function sameRunResumeClaimEntry(input: { + at: Date; controllerBootId: string; providerAttempt: number; failureDetail: Record | null | undefined; +}) { + const detail = input.failureDetail ?? {}; + return { + at: input.at.toISOString(), + disposition: SAME_RUN_RESUME_DISPOSITION, + controllerBootId: input.controllerBootId, + providerAttempt: input.providerAttempt, + evidence: { + recoveryMode: detail.recoveryMode ?? null, + recoverable: detail.recoverable ?? null, + providerSessionEstablished: detail.providerSessionEstablished ?? null, + checkpointExists: detail.checkpointExists ?? null, + successorRunId: detail.successorRunId ?? null, + replacementDenied: detail.replacementDenied ?? null, + }, + }; +} + export type ScheduledResumeContinuationRefusal = | "restart_recovery_claim" | "remote_target" @@ -19,6 +41,8 @@ export interface ScheduledResumeContinuationEvidence { targetKind: "local" | "remote"; coordinatorPhase: string | null; failureDetail: Record | null | undefined; + resumeClaim?: Record | null; + controllerBootId?: string; admittedAttempt: number; localStop: { processPid: number } | null; } @@ -33,8 +57,10 @@ export function decideScheduledResumeContinuation( 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 ?? {}; + const claimed = evidence.coordinatorPhase === "observed" && evidence.resumeClaim?.disposition === SAME_RUN_RESUME_DISPOSITION + && typeof evidence.controllerBootId === "string" && evidence.resumeClaim.controllerBootId === evidence.controllerBootId; + if (evidence.coordinatorPhase !== "retryable_failure" && !claimed) return refuse("not_a_scheduled_same_run_resume"); + const detail = (claimed ? evidence.resumeClaim?.evidence as Record | undefined : 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");