diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 2057e8ee56b..3ac90be6d7d 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -118,6 +118,20 @@ export function nativeRestartInterruptedTurnId(snapshot: Pick): boolean { + const activeTurnId = snapshot.activeTurnId; + if (typeof activeTurnId !== "string" || !activeTurnId) return false; + const last = Array.isArray(snapshot.terminalTurns) ? snapshot.terminalTurns.at(-1) : null; + if (!last || typeof last.fingerprint !== "string" || !last.turnId || last.turnId === activeTurnId) return false; + try { + const fingerprint = JSON.parse(last.fingerprint); + return fingerprint?.terminalState === "failed" && fingerprint.result === null && + isNativeRestartInterruption(fingerprint.error); + } catch { return false; } +} + /** * Cancellation is a synchronous authority transition followed by passive * provider cleanup. Once `cancel` returns, the provider session must no longer diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 73bff682ae6..9d40f756c99 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -22,7 +22,7 @@ import { NativeSessionCloseUnrecoverableError, NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, - nativeRestartInterruptedTurnId, + nativeRestartContinuationInFlight, nativeRestartInterruptedTurnId, } from "./contracts/native-session-backend.js"; import { PROVIDER_ATTACH_REFUSED, PROVIDER_SIDE_EFFECT_PENDING, PROVIDER_STATE_INCOMPATIBLE } from "./contracts/harness-driver.js"; import type { @@ -642,6 +642,53 @@ describe("executeNativeSession recovery", () => { expect(events.filter(event => event.eventType === "turn.failed").map(event => event.turnId)).toEqual(["turn-unrecorded-continuation"]); }); + it("recovers a restart continuation whose checkpoint still carries the lost turn's failed terminal", async () => { + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind: "codex_app_server", sessionId: "thread-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-continuation", + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "failed", turnTerminalState: "failed", reportedWorkDisposition: "yielded" }, + terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }], + semanticResult: null, pendingRuntimeRequests: [], lineage: [], + dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-continuation", + }; + const startTurn = vi.fn(async () => ({ turnId: "turn-unexpected" })); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + yield { ...runnerEvent(1, "turn.completed", { status: "completed" }), turnId: "turn-continuation" }; + }, + startTurn, + async result() { return { result, terminal, turnId: "turn-continuation" }; }, + async snapshot() { return structuredClone(checkpoint); }, + async close() {}, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "runner", name: "restart", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { throw new Error("must preserve the provider session"); }, + async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event: PrpEvent) { return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + await expect(executeNativeSession({ + input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, + })).resolves.toMatchObject({ result, turnId: "turn-continuation", terminal: { runTerminalState: "succeeded" } }); + expect(startTurn).not.toHaveBeenCalled(); + expect(nativeRestartContinuationInFlight(checkpoint)).toBe(true); + expect(nativeRestartContinuationInFlight({ ...checkpoint, activeTurnId: "turn-recovery" })).toBe(false); + expect(nativeRestartContinuationInFlight({ ...checkpoint, terminalTurns: [{ turnId: "turn-recovery", + fingerprint: JSON.stringify({ terminalState: "failed", error: { code: "model_not_found", recoverable: false }, result: null }) }] })) + .toBe(false); + }); + it.each(["acpx_runtime", "opencode_server"] as const)("continues a restart-lost %s turn once in the same conversation", async (driverKind) => { const qualified = resolveQualifiedAcpxProfile("claude", "claude-sonnet-5"); const acpxInput: NativeExecutionInputV1 = (driverKind === "acpx_runtime" ? { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 180521404b6..2e8bcb37074 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -33,6 +33,7 @@ import { NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, isNativeRestartInterruption, + nativeRestartContinuationInFlight, nativeRestartInterruptedTurnId, } from "./contracts/native-session-backend.js"; import { @@ -2327,7 +2328,7 @@ export async function executeNativeSession( } | null = null; let reconciledRecoveryCheckpoint: PersistedNativeSession | null = null; const checkpointedInterruption = persistedSession && nativeRestartInterruptedTurnId(persistedSession); - if (checkpointedInterruption && persistedSession) { + if (persistedSession && (checkpointedInterruption || nativeRestartContinuationInFlight(persistedSession))) { // The turn failed because its process vanished; the conversation is still // usable. Keep its terminal fingerprint so replay cannot execute it twice. persistedSession = { ...persistedSession, terminal: null }; diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index a5c799ff64a..bf8033add40 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -47,7 +47,7 @@ import { logger } from "../../middleware/logger.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; -import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { nativeRestartContinuationInFlight, nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js"; @@ -1436,7 +1436,8 @@ export async function claimNativeRestartRecoveries(input: { checkpointFailed: (checkpointRecord.terminal as Record | undefined) ?.runTerminalState === "failed" && - nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null, + nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null && + !nativeRestartContinuationInFlight(checkpointRecord as unknown as PersistedNativeSession), providerAttempt: row.coordinator.attempt, resumeChargesAttempt: !resumeKeepsAttempt, }); diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 31ef5944755..5dad4687c4e 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2892,8 +2892,10 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); { attempt: 2, driverKind: "opencode_server" }, // A second restart that loses the first restart's continuation turn. { attempt: 2, driverKind: "codex_app_server", lostContinuation: true }, - ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind, lostContinuation }) => { - const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}${lostContinuation ? "-CONTINUATION" : ""}`); + { attempt: 2, driverKind: "codex_app_server", lostContinuation: true, staleTerminal: true }, + { attempt: 2, driverKind: "acpx_runtime", lostContinuation: true, staleTerminal: true, heldResult: true }, + ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind, lostContinuation, staleTerminal, heldResult }) => { + const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}${lostContinuation ? "-CONTINUATION" : ""}${staleTerminal ? "-STALE" : ""}${heldResult ? "-RESULT" : ""}`); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { driverKind, sessionId: fixture.native.normalizedSessionId, @@ -2908,7 +2910,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); }) }], // A restart during the first continuation: its turn is still the // checkpointed active turn and the consumed marker names it. - ...(lostContinuation ? { activeTurnId: "continuation-turn", terminal: null, + ...(heldResult ? { semanticResult: { schema: "paperclip.run_result.v1", summary: "Saved before the restart." } } : {}), + ...(lostContinuation ? { activeTurnId: "continuation-turn", ...(staleTerminal ? {} : { terminal: null }), dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "continuation-turn" } : {}), } } }).where(eq(heartbeatRuns.id, fixture.runId)); await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId)); @@ -2923,6 +2926,26 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); : { kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]); }); + it("keeps a run whose active turn follows a real provider failure blocked as permanently failed", async () => { + const fixture = await seedRun("ACTIVE-AFTER-PROVIDER-FAILURE"); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { + driverKind: "codex_app_server", + sessionId: fixture.native.normalizedSessionId, + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "failed-provider", + activeTurnId: "later-turn", + semanticResult: null, + terminal: { runTerminalState: "failed", turnTerminalState: "failed" }, + terminalTurns: [{ turnId: "failed-turn", fingerprint: JSON.stringify({ + terminalState: "failed", result: null, error: { code: "model_not_found", recoverable: false }, + }) }], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }, { eventType: "turn.accepted", turnId: "later-turn" }]); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, restartKind: "hot", runIds: [fixture.runId] })) + .toEqual([expect.objectContaining({ kind: "blocked", runId: fixture.runId, reason: "provider_checkpoint_permanently_failed" })]); + }); + // RC-22b: controller crashes (no drain record) that each interrupted a resume // which had already done work exhausted the three-attempt budget. async function seedInterruptedCheckpoint(label: string) { diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index 38108ec7a08..732c76e741d 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -193,6 +193,7 @@ export const NativeProviderStateIncompatibleError = runner.NativeProviderStateIn export const NativeProviderSideEffectPendingError = runner.NativeProviderSideEffectPendingError; export const NativeRunnerSourceReconciliationRequiredError = runner.NativeRunnerSourceReconciliationRequiredError; export const nativeRestartInterruptedTurnId = runner.nativeRestartInterruptedTurnId; +export const nativeRestartContinuationInFlight = runner.nativeRestartContinuationInFlight; export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTerminatedRemoteNativeSessionCleanup; export const completeTerminatedLocalNativeSessionCleanup = runner.completeTerminatedLocalNativeSessionCleanup;