diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 1812b0839ae..1f0004da3a9 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3029,6 +3029,24 @@ ends with `native_provider_state_incompatible`, and Retry or the next message continues in a fresh provider session in the same workspace. A session with work in flight keeps the unclassified behavior. +A follow-up after a governed wait (`allow_replacement_after_governed_wait`) +normally requires the same provider session. For this one refusal it continues +on its own in a fresh provider session in the same run, without a Retry, when +all of these hold: the driver reports the Runner's typed missing-state refusal +for a session with nothing in flight; the reconciled checkpoint is the untouched +rebind of the previous run (no cursor progress, turn, runtime request, proposed +result, terminal, or governed wait of the follow-up run, and the at-most-once +recovery marker present); and the control plane proves from durable rows, +across every runner instance, that the follow-up run has no provider event, +source-sequenced runner event, tool invocation, or accepted result. The follow-up +has not started a turn, so a fresh session cannot repeat provider work or a tool +side effect; only the earlier conversation context is lost. The fresh session +gets the full task prompt, the interaction responses, the stored fresh-session +handoff, and queued input, as any replacement does, and the run log records the +existing provider session continuity break with the reason. Any other +incompatible state, a control plane without that proof, or a checkpoint saved +without those fields keeps `native_provider_state_incompatible` and its Retry. + Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A crash restart (no drain record) consumes an attempt only when the attempt it interrupted spent provider work without progress. This is one rule over recorded outcomes, not over intents. The attempt counts as started only when the provider accepted a turn: a `turn.accepted` event or the provider's own `turn.started`. A startup still marked `beforeSpawn`, a runner that spawned but was never admitted, and a `turn.start` the provider did not accept are not charged. Progress means that an accepted turn then completed work: a tool execution or a completed item other than a usage report. An accepted turn with only usage, a failure, or no events is charged, so a resume that fails each time still uses up the budget. The first restart judges the original execution. Each later restart claim records the run's event-log position, and the next claim judges only the attempt recorded after it; a normal retry in between starts a new attempt (at its `native.process_start_requested`), so earlier work does not count for it. A reattachment adopts a provider that already started, so for it only progress counts. A resume that launches a new runner and that the controller died before starting used no provider work, so the next claim replaces it and the attempt ends at the attempt that resume was to run as. Uncharged restarts stop after 12 controller generations, so a crash loop that never starts the provider, or that makes progress and then crashes the controller each time, is still bounded. Legacy runs follow the same rule: a preparation interrupted before provider work started (`providerWorkStarted: false`) is retried in its own `interrupted_preparation` lane, which never spends the failure retry budget and is bounded at 12 retries. This holds for a hard kill as well as a graceful drain: once the killed controller's recorded process is proven gone, restart recovery fails the run as `process_lost` with the bootstrap certificate and schedules it in that lane. A run saved by an older release without a controller process stamp cannot be proven and stays in the charged lane. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. A due failed remote retry can retain its provider PID after heartbeat completion diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index dfaee9efff3..e6623853776 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -167,6 +167,7 @@ export class HarnessDriverBackend implements NativeSessionBackend { recovered: false, reason: recovered.reason, ...(recovered.failureClass ? { failureClass: recovered.failureClass } : {}), + ...(recovered.resumeStateUnavailable ? { resumeStateUnavailable: true as const } : {}), }; } try { diff --git a/packages/paperclip-runner/src/backends/native-backend-factory.test.ts b/packages/paperclip-runner/src/backends/native-backend-factory.test.ts index 8ccefae294d..6886e845d0a 100644 --- a/packages/paperclip-runner/src/backends/native-backend-factory.test.ts +++ b/packages/paperclip-runner/src/backends/native-backend-factory.test.ts @@ -488,6 +488,7 @@ describe("native backend factory", () => { recovered: false, reason: expect.stringContaining("PRP provider resume state is unavailable"), failureClass: PROVIDER_STATE_INCOMPATIBLE, + resumeStateUnavailable: true, }); }); @@ -499,6 +500,7 @@ describe("native backend factory", () => { const recovered = await backend.recoverSession!({ ...snapshot, ...inFlight } as never, { signal: new AbortController().signal }); expect(recovered).toMatchObject({ recovered: false, reason: expect.stringContaining("PRP provider resume state is unavailable") }); expect(recovered).not.toHaveProperty("failureClass"); + expect(recovered).not.toHaveProperty("resumeStateUnavailable"); }); }); diff --git a/packages/paperclip-runner/src/contracts/control-plane-port.ts b/packages/paperclip-runner/src/contracts/control-plane-port.ts index b6e3aa685ea..6b91bfd4492 100644 --- a/packages/paperclip-runner/src/contracts/control-plane-port.ts +++ b/packages/paperclip-runner/src/contracts/control-plane-port.ts @@ -99,4 +99,11 @@ export interface ControlPlanePort { result: NativeRunResult | CompleteControlPlaneRunInput, options?: FinalizeControlPlaneOperationOptions, ): Promise; + /** + * Durable proof, across every runner instance, that this run has recorded no + * provider event, tool invocation, or accepted result. True only when the + * control plane can prove it; a port without this proof keeps every + * recovery refusal unchanged. + */ + provesNoRunProviderWork?(options: FinalizeControlPlaneOperationOptions): Promise; } diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index b8903443c8c..7a9396eac64 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -557,6 +557,10 @@ export interface HarnessSessionRecoveryResult { reason?: string; /** Closed runner class of a deterministic recovery failure. */ failureClass?: RunnerRecoveryFailureClass; + /** Set only with PROVIDER_STATE_INCOMPATIBLE: the Runner proved it holds no + * saved state for this session (NativeRunnerResumeStateUnavailableError) and + * the checkpoint had no turn or runtime request in flight. */ + resumeStateUnavailable?: true; } /** Reads the closed class of an authenticated runner command failure, or of a diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 2931aa8b1e5..b8e9d775fec 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -86,6 +86,7 @@ export interface NativeSessionRecoveryResult { session?: NativeSession; reason?: string; failureClass?: HarnessSessionRecoveryResult["failureClass"]; + resumeStateUnavailable?: HarnessSessionRecoveryResult["resumeStateUnavailable"]; } /** Runner-owned cause: the process was restored, but its active turn was lost. */ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index 834fdf100e8..9b78d7403ce 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -773,13 +773,15 @@ export class CodexAppServerDriver implements HarnessDriver { // resume it here. With nothing in flight, the runtime's recovery policy // decides, as for a moved workspace root. An in-flight session keeps the // unclassified failure. - const failureClass = runnerRecoveryFailureClass(error) - ?? (isRunnerResumeStateUnavailable(error) && !persistedSessionInFlight(snapshot) - ? PROVIDER_STATE_INCOMPATIBLE : undefined); + const classified = runnerRecoveryFailureClass(error); + const resumeStateUnavailable = !classified && isRunnerResumeStateUnavailable(error) + && !persistedSessionInFlight(snapshot); + const failureClass = classified ?? (resumeStateUnavailable ? PROVIDER_STATE_INCOMPATIBLE : undefined); return { recovered: false, reason: redactCodexDiagnostic(String(error)), ...(failureClass ? { failureClass } : {}), + ...(resumeStateUnavailable ? { resumeStateUnavailable: true as const } : {}), }; } finally { cancellation.detach(); diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 0f30e117452..ba31e5da9b8 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -17,6 +17,7 @@ import type { NativeRunIdentity } from "./contracts/types.js"; import type { NativeSession, NativeSessionBackend, + NativeSessionRecoveryResult, PersistedNativeSession, } from "./contracts/native-session-backend.js"; import { @@ -6664,6 +6665,171 @@ describe("executeNativeSession recovery", () => { } }); + // Staging class P after #20757: a follow-up after a governed (approval) wait + // rebinds the previous run's session, but the new sandbox holds no saved + // Runner state for it. The follow-up has not started a turn, so a fresh + // provider session cannot repeat provider work or a tool side effect; only + // the earlier conversation context is lost. The run continues in a fresh + // session with the full task, the answer and the stored handoff. + describe("a governed-wait follow-up whose saved provider state is gone", () => { + const approval = { interactionId: "interaction-approval", kind: "request_confirmation" as const, + response: { outcome: "accepted", note: "APPROVED_BY_BOARD" } }; + const followUpInput = { + ...input, + task: { ...input.task, prompt: "FULL_TASK_CONTEXT\nAPPROVAL_FOLLOW_UP" }, + interactionResponses: [approval], + }; + const reboundCheckpoint = (): PersistedNativeSession => ({ + backendKind: "mock", driverKind: "codex_app_server", sessionId: "driver-lost", identity, + providerSessionId: "provider-lost", providerRecoveryPolicy: "allow_replacement_after_governed_wait", + cursor: null, semanticResult: null, terminal: null, activeTurnId: null, terminalTurns: [], + pendingRuntimeRequests: [], dispositionOnlyRecoveryConsumed: false, dispositionOnlyRecoveryTurnId: null, + lineage: [], + }); + const stateUnavailable: NativeSessionRecoveryResult = { + recovered: false, failureClass: PROVIDER_STATE_INCOMPATIBLE, resumeStateUnavailable: true, + reason: "Error: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable", + }; + function harness(options: { + checkpoint?: PersistedNativeSession; + recovery?: Awaited>>; + proof?: boolean | null | "rejects" | "hangs"; + replay?: PrpEvent[]; + } = {}) { + const startTurn = vi.fn(async () => ({ turnId: "turn-fresh" })); + const replacementSnapshot: PersistedNativeSession = { + ...reboundCheckpoint(), sessionId: "driver-fresh", providerSessionId: "provider-fresh", + }; + const replacementSession: NativeSession = { + identity: () => identity, + capabilities: async () => ({ resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }), + async *events() { yield runnerEvent(1, "turn.completed"); }, + startTurn, + result: async () => ({ result, terminal, turnId: "turn-fresh" }), + snapshot: async () => structuredClone(replacementSnapshot), + close: async () => {}, + }; + const openReplacementSession = vi.fn(async () => replacementSession); + const openSession = vi.fn(async () => { throw new Error("replacement seam must be used"); }); + const recoverSession = vi.fn(async () => options.recovery ?? stateUnavailable); + const backend: NativeSessionBackend = { + descriptor: async () => ({ kind: "mock", name: "governed-follow-up", version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true } }), + openSession, openReplacementSession, recoverSession, + }; + const events: PrpEvent[] = []; + const provesNoRunProviderWork = vi.fn(async ({ signal }: { signal: AbortSignal }) => { + if (options.proof === "rejects") throw new Error("database unavailable"); + if (options.proof === "hangs") return await new Promise((_resolve, reject) => + signal.addEventListener("abort", () => reject(signal.reason), { once: true })); + return options.proof === true; + }); + const openRun = vi.fn(async () => undefined); + const replay = options.replay ?? []; + const controlPlane: ControlPlanePort = { + openRun, + loadSessionCheckpoint: async () => structuredClone(options.checkpoint ?? reboundCheckpoint()), + checkpointSession: async () => {}, + async appendEvent(event) { + events.push(structuredClone(event as PrpEvent)); + return { cursor: events.length, highestContiguousSourceSeq: highestContiguous(events), disposition: "committed" }; + }, + replayEvents: async ({ afterSourceSeq }) => { + const page = replay.filter(event => event.sourceSeq > afterSourceSeq); + return { events: page, highestContiguousSourceSeq: highestContiguous(page) }; + }, + completeRun: async () => {}, + ...(options.proof === null ? {} : { provesNoRunProviderWork }), + }; + const onContinuityBreak = vi.fn(async () => undefined); + const getFreshSessionHandoff = vi.fn(async () => "FRESH_HANDOFF: prior conversation summary"); + const run = () => executeNativeSession({ + input: followUpInput, backend, controlPlane, getFreshSessionHandoff, onContinuityBreak, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + }); + return { run, startTurn, openReplacementSession, openSession, recoverSession, provesNoRunProviderWork, + openRun, onContinuityBreak, getFreshSessionHandoff }; + } + + it("continues in a fresh provider session with the task, the answer and the stored handoff", async () => { + const h = harness({ proof: true }); + await expect(h.run()).resolves.toMatchObject({ providerSessionId: "provider-fresh" }); + expect(h.recoverSession).toHaveBeenCalledOnce(); + expect(h.provesNoRunProviderWork).toHaveBeenCalledOnce(); + expect(h.openReplacementSession).toHaveBeenCalledOnce(); + expect(h.openSession).not.toHaveBeenCalled(); + expect(h.startTurn).toHaveBeenCalledOnce(); + const envelope = JSON.parse(h.startTurn.mock.calls[0]![0].message.text) as { + task: { prompt: string }; interactionResponses: unknown[]; + }; + expect(envelope.task.prompt).toBe("FRESH_HANDOFF: prior conversation summary\n\nFULL_TASK_CONTEXT\nAPPROVAL_FOLLOW_UP"); + expect(envelope.interactionResponses).toEqual([approval]); + expect(h.startTurn.mock.calls[0]![0]).not.toHaveProperty("continuation"); + // The user sees that the earlier conversation context was reset. + expect(h.onContinuityBreak).toHaveBeenCalledWith({ + reason: expect.stringContaining("saved provider state is unavailable, so the follow-up continues in a fresh provider session without the earlier conversation context"), + previousDriverSessionId: "driver-lost", previousProviderSessionId: "provider-lost", + replacementDriverSessionId: "driver-fresh", replacementProviderSessionId: "provider-fresh", + }); + }); + + it.each([ + ["the control plane cannot prove the run started no provider work", { proof: false }], + ["the control plane predates the provider-work proof", { proof: null }], + ["the provider-work proof fails", { proof: "rejects" }], + ["the saved state is incompatible for another reason", { proof: true, recovery: { + recovered: false as const, failureClass: PROVIDER_STATE_INCOMPATIBLE, reason: "saved launch profile changed" } }], + ["the checkpoint requires the same session", { proof: true, + checkpoint: { ...reboundCheckpoint(), providerRecoveryPolicy: "same_session_only" as const } }], + ["this run already completed a turn", { proof: true, + checkpoint: { ...reboundCheckpoint(), cursor: "2", terminalTurns: [{ turnId: "turn-own", fingerprint: "completed" }] } }], + ["this run already advanced the durable cursor", { proof: true, checkpoint: { ...reboundCheckpoint(), cursor: "3" } }], + ["this run already recorded a provider event", { proof: true, replay: [runnerEvent(1, "turn.started")] }], + ["this run already proposed a result", { proof: true, checkpoint: { ...reboundCheckpoint(), semanticResult: result } }], + ["the checkpoint was saved without the at-most-once recovery marker", { proof: true, + checkpoint: (() => { const { dispositionOnlyRecoveryConsumed: _omitted, ...older } = reboundCheckpoint(); return older; })() }], + ] as const)("keeps the named refusal and a working Retry when %s", async (_label, options) => { + const h = harness(options as Parameters[0]); + const failure = await h.run().catch((error: unknown) => error); + expect(failure).toBeInstanceOf(NativeProviderStateIncompatibleError); + expect((failure as NativeProviderStateIncompatibleError).code).toBe("native_provider_state_incompatible"); + expect(h.openReplacementSession).not.toHaveBeenCalled(); + expect(h.openSession).not.toHaveBeenCalled(); + expect(h.openRun).not.toHaveBeenCalled(); + expect(h.startTurn).not.toHaveBeenCalled(); + }); + + it("never replaces a session whose checkpoint records this run's own governed wait", async () => { + const h = harness({ proof: true, checkpoint: { ...reboundCheckpoint(), + governedWait: { sourceEvent: runnerEvent(1, "turn.started"), result: yieldedResult } } }); + await expect(h.run()).rejects.toThrow(); + expect(h.openReplacementSession).not.toHaveBeenCalled(); + expect(h.startTurn).not.toHaveBeenCalled(); + }); + + it("keeps the named refusal when the provider-work proof does not answer in time", async () => { + vi.useFakeTimers(); + try { + const h = harness({ proof: "hangs" }); + const failure = h.run().catch((error: unknown) => error); + await vi.advanceTimersByTimeAsync(30_001); + expect(await failure).toBeInstanceOf(NativeProviderStateIncompatibleError); + expect(h.provesNoRunProviderWork).toHaveBeenCalledOnce(); + expect(h.openReplacementSession).not.toHaveBeenCalled(); + } finally { + vi.useRealTimers(); + } + }); + + it("stops by name when the follow-up's saved session still owns a runtime request", async () => { + const h = harness({ proof: true, recovery: { recovered: false as const, failureClass: PROVIDER_SIDE_EFFECT_PENDING, + reason: "native_runner_source_runtime_reconciliation_required" } }); + await expect(h.run()).rejects.toBeInstanceOf(NativeProviderSideEffectPendingError); + expect(h.provesNoRunProviderWork).not.toHaveBeenCalled(); + expect(h.openReplacementSession).not.toHaveBeenCalled(); + }); + }); + // A remote workspace epoch was abandoned, so the run was assigned a new // sibling root while its checkpoint still names the old one. A checkpoint // that requires the same session (an unfinished goal) ends with the diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 7b9d0971362..1d480fb524f 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -18,7 +18,7 @@ import type { NativeSession, NativeSessionBackend, } from "./contracts/native-session-backend.js"; -import type { PersistedNativeSession } from "./contracts/native-session-backend.js"; +import type { NativeSessionRecoveryResult, PersistedNativeSession } from "./contracts/native-session-backend.js"; import { PROVIDER_ATTACH_REFUSED, PROVIDER_SIDE_EFFECT_PENDING, @@ -1917,6 +1917,61 @@ function checkpointCursor(cursor: string | null | undefined): number { return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : 0; } +/** + * A follow-up after a governed wait normally requires the same provider + * session. When the Runner proved the saved session state is gone, and nothing + * proves this run started provider work of its own, a fresh session cannot + * repeat a turn or a tool side effect: only the earlier conversation context is + * lost. Every other case keeps the named refusal. The reconciled checkpoint must + * be the untouched rebind of the previous run (no cursor progress, turn, + * runtime request, result, terminal or governed wait of this run), and the + * control plane must prove durably, across runner instances, that the run has + * no provider event, tool invocation or accepted result. A control plane + * without that proof, or a checkpoint missing those fields, keeps the refusal. + */ +const GOVERNED_WAIT_PROVIDER_WORK_PROOF_TIMEOUT_MS = 30_000; + +async function governedWaitFollowUpMayStartFresh(input: { + input: NativeExecutionInput; + checkpoint: PersistedNativeSession; + recovery: NativeSessionRecoveryResult; + restartInterruption: boolean; + controlPlane: ControlPlanePort; + timeoutMs: number; +}): Promise { + const { checkpoint, recovery } = input; + if ( + checkpoint.providerRecoveryPolicy !== "allow_replacement_after_governed_wait" || + ("homeMode" in input.input && input.input.homeMode === "preconfigured") || + recovery.failureClass !== PROVIDER_STATE_INCOMPATIBLE || + recovery.resumeStateUnavailable !== true || + input.restartInterruption || + checkpoint.identity.runId !== input.input.binding.runId || + !(checkpoint.cursor === null || checkpoint.cursor === "0") || + checkpoint.activeTurnId !== null || + !Array.isArray(checkpoint.terminalTurns) || checkpoint.terminalTurns.length > 0 || + !Array.isArray(checkpoint.pendingRuntimeRequests) || checkpoint.pendingRuntimeRequests.length > 0 || + checkpoint.semanticResult !== null || + checkpoint.terminal !== null || + checkpoint.governedWait !== undefined || + checkpoint.dispositionOnlyRecoveryConsumed !== false || + !input.controlPlane.provesNoRunProviderWork + ) return false; + const controlPlane = input.controlPlane; + // A proof that fails or expires is no proof: keep the named refusal and its + // Retry rather than surfacing a generic interruption. + const timeoutMs = Math.min(input.timeoutMs, GOVERNED_WAIT_PROVIDER_WORK_PROOF_TIMEOUT_MS); + try { + return await runAbortableOperationWithin({ + timeoutMs, + timeoutMessage: `native session provider-work proof timed out after ${timeoutMs}ms`, + operation: (signal) => controlPlane.provesNoRunProviderWork!({ signal }), + }) === true; + } catch { + return false; + } +} + async function reconcileRecoveryCheckpoint(input: { controlPlane: ControlPlanePort; checkpoint: PersistedNativeSession; @@ -2520,7 +2575,15 @@ export async function executeNativeSession( if (recovery.failureClass === PROVIDER_ATTACH_REFUSED) { throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown"); } - if (!replacementAllowed) { + const governedWaitFollowUpLostState = !replacementAllowed && await governedWaitFollowUpMayStartFresh({ + input, + checkpoint: providerRecoveryCheckpoint, + recovery, + restartInterruption: Boolean(checkpointedInterruption), + controlPlane: options.controlPlane, + timeoutMs: recoveryTimeoutMs, + }); + if (!replacementAllowed && !governedWaitFollowUpLostState) { if (recovery.failureClass === PROVIDER_STATE_INCOMPATIBLE) { throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown"); } @@ -2529,7 +2592,9 @@ export async function executeNativeSession( ); } continuityBreak = { - reason: recovery.reason ?? "provider session is no longer recoverable", + reason: governedWaitFollowUpLostState + ? `saved provider state is unavailable, so the follow-up continues in a fresh provider session without the earlier conversation context (${recovery.reason ?? "unknown"})` + : recovery.reason ?? "provider session is no longer recoverable", previousDriverSessionId: providerRecoveryCheckpoint.sessionId, previousProviderSessionId: providerRecoveryCheckpoint.providerSessionId ?? null, diff --git a/server/src/services/native-runtime/native-governed-wait-lost-state-continuation.integration.test.ts b/server/src/services/native-runtime/native-governed-wait-lost-state-continuation.integration.test.ts new file mode 100644 index 00000000000..932ba718e59 --- /dev/null +++ b/server/src/services/native-runtime/native-governed-wait-lost-state-continuation.integration.test.ts @@ -0,0 +1,175 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { eq, sql } from "drizzle-orm"; +import { afterEach, expect, it, vi } from "vitest"; +import { completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, toolInvocations } from "@paperclipai/db"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; +import { DurablePrpControlPlane, parseNativeExecutionInput, type NativeSessionBackend, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { createNativeControllerRecovery } from "./native-controller-recovery.js"; +import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import { executePaperclipNativeSession, nativeControllerMigrationDescriptor } from "./native-session-executor.js"; +import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; +import { nativeToolContractFingerprintForTarget, rebindNativeSessionCheckpoint } from "./native-session-resume.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; + +vi.mock("./paperclip-runner-tool-authority.js", () => ({ PaperclipRunnerToolAuthority: class { + readonly binding = {}; + async definitions() { return []; } + close() {} +} })); + +// Staging class P after #20757: run A yielded for an approval, its harness +// backup was never saved, and its sandbox was destroyed. The follow-up run B +// rebinds A's session in a new sandbox whose Runner has no saved state for it. +// B has not started a turn, so a fresh provider session cannot repeat provider +// work or a tool side effect. B continues in that fresh session on its own, +// with no Retry. Any durable sign that B already did provider work keeps the +// permanent named refusal. +describeEmbeddedPostgres("governed-wait follow-up whose saved provider state is gone", () => { + const environment = nativeRestartFixture(); + const homes: string[] = []; + afterEach(async () => { vi.unstubAllEnvs(); for (const home of homes.splice(0)) await rm(home, { recursive: true, force: true }); }); + + async function followUpRun() { + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "61".repeat(32)); + const f = await environment.seedRun(randomUUID()); + const { companyId, agentId, runtimeRoot } = environment.current; + const home = await mkdtemp(join(tmpdir(), "native-governed-lost-state-")); homes.push(home); + vi.stubEnv("PAPERCLIP_HOME", home); + const [sourceRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [contract] = await f.db.select().from(completionContracts).where(eq(completionContracts.id, sourceRun!.completionContractId!)); + const execution = parseNativeExecutionInput({ + schema: "paperclip.native-execution-input.v5", executionMode: "default", planningContext: null, + provider: { kind: "codex", model: null, approvalPolicy: "never" }, + binding: { companyId, agentId, issueId: f.issueId, runId: f.runId, executionWorkspaceId: randomUUID() }, + task: { identifier: "GWL-1", title: "Approved work", description: null, prompt: "Continue after the approval.", workMode: "standard" }, + workspace: { cwd: runtimeRoot, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: f.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: contract!.schemaVersion, contract: contract!.contractJson }, + runtimeContext: nativeRuntimeContextFixture(), credentialBindings: [], interactionResponses: [], + }); + const yielded = { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", summary: "Waiting for approval", + completionClaim: { contractRevision: "1", objectiveSatisfied: false, criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [] }], remainingWork: [] }, + evidence: [], verification: [], attentionRequests: [], artifacts: [], + continuation: { kind: "response_wake", summary: "Continue after the answer", idempotencyKey: "fixture-approval" } }; + const snapshot: PersistedNativeSession = { backendKind: "runner", driverKind: "codex_app_server", sessionId: "lost-driver-session", + providerSessionId: "lost-provider-session", identity: { companyId, agentId, issueId: f.issueId, runId: f.runId, sessionId: f.native.normalizedSessionId }, + cursor: "0", activeTurnId: null, pendingRuntimeRequests: [], terminalTurns: [], lineage: [], semanticResult: yielded as never, goal: null }; + const bootId = randomUUID(), leaseOwner = randomUUID(); + await f.db.update(nativeRunFinalizations).set({ controllerBootId: bootId, controllerGeneration: 1, controllerPid: 2_147_483_000, + controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const core = new DurablePrpControlPlane({ stateDirectory: join(runtimeRoot, randomUUID()), identity: { runId: f.runId, normalizedSessionId: f.native.normalizedSessionId, + runnerInstanceId: f.native.runnerInstanceId, environmentLeaseId: f.native.environmentLeaseId, turnId: f.native.turnId, itemId: f.native.itemId }, + expectedRunnerVersion: "governed-lost-state-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + for (const type of ["run.prepare", "session.open", "turn.start", "runner.suspend"]) Object.assign(core.queueCommand(type), { status: "completed", result: { status: "completed" } }); + core.store.state.committedEvents.push({ sourceSeq: 1, sourceEventId: "fixture-terminal", eventType: "run.terminal", priority: 0, envelope: { ...core.store.state.identity }, deliveryCount: 1, logicalEffectCount: 1 }); + core.store.state.ackedSourceSeq = 1; + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(f.db, { companyId, issueId: f.issueId, runId: f.runId, authorityRunId: f.runId, bootId, generation: 1, leaseOwner })); + await core.flushCheckpoint(); await core.stop(); + const [result] = await f.db.insert(nativeRunResults).values({ companyId, issueId: f.issueId, runId: f.runId, completionContractId: sourceRun!.completionContractId!, + serverFingerprint: randomUUID(), schemaStatus: "accepted", resultJson: yielded, canonicalSha256: "a".repeat(64) }).returning(); + const fingerprint = nativeToolContractFingerprintForTarget("remote", "codex"); + const descriptor = nativeControllerMigrationDescriptor(execution, fingerprint); + const sourceProfile = { ...f.preparedProfile, nativeExecutionInput: execution, sessionCheckpoint: snapshot, nativeToolContractFingerprint: fingerprint, + nativeControllerRecovery: createNativeControllerRecovery(snapshot, { configDigest: descriptor.configDigest, processConfigDigest: null }, + { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId }) }; + await f.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: sourceProfile }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result!.id, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + + const runId = randomUUID(); + const [run] = await f.db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: f.issueId, issueId: f.issueId, status: "running", + runtimeMode: "native", nativeSessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId, + completionContractId: sourceRun!.completionContractId, completionContractSha256: sourceRun!.completionContractSha256 }).returning(); + const native = await prepareNativeHeartbeatRun({ db: f.db, run: run!, issue: { id: f.issueId, title: "Approved work", description: null }, environmentLeaseId: randomUUID() }); + const current = { ...execution, binding: { ...execution.binding, runId } }; + const rebound = rebindNativeSessionCheckpoint({ previousRun: { id: f.runId, companyId, agentId, nativeSessionId: f.native.normalizedSessionId, runnerProfileJson: sourceProfile }, + currentExecution: current, executionTargetKind: "remote" }); + expect(rebound?.providerRecoveryPolicy).toBe("allow_replacement_after_governed_wait"); + const [prepared] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...prepared!.runnerProfileJson, nativeExecutionInput: current, sessionCheckpoint: rebound } }).where(eq(heartbeatRuns.id, runId)); + await f.db.insert(nativeRunFinalizations).values({ runId, companyId, issueId: f.issueId, phase: "observed" }); + await f.db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, f.issueId)); + + const recover = vi.fn(async () => ({ recovered: false as const, failureClass: "provider_state_incompatible" as const, resumeStateUnavailable: true as const, + reason: "Error: NativeRunnerResumeStateUnavailableError: PRP provider resume state is unavailable" })); + const replacementObserved = new Error("fixture fresh provider session opened"); + const fresh = vi.fn(async () => { throw replacementObserved; }); + const backend: NativeSessionBackend = { + descriptor: async () => ({ kind: "runner", name: "controlled provider boundary", version: "fixture", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }), + recoverSession: recover, openSession: fresh, openReplacementSession: fresh, + }; + const execute = () => executePaperclipNativeSession({ db: f.db, execution: current, runnerInstanceId: native.runnerInstanceId, useRunnerd: true, backend, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: runtimeRoot, environmentId: randomUUID(), leaseId: randomUUID(), + providerKey: "fixture", reusableLeaseConfigured: true, runner: { execute: async () => { throw new Error("unexpected remote process execution"); } }, + sandboxLeaseAcquisition: { outcome: "created", providerLeaseId: "new-fixture-sandbox" } } as never }).catch((error: unknown) => error); + return { f, companyId, agentId, runId, recover, fresh, replacementObserved, execute }; + } + + const proofFor = (h: Awaited>) => new PaperclipControlPlanePort(h.f.db, { + companyId: h.companyId, issueId: h.f.issueId, runId: h.runId, agentId: h.agentId, sessionId: "proof-session", + completionContractId: "proof-contract", completionContractSha256: "proof-contract", sourceInstanceId: "proof-runner", + controlPlaneSourceInstanceId: "proof-control", + }).provesNoRunProviderWork({ signal: new AbortController().signal }); + + it("opens a fresh provider session for the follow-up without a Retry", async () => { + const h = await followUpRun(); + expect(await proofFor(h)).toBe(true); + const failure = await h.execute(); + expect(h.recover, String(failure)).toHaveBeenCalledOnce(); + // The fixture stops at the fresh session's boundary; on master the run + // ended with native_provider_state_incompatible before reaching it. + expect(failure).toBe(h.replacementObserved); + expect(h.fresh).toHaveBeenCalledOnce(); + }, 60_000); + + it.each([ + ["a provider event", "event"], + ["a tool invocation", "tool"], + ["an accepted result", "result"], + // A runner event of a non-provider type still carries a source identity: + // a runner wrote it for this run, so the run is not untouched. + ["a sourced runner event of a non-provider type", "sourced_event"], + // The foreign key normally forbids this. A row restored without its result + // must still deny the proof rather than look like an untouched run. + ["a finalized result whose result row is missing", "dangling_result"], + ] as const)("keeps the named refusal when the follow-up run already recorded %s", async (_label, evidence) => { + const h = await followUpRun(); + if (evidence === "event") await h.f.db.insert(heartbeatRunEvents).values({ companyId: h.companyId, runId: h.runId, agentId: h.agentId, + seq: 1_000, eventType: "turn.started", stream: "system", level: "info", message: "turn started", payload: {} }); + if (evidence === "tool") await h.f.db.insert(toolInvocations).values({ companyId: h.companyId, runId: h.runId, agentId: h.agentId, + toolName: "paperclip_get_issue", riskLevel: "read", status: "succeeded" } as never); + if (evidence === "result") { + const [run] = await h.f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, h.runId)); + await h.f.db.insert(nativeRunResults).values({ companyId: h.companyId, issueId: h.f.issueId, runId: h.runId, completionContractId: run!.completionContractId!, + serverFingerprint: randomUUID(), schemaStatus: "accepted", resultJson: {}, canonicalSha256: "b".repeat(64) }); + } + if (evidence === "sourced_event") await h.f.db.insert(heartbeatRunEvents).values({ companyId: h.companyId, runId: h.runId, agentId: h.agentId, + seq: 1_000, eventType: "native.diagnostic", stream: "system", level: "info", message: "runner diagnostic", payload: {}, + sourceInstanceId: "foreign-runner-instance", sourceEventId: "foreign-runner-instance:event:1", sourceSeq: 1 }); + if (evidence === "dangling_result") await h.f.db.transaction(async (tx) => { + await tx.execute(sql`set local session_replication_role = 'replica'`); + await tx.update(nativeRunFinalizations).set({ resultId: randomUUID() }).where(eq(nativeRunFinalizations.runId, h.runId)); + }); + if (evidence === "sourced_event" || evidence === "dangling_result") { + // The fixture is the only evidence: no tool invocation or result row. + expect(await h.f.db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, h.runId))).toEqual([]); + expect(await h.f.db.select().from(toolInvocations).where(eq(toolInvocations.runId, h.runId))).toEqual([]); + } + // The durable proof itself refuses each kind of evidence. + expect(await proofFor(h)).toBe(false); + const failure = await h.execute(); + expect(h.fresh).not.toHaveBeenCalled(); + if (evidence === "dangling_result") { + // Admission stops even earlier on a finalized result: still no fresh session. + expect(String(failure)).toContain("native_result_pending_finalization"); + return; + } + expect(h.recover, String(failure)).toHaveBeenCalledOnce(); + expect((failure as { code?: string }).code).toBe("native_provider_state_incompatible"); + }, 60_000); +}); diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index 39af5d919de..5b21bc744db 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -10,6 +10,7 @@ import { nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints, + toolInvocations, } from "@paperclipai/db"; import type { CompleteControlPlaneRunInput, @@ -36,6 +37,7 @@ import { eligibleNativeTaskSessionPublication, publishNativeTaskSessionInTransac type NativeTaskSessionPublicationContext, type NativeTaskSessionPublicationIssue } from "./native-task-session-publication.js"; import { lockNativeRunnerCheckpointOwner, nativeRunnerCheckpointOwnerLock, validateNativeRunnerCheckpointOwner, inspectNativeRunnerUpgradeHeader, type NativeRunnerCheckpointOwner } from "./native-runner-checkpoint.js"; import { crashPoint } from "@paperclipai/shared/crash-points"; +import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES } from "./native-retired-backup-source.js"; export interface PaperclipControlPlaneBinding { companyId: string; @@ -512,6 +514,34 @@ export class PaperclipControlPlanePort implements ControlPlanePort { return { events, highestContiguousSourceSeq: cursor }; } + /** + * Proves from durable rows, across every runner instance, that this run has + * no provider event, source-sequenced runner event, tool invocation, or + * accepted result. Any such row, or a read failure, denies the proof. + */ + async provesNoRunProviderWork(options: { signal: AbortSignal }): Promise { + options.signal.throwIfAborted(); + const { companyId, runId } = this.#binding; + const providerEvents = sql.join([...NATIVE_PROVIDER_DURABLE_EVENT_TYPES].map(value => sql`${value}`), sql`, `); + const result = await this.#db.execute(sql`select + exists (select 1 from ${heartbeatRunEvents} event where event.company_id = ${companyId} and event.run_id = ${runId} + and (event.source_event_id is not null or event.source_instance_id is not null + or event.event_type in (${providerEvents}) + or event.event_type ~ '^(turn|item|tool|semantic_tool|run[.]result)[.]')) as "providerEvent", + exists (select 1 from ${toolInvocations} invocation where invocation.company_id = ${companyId} + and invocation.run_id = ${runId}) as "toolInvocation", + exists (select 1 from ${nativeRunResults} result where result.company_id = ${companyId} + and result.run_id = ${runId}) as "acceptedResult", + exists (select 1 from ${nativeRunFinalizations} coordinator where coordinator.run_id = ${runId} + and coordinator.result_id is not null) as "finalizedResult"`); + options.signal.throwIfAborted(); + const rows = (Array.isArray(result) ? result : (result as { rows?: unknown[] }).rows ?? []) as Array>; + const row = rows[0]; + if (!row) return false; + return row.providerEvent === false && row.toolInvocation === false + && row.acceptedResult === false && row.finalizedResult === false; + } + async completeRun(value: NativeRunResult | CompleteControlPlaneRunInput, options?: { signal: AbortSignal }): Promise { this.#assertActive(options); if (this.#deferredEventObservers.size > 0 || this.#deferredObserverFailure) await this.#settleDeferredObservers(options);