diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts index abcf8b8638..6a066dd4d1 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import type { HarnessDriver, HarnessSession, PersistedHarnessSession } from "../contracts/harness-driver.js"; import type { PrpEvent, PrpStructuredRunResult, PrpTerminalState } from "../protocol/replay-contract.js"; @@ -445,6 +445,242 @@ describe("HarnessDriverBackend", () => { expect(session.identity()).toEqual({ ...originalIdentity, runId: "run-2" }); }); + it("restores a persisted terminal before the recovered stream is consumed", async () => { + const recoveryDriver: HarnessDriver = { + ...driver, + async recoverSession() { + return { recovered: true, session: new FakeHarnessSession() }; + }, + }; + const backend = new HarnessDriverBackend(recoveryDriver); + const terminal = { + schema: "paperclip.prp.terminal.v1" as const, + turnTerminalState: "completed" as const, + runTerminalState: "succeeded" as const, + reportedWorkDisposition: "done" as const, + }; + const recovery = await backend.recoverSession({ + backendKind: "runner", + driverKind: "fake", + sessionId: "driver-1", + providerSessionId: "provider-1", + identity: { + runId: "run-terminal-recovery", + sessionId: "session-1", + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + }, + semanticResult: result, + terminal, + activeTurnId: "turn-1", + terminalTurns: [ + { turnId: "turn-1", fingerprint: "terminal-fingerprint" }, + ], + }, { + signal: new AbortController().signal, + }); + + expect(recovery).toMatchObject({ recovered: true }); + await expect(recovery.session!.snapshot()).resolves.toMatchObject({ + semanticResult: result, + terminal, + }); + await expect(recovery.session!.result()).resolves.toEqual({ + result, + terminal, + turnId: "turn-1", + }); + }); + + it("reconstructs a missing top-level terminal from a completed semantic turn", async () => { + const recoveryDriver: HarnessDriver = { + ...driver, + async recoverSession() { + return { recovered: true, session: new FakeHarnessSession() }; + }, + }; + const backend = new HarnessDriverBackend(recoveryDriver); + const semanticFingerprint = canonicalTestJson(result); + const recovery = await backend.recoverSession({ + backendKind: "runner", + driverKind: "fake", + sessionId: "driver-1", + providerSessionId: "provider-1", + identity: { + runId: "run-terminal-inference", + sessionId: "session-1", + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + }, + semanticResult: result, + terminal: null, + activeTurnId: null, + terminalTurns: [ + { + turnId: "turn-1", + fingerprint: JSON.stringify({ + status: "completed", + semanticResult: semanticFingerprint, + }), + }, + ], + }, { + signal: new AbortController().signal, + }); + + expect(recovery).toMatchObject({ recovered: true }); + await expect(recovery.session!.result()).resolves.toEqual({ + result, + terminal: { + schema: "paperclip.prp.terminal.v1", + turnTerminalState: "completed", + runTerminalState: "succeeded", + reportedWorkDisposition: "done", + }, + turnId: "turn-1", + }); + }); + + it("does not pair a recovered semantic result with another turn's terminal", async () => { + let recoveredPersisted: PersistedHarnessSession | null = null; + class RecoveredHarnessSession extends FakeHarnessSession { + override async snapshot(): Promise { + if (recoveredPersisted === null) throw new Error("missing recovered snapshot"); + return structuredClone(recoveredPersisted); + } + } + const recoveryDriver: HarnessDriver = { + ...driver, + async recoverSession(snapshot) { + recoveredPersisted = snapshot; + return { recovered: true, session: new RecoveredHarnessSession() }; + }, + }; + const backend = new HarnessDriverBackend(recoveryDriver); + const semanticFingerprint = canonicalTestJson(result); + const recovery = await backend.recoverSession({ + backendKind: "runner", + driverKind: "fake", + sessionId: "driver-1", + providerSessionId: "provider-1", + identity: { + runId: "run-cross-turn-terminal", + sessionId: "session-1", + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + }, + semanticResult: result, + terminal: { + schema: "paperclip.prp.terminal.v1", + turnTerminalState: "cancelled", + runTerminalState: "cancelled", + reportedWorkDisposition: "yielded", + }, + activeTurnId: null, + terminalTurns: [ + { + turnId: "turn-with-result", + fingerprint: JSON.stringify({ + status: "completed", + semanticResult: semanticFingerprint, + }), + }, + { + turnId: "later-cancelled-turn", + fingerprint: JSON.stringify({ status: "cancelled" }), + }, + ], + }, { + signal: new AbortController().signal, + }); + + expect(recovery).toMatchObject({ recovered: true }); + await expect(recovery.session!.result()).resolves.toEqual({ + result, + terminal: { + schema: "paperclip.prp.terminal.v1", + turnTerminalState: "completed", + runTerminalState: "succeeded", + reportedWorkDisposition: "done", + }, + turnId: "turn-with-result", + }); + }); + + it("rejects oversized terminal history before inspecting the semantic result", async () => { + const recoverSession = vi.fn(async () => ({ + recovered: true, + session: new FakeHarnessSession(), + })); + const backend = new HarnessDriverBackend({ ...driver, recoverSession }); + const inaccessibleResult = new Proxy(result, { + ownKeys() { + throw new Error("semantic result must not be inspected"); + }, + }); + + await expect(backend.recoverSession({ + backendKind: "runner", + driverKind: "fake", + sessionId: "driver-1", + providerSessionId: "provider-1", + identity: { + runId: "run-oversized-terminals", + sessionId: "session-1", + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + }, + semanticResult: inaccessibleResult, + terminalTurns: Array.from({ length: 4_097 }, (_, index) => ({ + turnId: `turn-${index}`, + fingerprint: "terminal", + })), + }, { + signal: new AbortController().signal, + })).resolves.toEqual({ + recovered: false, + reason: "persisted harness terminal history exceeds its recovery limit", + }); + expect(recoverSession).not.toHaveBeenCalled(); + }); + + it("rejects an oversized semantic result before canonicalization", async () => { + const recoverSession = vi.fn(async () => ({ + recovered: true, + session: new FakeHarnessSession(), + })); + const backend = new HarnessDriverBackend({ ...driver, recoverSession }); + + await expect(backend.recoverSession({ + backendKind: "runner", + driverKind: "fake", + sessionId: "driver-1", + providerSessionId: "provider-1", + identity: { + runId: "run-oversized-semantic-result", + sessionId: "session-1", + companyId: "company-1", + issueId: "issue-1", + agentId: "agent-1", + }, + semanticResult: { + ...result, + summary: "x".repeat(8 * 1024 * 1024), + }, + terminalTurns: [], + }, { + signal: new AbortController().signal, + })).resolves.toEqual({ + recovered: false, + reason: "persisted harness semantic result exceeds its recovery limit", + }); + expect(recoverSession).not.toHaveBeenCalled(); + }); + it("delegates native runtime-request resolutions to the harness session", async () => { runtimeResolutions.length = 0; const backend = new HarnessDriverBackend(driver); @@ -580,3 +816,17 @@ describe("HarnessDriverBackend", () => { await expect(iterator.next()).rejects.toThrow("provider stopped after cancellation"); }); }); + +function canonicalTestJson(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map(canonicalTestJson).join(",")}]`; + } + if (typeof value === "object" && value !== null) { + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalTestJson(record[key])}`) + .join(",")}}`; + } + return JSON.stringify(value) ?? "undefined"; +} diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index a5164d6298..a53c768404 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -17,6 +17,13 @@ import type { PrpTerminalState, } from "../protocol/replay-contract.js"; +const MAX_RECOVERY_TERMINAL_TURNS = 4_096; +const MAX_RECOVERY_TERMINAL_BYTES = 8 * 1024 * 1024; +const MAX_RECOVERY_TERMINAL_FINGERPRINT_BYTES = 256 * 1024; +const MAX_RECOVERY_SEMANTIC_RESULT_BYTES = 8 * 1024 * 1024; +const MAX_RECOVERY_SEMANTIC_RESULT_NODES = 65_536; +const MAX_RECOVERY_SEMANTIC_RESULT_DEPTH = 128; + /** * Package-owned adapter from the concrete harness driver contract to the * normalized session boundary consumed by Paperclip. Provider mechanics stay @@ -70,11 +77,34 @@ export class HarnessDriverBackend implements NativeSessionBackend { if (this.#driver.recoverSession === undefined) { return { recovered: false, reason: "driver does not support recovery" }; } + const prevalidationFailure = recoveryPrevalidationFailure(snapshot); + if (prevalidationFailure !== null) { + return { recovered: false, reason: prevalidationFailure }; + } + const recoveredSemanticTurn = completedSemanticTurn(snapshot); + const semanticTurnId = + recoveredSemanticTurn?.turnId ?? + snapshot.activeTurnId ?? + snapshot.terminalTurns?.at(-1)?.turnId ?? + null; + const recoveredTerminal = + recoveredSemanticTurn && snapshot.semanticResult + ? { + schema: "paperclip.prp.terminal.v1" as const, + turnTerminalState: "completed" as const, + runTerminalState: "succeeded" as const, + reportedWorkDisposition: + snapshot.semanticResult.reportedWorkDisposition, + } + : snapshot.terminal ?? null; // `null` is a durable "no active turn" checkpoint. Only an omitted legacy // field may use the compatibility fallback; nullish coalescing here would - // otherwise turn the most recent terminal turn back into an active one. + // otherwise turn a settled semantic or terminal turn back into an active + // one on every subsequent recovery. const activeTurnId = snapshot.activeTurnId === undefined - ? snapshot.terminalTurns?.at(-1)?.turnId ?? null + ? recoveredSemanticTurn?.turnId + ?? snapshot.terminalTurns?.at(-1)?.turnId + ?? null : snapshot.activeTurnId; const persisted: PersistedHarnessSession = { driverKind: snapshot.driverKind ?? snapshot.backendKind, @@ -96,7 +126,7 @@ export class HarnessDriverBackend implements NativeSessionBackend { semanticResult: { result: snapshot.semanticResult, fingerprint: canonicalJson(snapshot.semanticResult), - turnId: activeTurnId ?? snapshot.terminalTurns?.at(-1)?.turnId ?? "recovered", + turnId: semanticTurnId ?? "recovered", }, }), terminalTurns: snapshot.terminalTurns ?? [], @@ -135,12 +165,190 @@ export class HarnessDriverBackend implements NativeSessionBackend { session: new HarnessNativeSession( { identity: snapshot.identity }, recovered.session, - snapshot.terminal, + recoveredTerminal, ), }; } } +function completedSemanticTurn( + snapshot: PersistedNativeSession, +): { turnId: string; fingerprint: string } | null { + if (snapshot.semanticResult === undefined || snapshot.semanticResult === null) { + return null; + } + const semanticFingerprint = canonicalJson(snapshot.semanticResult); + const terminalTurns = snapshot.terminalTurns ?? []; + for (let index = terminalTurns.length - 1; index >= 0; index -= 1) { + const terminal = terminalTurns[index]!; + try { + const value: unknown = JSON.parse(terminal.fingerprint); + const record = plainRecord(value); + if ( + record?.status === "completed" && + record.semanticResult === semanticFingerprint + ) { + return terminal; + } + } catch { + // Non-canonical or legacy terminal fingerprints cannot prove completion. + } + } + return null; +} + +function recoveryPrevalidationFailure( + snapshot: PersistedNativeSession, +): string | null { + if ( + snapshot.terminalTurns !== undefined && + !Array.isArray(snapshot.terminalTurns) + ) { + return "persisted harness terminal history is invalid"; + } + const terminalTurns = snapshot.terminalTurns ?? []; + if (terminalTurns.length > MAX_RECOVERY_TERMINAL_TURNS) { + return "persisted harness terminal history exceeds its recovery limit"; + } + let terminalBytes = 0; + for (const terminal of terminalTurns) { + if ( + typeof terminal?.turnId !== "string" || + typeof terminal.fingerprint !== "string" + ) { + return "persisted harness terminal history is invalid"; + } + if ( + terminal.turnId.length > MAX_RECOVERY_TERMINAL_BYTES || + terminal.fingerprint.length > MAX_RECOVERY_TERMINAL_FINGERPRINT_BYTES + ) { + return "persisted harness terminal history exceeds its recovery limit"; + } + const fingerprintBytes = Buffer.byteLength(terminal.fingerprint); + terminalBytes += Buffer.byteLength(terminal.turnId) + fingerprintBytes; + if ( + fingerprintBytes > MAX_RECOVERY_TERMINAL_FINGERPRINT_BYTES || + terminalBytes > MAX_RECOVERY_TERMINAL_BYTES + ) { + return "persisted harness terminal history exceeds its recovery limit"; + } + } + if ( + snapshot.semanticResult !== undefined && + snapshot.semanticResult !== null && + !isBoundedPersistedJson( + snapshot.semanticResult, + MAX_RECOVERY_SEMANTIC_RESULT_BYTES, + ) + ) { + return "persisted harness semantic result exceeds its recovery limit"; + } + return null; +} + +/** Measure persisted JSON without first serializing or cloning the payload. */ +function isBoundedPersistedJson(value: unknown, maxBytes: number): boolean { + let bytes = 0; + let nodes = 0; + const ancestors = new WeakSet(); + const add = (amount: number): boolean => { + bytes += amount; + return bytes <= maxBytes; + }; + const visit = (candidate: unknown, depth: number): boolean => { + nodes += 1; + if ( + nodes > MAX_RECOVERY_SEMANTIC_RESULT_NODES || + depth > MAX_RECOVERY_SEMANTIC_RESULT_DEPTH + ) { + return false; + } + if (candidate === null) return add(4); + switch (typeof candidate) { + case "string": + return candidate.length + 2 <= maxBytes - bytes + ? add(jsonStringBytes(candidate, maxBytes - bytes)) + : false; + case "boolean": + return add(candidate ? 4 : 5); + case "number": + return Number.isFinite(candidate) + ? add(String(candidate).length) + : false; + case "object": + break; + default: + return false; + } + if (ancestors.has(candidate)) return false; + ancestors.add(candidate); + try { + if (Array.isArray(candidate)) { + if (!add(2)) return false; + for (let index = 0; index < candidate.length; index += 1) { + if (index > 0 && !add(1)) return false; + if (!visit(candidate[index], depth + 1)) return false; + } + return true; + } + const prototype = Object.getPrototypeOf(candidate); + if (prototype !== Object.prototype && prototype !== null) return false; + if (!add(2)) return false; + let propertyCount = 0; + for (const key in candidate as Record) { + if (!Object.prototype.hasOwnProperty.call(candidate, key)) continue; + if (propertyCount > 0 && !add(1)) return false; + propertyCount += 1; + if ( + key.length + 3 > maxBytes - bytes || + !add(jsonStringBytes(key, maxBytes - bytes) + 1) || + !visit((candidate as Record)[key], depth + 1) + ) { + return false; + } + } + return true; + } finally { + ancestors.delete(candidate); + } + }; + try { + return visit(value, 0); + } catch { + return false; + } +} + +function jsonStringBytes(value: string, maxBytes: number): number { + let bytes = 2; + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index); + if (code === 0x22 || code === 0x5c) { + bytes += 2; + } else if (code <= 0x1f) { + bytes += [0x08, 0x09, 0x0a, 0x0c, 0x0d].includes(code) ? 2 : 6; + } else if (code <= 0x7f) { + bytes += 1; + } else if (code <= 0x7ff) { + bytes += 2; + } else if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (next >= 0xdc00 && next <= 0xdfff) { + bytes += 4; + index += 1; + } else { + bytes += 6; + } + } else if (code >= 0xdc00 && code <= 0xdfff) { + bytes += 6; + } else { + bytes += 3; + } + if (bytes > maxBytes) return maxBytes + 1; + } + return bytes; +} + function assertProviderSessionIdentity( session: HarnessSession, provider: string, 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 004193d6b4..01914acf9c 100644 --- a/packages/paperclip-runner/src/backends/native-backend-factory.test.ts +++ b/packages/paperclip-runner/src/backends/native-backend-factory.test.ts @@ -134,7 +134,7 @@ describe("native backend factory", () => { name: "acpx_runtime", version: "0.13.1", capabilities: { - resume: false, + resume: true, interruption: true, dynamicTools: true, }, diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts index 1806909fcf..0069921f40 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts @@ -17,6 +17,7 @@ import type { AcpxRuntimeTurn, OpenAcpxRuntimeHostOptions, } from "./runtime-host.js"; +import type { AcpxRecoveryWorkspaceLease } from "./runtime-sandbox.js"; describe("Codex ACPX harness driver", () => { it("rejects a pre-aborted open before starting host admission", async () => { @@ -136,7 +137,7 @@ describe("Codex ACPX harness driver", () => { kind: "acpx_runtime", displayName: "Codex via ACPX", capabilities: { - resume: false, + resume: true, interruption: true, dynamicTools: true, runtimeRequestResolution: false, @@ -1296,6 +1297,749 @@ describe("Codex ACPX harness driver", () => { ), }); }); + + it("recovers a settled session with the exact persisted identity", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const terminalEvents = collectUntil(session.events(), "turn.completed"); + await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + await fixture.hostOptions!.semanticTools!.handler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-recovery", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await terminalEvents; + const snapshot = await session.snapshot(); + expect(snapshot.terminalTurns?.at(-1)?.turnId).toBe( + snapshot.semanticResult?.turnId, + ); + await session.close({ reason: "simulate restart" }); + + const recovery = await fixture.driver.recoverSession!(snapshot); + + expect(recovery).toMatchObject({ recovered: true }); + expect(fixture.readRecoveryWorkspace).toHaveBeenCalledWith({ + runtimeDirectory: "/runtime", + normalizedSessionId: "session-1", + signal: expect.any(AbortSignal), + }); + expect(fixture.hostOptions?.expectedIdentity).toEqual( + snapshot.providerIdentity, + ); + const workspaceLease = (await fixture.readRecoveryWorkspace.mock + .results[0]!.value) as AcpxRecoveryWorkspaceLease; + expect(fixture.hostOptions?.assertWorkspaceHeld).toBe( + workspaceLease.assertHeld, + ); + expect(workspaceLease.close).toHaveBeenCalledOnce(); + await expect(recovery.session!.snapshot()).resolves.toMatchObject({ + driverSessionId: snapshot.driverSessionId, + providerSessionId: snapshot.providerSessionId, + providerRecoveryPolicy: "same_session_only", + lastSourceSequence: snapshot.lastSourceSequence, + semanticResult: snapshot.semanticResult, + activeTurnId: null, + }); + await recovery.session!.close({ reason: "recovery verified" }); + }); + + it("rejects a pre-aborted recovery before reading its workspace", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-recovery-pre-abort", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const snapshot = await session.snapshot(); + await session.close({ reason: "prepare pre-aborted recovery" }); + const controller = new AbortController(); + const cancellation = new Error("recovery cancelled before start"); + controller.abort(cancellation); + + await expect(fixture.driver.recoverSession!(snapshot, { + signal: controller.signal, + })).resolves.toEqual({ + recovered: false, + reason: cancellation.message, + }); + + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + expect(fixture.openHost).toHaveBeenCalledOnce(); + }); + + it("aborts a blocked recovery workspace read without opening a host", async () => { + const workspaceRead = deferred(); + const lateWorkspaceLease = recoveryWorkspaceLease(); + const fixture = driverFixture({}, { + readRecoveryWorkspace: () => workspaceRead.promise, + }); + const session = await fixture.driver.openSession({ + runId: "run-recovery-read-abort", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const snapshot = await session.snapshot(); + await session.close({ reason: "prepare blocked workspace recovery" }); + const controller = new AbortController(); + const cancellation = new Error("recovery workspace read cancelled"); + const recovery = fixture.driver.recoverSession!(snapshot, { + signal: controller.signal, + }); + await vi.waitFor(() => + expect(fixture.readRecoveryWorkspace).toHaveBeenCalledOnce(), + ); + expect(fixture.readRecoveryWorkspace).toHaveBeenCalledWith({ + runtimeDirectory: "/runtime", + normalizedSessionId: "session-1", + signal: controller.signal, + }); + + controller.abort(cancellation); + await expect(recovery).resolves.toEqual({ + recovered: false, + reason: cancellation.message, + }); + workspaceRead.resolve(lateWorkspaceLease); + await vi.waitFor(() => + expect(lateWorkspaceLease.close).toHaveBeenCalledOnce(), + ); + + expect(fixture.openHost).toHaveBeenCalledOnce(); + }); + + it("closes and quarantines a recovered host that resolves after abort", async () => { + const fixture = driverFixture({}, { closeSettlementTimeoutMs: 5 }); + const session = await fixture.driver.openSession({ + runId: "run-recovery-late-host", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const snapshot = await session.snapshot(); + await session.close({ reason: "prepare late recovered host" }); + fixture.host.close.mockClear(); + fixture.host.close.mockRejectedValueOnce( + new Error("late recovered host close failed once"), + ); + const hostAdmission = deferred>(); + let recoveryHostOptions: OpenAcpxRuntimeHostOptions | undefined; + fixture.openHost.mockImplementationOnce((options) => { + recoveryHostOptions = options; + return hostAdmission.promise; + }); + const controller = new AbortController(); + const cancellation = new Error("recovered host admission cancelled"); + const recovery = fixture.driver.recoverSession!(snapshot, { + signal: controller.signal, + }); + await vi.waitFor(() => expect(fixture.openHost).toHaveBeenCalledTimes(2)); + expect(recoveryHostOptions?.signal).toBe(controller.signal); + + controller.abort(cancellation); + await expect(recovery).resolves.toEqual({ + recovered: false, + reason: cancellation.message, + }); + hostAdmission.resolve(fixture.host); + + await vi.waitFor(() => expect(fixture.host.close).toHaveBeenCalledTimes(2)); + expect(fixture.host.close).toHaveBeenNthCalledWith(1, { + reason: "Codex ACPX host resolved after admission was aborted", + }); + expect(fixture.host.close).toHaveBeenNthCalledWith(2, { + reason: expect.stringContaining("quarantined cleanup recovery"), + }); + }); + + it.each([ + [ + "failed", + "turn.failed", + { + status: "failed", + error: { + code: "provider_failure", + message: "The follow-up failed", + retryable: false, + }, + }, + ], + [ + "cancelled", + "turn.interrupted", + { status: "cancelled", stopReason: "operator_cancelled" }, + ], + ] as const)( + "rejects an earlier semantic settlement after a later %s turn", + async (_status, terminalType, terminalResult) => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-later-unsuccessful-turn", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + + const semanticTerminal = collectUntil(session.events(), "turn.completed"); + const semanticTurn = await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + await fixture.hostOptions!.semanticTools!.handler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-before-follow-up", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await semanticTerminal; + + const laterTerminal = collectUntil(session.events(), terminalType); + const laterTurn = await session.startTurn({ + message: { role: "user", text: "Attempt a follow-up." }, + }); + fixture.finishTurn(terminalResult); + await laterTerminal; + + const snapshot = await session.snapshot(); + expect(snapshot).toMatchObject({ + activeTurnId: null, + semanticResult: { turnId: semanticTurn.turnId }, + }); + expect(snapshot.terminalTurns?.at(-1)?.turnId).toBe(laterTurn.turnId); + await session.close({ reason: "simulate unsuccessful follow-up recovery" }); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX semantic result is not the latest terminal settlement", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + }, + ); + + it("transfers an identical semantic retry from a failed turn to its successful turn", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-semantic-retry", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const bridgeHandler = fixture.hostOptions!.semanticTools!.handler; + + const firstTerminal = collectUntil(session.events(), "turn.failed"); + const first = await session.startTurn({ + message: { role: "user", text: "Attempt the task." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-failed-attempt", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ + status: "failed", + error: { code: "provider_retry", message: "Retry the turn", retryable: true }, + }); + await firstTerminal; + + const secondTerminal = collectUntil(session.events(), "turn.completed"); + const second = await session.startTurn({ + message: { role: "user", text: "Record the successful retry." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-successful-retry", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await secondTerminal; + + const snapshot = await session.snapshot(); + expect(snapshot.semanticResult).toMatchObject({ + callId: "finish-successful-retry", + turnId: second.turnId, + }); + expect(snapshot.semanticResult?.turnId).not.toBe(first.turnId); + expect(snapshot.terminalTurns?.at(-1)?.turnId).toBe(second.turnId); + expect(snapshot.terminalTurns).toEqual(expect.arrayContaining([ + expect.objectContaining({ turnId: first.turnId }), + expect.objectContaining({ turnId: second.turnId }), + ])); + const successfulTerminal = snapshot.terminalTurns?.find( + (terminal) => terminal.turnId === second.turnId, + ); + expect(JSON.parse(successfulTerminal!.fingerprint)).toEqual({ + status: "completed", + semanticResult: snapshot.semanticResult!.fingerprint, + }); + await session.close({ reason: "simulate successful retry recovery" }); + + await expect(fixture.driver.recoverSession!({ + ...snapshot, + activeTurnId: first.turnId, + })).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX active turn is not the completed semantic settlement", + }); + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toMatchObject({ + recovered: true, + }); + }); + + it("transfers an identical reaffirmed result to the latest successful turn", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-semantic-reaffirmation", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const bridgeHandler = fixture.hostOptions!.semanticTools!.handler; + + const firstTerminal = collectUntil(session.events(), "turn.completed"); + const first = await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-first-success", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await firstTerminal; + + const secondTerminal = collectUntil(session.events(), "turn.completed"); + const second = await session.startTurn({ + message: { role: "user", text: "Reaffirm the same disposition." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-reaffirmed-success", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + const reaffirmedEvents = await secondTerminal; + + expect( + reaffirmedEvents.filter( + (event) => event.eventType === "run.result.proposed", + ), + ).toHaveLength(1); + const snapshot = await session.snapshot(); + expect(snapshot.semanticResult).toMatchObject({ + callId: "finish-reaffirmed-success", + turnId: second.turnId, + }); + expect(snapshot.semanticResult?.turnId).not.toBe(first.turnId); + expect(JSON.parse(snapshot.terminalTurns!.at(-1)!.fingerprint)).toEqual({ + status: "completed", + semanticResult: snapshot.semanticResult!.fingerprint, + }); + await session.close({ reason: "simulate reaffirmed result recovery" }); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toMatchObject({ + recovered: true, + }); + }); + + it("rejects an earlier completed result after an identical retry fails", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-failed-semantic-reaffirmation", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const bridgeHandler = fixture.hostOptions!.semanticTools!.handler; + + const firstTerminal = collectUntil(session.events(), "turn.completed"); + const first = await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-before-failed-reaffirmation", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await firstTerminal; + + const failedTerminal = collectUntil(session.events(), "turn.failed"); + const second = await session.startTurn({ + message: { role: "user", text: "Reaffirm before a failed retry." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-failed-reaffirmation", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ + status: "failed", + error: { + code: "provider_retry", + message: "Retry failed after reaffirming", + retryable: true, + }, + }); + const failedEvents = await failedTerminal; + expect( + failedEvents.filter( + (event) => event.eventType === "run.result.proposed", + ), + ).toHaveLength(1); + + const snapshot = await session.snapshot(); + expect(snapshot.semanticResult).toMatchObject({ + callId: "finish-before-failed-reaffirmation", + turnId: first.turnId, + }); + expect(snapshot.semanticResult?.turnId).not.toBe(second.turnId); + expect(JSON.parse(snapshot.terminalTurns!.at(-1)!.fingerprint)).toEqual({ + status: "failed", + reaffirmedSemanticResult: snapshot.semanticResult!.fingerprint, + }); + await session.close({ reason: "simulate failed reaffirmation recovery" }); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX semantic result is not the latest terminal settlement", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + }); + + it("rejects an earlier completed result after close interrupts an identical retry", async () => { + const fixture = driverFixture({}, { closeSettlementTimeoutMs: 1 }); + const session = await fixture.driver.openSession({ + runId: "run-close-interrupted-semantic-reaffirmation", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const bridgeHandler = fixture.hostOptions!.semanticTools!.handler; + + const firstTerminal = collectUntil(session.events(), "turn.completed"); + const first = await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-before-close-reaffirmation", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await firstTerminal; + + const interruptedTerminal = collectUntil( + session.events(), + "turn.interrupted", + ); + const second = await session.startTurn({ + message: { role: "user", text: "Reaffirm while shutdown begins." }, + }); + await bridgeHandler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-close-reaffirmation", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.host.close.mockResolvedValue(undefined); + await session.close({ reason: "close before provider result settles" }); + await interruptedTerminal; + + const snapshot = await session.snapshot(); + expect(snapshot.semanticResult).toMatchObject({ + callId: "finish-before-close-reaffirmation", + turnId: first.turnId, + }); + expect(snapshot.semanticResult?.turnId).not.toBe(second.turnId); + expect(JSON.parse(snapshot.terminalTurns!.at(-1)!.fingerprint)).toEqual({ + status: "interrupted", + reaffirmedSemanticResult: snapshot.semanticResult!.fingerprint, + }); + + fixture.finishTurn({ status: "cancelled", stopReason: "session_closed" }); + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX semantic result is not the latest terminal settlement", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + }); + + it("does not transfer a failed turn's semantic result to an unrelated resultless turn", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-resultless-semantic-retry", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + + const firstTerminal = collectUntil(session.events(), "turn.failed"); + const first = await session.startTurn({ + message: { role: "user", text: "Attempt the task." }, + }); + await fixture.hostOptions!.semanticTools!.handler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-before-provider-retry", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ + status: "failed", + error: { + code: "provider_retry", + message: "Retry the turn", + retryable: true, + }, + }); + await firstTerminal; + + const secondTerminal = collectUntil(session.events(), "turn.completed"); + const second = await session.startTurn({ + message: { role: "user", text: "Confirm the completed work." }, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await secondTerminal; + + const snapshot = await session.snapshot(); + expect(snapshot.semanticResult).toMatchObject({ + callId: "finish-before-provider-retry", + turnId: first.turnId, + }); + const successfulTerminal = snapshot.terminalTurns?.find( + (terminal) => terminal.turnId === second.turnId, + ); + expect(JSON.parse(successfulTerminal!.fingerprint)).toEqual({ + status: "completed", + semanticResult: null, + }); + await session.close({ reason: "simulate resultless retry recovery" }); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX semantic result has no completed terminal turn", + }); + }); + + it("clears a checkpoint race when the active turn is already terminal", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-terminal-race", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const terminalEvents = collectUntil(session.events(), "turn.completed"); + const { turnId } = await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await terminalEvents; + const snapshot = await session.snapshot(); + snapshot.activeTurnId = turnId; + await session.close({ reason: "simulate checkpoint race" }); + + const recovery = await fixture.driver.recoverSession!(snapshot); + + expect(recovery).toMatchObject({ recovered: true }); + await expect(recovery.session!.snapshot()).resolves.toMatchObject({ + activeTurnId: null, + terminalTurns: [expect.objectContaining({ turnId })], + }); + await recovery.session!.close({ reason: "checkpoint race verified" }); + }); + + it("fails closed when a checkpoint contains an unproved active turn", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-active-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + await session.startTurn({ + message: { role: "user", text: "Continue working." }, + }); + const snapshot = await session.snapshot(); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: "active Codex ACPX turn continuity is unavailable", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + expect(fixture.openHost).toHaveBeenCalledOnce(); + await session.close({ reason: "active recovery rejected" }); + }); + + it("rejects a tampered recovery result before reopening the provider", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-tampered-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const snapshot = await session.snapshot(); + snapshot.semanticResult = { + result: completedResult(), + fingerprint: "tampered", + turnId: "turn-settled", + }; + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: "persisted Codex ACPX semantic result is invalid", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + expect(fixture.openHost).toHaveBeenCalledOnce(); + await session.close({ reason: "tampered recovery rejected" }); + }); + + it("rejects semantic results from failed terminal turns", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-failed-semantic-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const terminalEvents = collectUntil(session.events(), "turn.completed"); + await session.startTurn({ + message: { role: "user", text: "Complete the task." }, + }); + await fixture.hostOptions!.semanticTools!.handler({ + tool: PRP_COMPLETION_TOOL_NAME, + callId: "finish-before-failure", + arguments: completedResult(), + signal: new AbortController().signal, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await terminalEvents; + const snapshot = await session.snapshot(); + snapshot.terminalTurns = snapshot.terminalTurns?.map((terminal) => ({ + ...terminal, + fingerprint: JSON.stringify({ status: "failed" }), + })); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX semantic result has no completed terminal turn", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + await session.close({ reason: "failed semantic recovery rejected" }); + }); + + it("rejects failed resultless terminals as disposition settlements", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-failed-resultless-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const terminalEvents = collectUntil(session.events(), "turn.completed"); + const { turnId } = await session.startTurn({ + message: { role: "user", text: "Attempt the task." }, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await terminalEvents; + const snapshot = await session.snapshot(); + snapshot.terminalTurns = snapshot.terminalTurns?.map((terminal) => ({ + ...terminal, + fingerprint: JSON.stringify({ status: "failed" }), + })); + + for (const activeTurnId of [turnId, null]) { + await expect(fixture.driver.recoverSession!({ + ...snapshot, + activeTurnId, + })).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX resultless recovery requires a completed terminal turn", + }); + } + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + await session.close({ reason: "failed resultless recovery rejected" }); + }); + + it("rejects a stale completed active turn before a later resultless failure", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-stale-resultless-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const terminalEvents = collectUntil(session.events(), "turn.completed"); + const { turnId } = await session.startTurn({ + message: { role: "user", text: "Complete without a semantic result." }, + }); + fixture.finishTurn({ status: "completed", stopReason: "end_turn" }); + await terminalEvents; + const snapshot = await session.snapshot(); + snapshot.activeTurnId = turnId; + snapshot.terminalTurns?.push({ + turnId: "turn-later-failed", + fingerprint: JSON.stringify({ status: "failed" }), + }); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: + "persisted Codex ACPX resultless recovery requires a completed terminal turn", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + await session.close({ reason: "stale resultless recovery rejected" }); + }); + + it("rejects an unimplemented replacement policy before reopening", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-policy-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const snapshot = await session.snapshot(); + snapshot.providerRecoveryPolicy = "allow_replacement_after_resume_failure"; + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: "persisted Codex ACPX recovery policy is unsupported", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + expect(fixture.openHost).toHaveBeenCalledOnce(); + await session.close({ reason: "replacement recovery rejected" }); + }); + + it("rejects oversized terminal history before reopening", async () => { + const fixture = driverFixture(); + const session = await fixture.driver.openSession({ + runId: "run-bounded-recovery", + normalizedSessionId: "session-1", + workingDirectory: "/workspace", + }); + const snapshot = await session.snapshot(); + snapshot.terminalTurns = Array.from({ length: 4_097 }, (_, index) => ({ + turnId: `turn-${index}`, + fingerprint: "terminal", + })); + + await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({ + recovered: false, + reason: "persisted Codex ACPX terminal history exceeds its limit", + }); + expect(fixture.readRecoveryWorkspace).not.toHaveBeenCalled(); + expect(fixture.openHost).toHaveBeenCalledOnce(); + await session.close({ reason: "bounded recovery rejected" }); + }); }); function driverFixture( @@ -1307,47 +2051,55 @@ function driverFixture( maxBufferedEvents?: number; terminalEventReserve?: number; openHost?: NonNullable; + readRecoveryWorkspace?: NonNullable< + CodexAcpxDriverDependencies["readRecoveryWorkspace"] + >; } = {}, ): { driver: CodexAcpxDriver; host: ReturnType; openHost: ReturnType; + readRecoveryWorkspace: ReturnType; hostOptions: OpenAcpxRuntimeHostOptions | null; finishTurn(result: Awaited): void; } { - const result = deferred>(); - const turn: AcpxRuntimeTurn = { - requestId: "provider-turn-1", - promptStarted: Promise.resolve(), - events: { - async *[Symbol.asyncIterator]() { - yield* fixtureOptions.runtimeEvents ?? [ - { - type: "text_delta" as const, - text: "Task complete.", - stream: "output" as const, - }, - { - type: "tool_call" as const, - toolCallId: "provider-tool-1", - title: "Read", - kind: "read" as const, - status: "pending", - tag: "tool_call", - text: "Reading", - }, - ]; - if (fixtureOptions.runtimeEventFailure) { - await fixtureOptions.runtimeEventFailure; - } + let turnCount = 0; + let activeResult: ReturnType>> | null = null; + const createTurn = (): AcpxRuntimeTurn => { + activeResult = deferred>(); + return { + requestId: `provider-turn-${++turnCount}`, + promptStarted: Promise.resolve(), + events: { + async *[Symbol.asyncIterator]() { + yield* fixtureOptions.runtimeEvents ?? [ + { + type: "text_delta" as const, + text: "Task complete.", + stream: "output" as const, + }, + { + type: "tool_call" as const, + toolCallId: "provider-tool-1", + title: "Read", + kind: "read" as const, + status: "pending", + tag: "tool_call", + text: "Reading", + }, + ]; + if (fixtureOptions.runtimeEventFailure) { + await fixtureOptions.runtimeEventFailure; + } + }, }, - }, - result: result.promise, - cancel: vi.fn(async () => undefined), - closeStream: vi.fn(async () => undefined), + result: activeResult.promise, + cancel: vi.fn(async () => undefined), + closeStream: vi.fn(async () => undefined), + }; }; - const host = fakeHost(turn, () => - result.resolve({ status: "cancelled", stopReason: "session_closed" }), + const host = fakeHost(createTurn, () => + activeResult?.resolve({ status: "cancelled", stopReason: "session_closed" }), ); let hostOptions: OpenAcpxRuntimeHostOptions | null = null; const openHost = vi.fn( @@ -1357,8 +2109,13 @@ function driverFixture( return host; }), ); + const readRecoveryWorkspace = vi.fn( + fixtureOptions.readRecoveryWorkspace ?? + (async () => recoveryWorkspaceLease()), + ); const dependencies: CodexAcpxDriverDependencies = { openHost, + readRecoveryWorkspace, closeSettlementTimeoutMs: fixtureOptions.closeSettlementTimeoutMs, maxBufferedEvents: fixtureOptions.maxBufferedEvents, terminalEventReserve: fixtureOptions.terminalEventReserve, @@ -1383,14 +2140,31 @@ function driverFixture( driver, host, openHost, + readRecoveryWorkspace, get hostOptions() { return hostOptions; }, - finishTurn: result.resolve, + finishTurn(result) { + if (!activeResult) throw new Error("No active fixture turn"); + activeResult.resolve(result); + }, }; } -function fakeHost(turn: AcpxRuntimeTurn, onClose: () => void) { +function recoveryWorkspaceLease( + path = "/workspace", +): AcpxRecoveryWorkspaceLease & { + assertHeld: ReturnType; + close: ReturnType; +} { + return { + path, + assertHeld: vi.fn(), + close: vi.fn(async () => undefined), + }; +} + +function fakeHost(createTurn: () => AcpxRuntimeTurn, onClose: () => void) { return { identity: () => ({ schema: "paperclip.runner.acpx-identity.v1" as const, @@ -1422,7 +2196,7 @@ function fakeHost(turn: AcpxRuntimeTurn, onClose: () => void) { availableModelIds: ["gpt-5.6-sol"], }, })), - startTurn: vi.fn(() => turn), + startTurn: vi.fn(createTurn), interruptActiveTurn: vi.fn(async () => undefined), close: vi.fn(async () => { onClose(); diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts index f28f669a94..8b1b965543 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts @@ -13,6 +13,8 @@ import { type HarnessDriverConfigValidation, type HarnessDriverDescriptor, type HarnessSession, + type HarnessSessionRecoveryOptions, + type HarnessSessionRecoveryResult, type HarnessTranscriptSnapshot, type OpenHarnessSessionInput, type PersistedHarnessSession, @@ -46,12 +48,18 @@ import { type AcpxRuntimeTurn, type OpenAcpxRuntimeHostOptions, } from "./runtime-host.js"; +import { + readAcpxRecoveryWorkspace, + type AcpxRecoveryWorkspaceLease, +} from "./runtime-sandbox.js"; const MAX_BUFFERED_EVENTS = 512; const TERMINAL_EVENT_RESERVE = 3; const TURN_START_EVENT_COUNT = 3; const MAX_TRANSCRIPT_EVENTS = 1_024; const MAX_TRANSCRIPT_BYTES = 8 * 1024 * 1024; +const MAX_RECOVERY_TERMINAL_TURNS = 4_096; +const MAX_RECOVERY_TERMINAL_BYTES = 8 * 1024 * 1024; const CLOSE_TURN_SETTLEMENT_TIMEOUT_MS = 2_000; const MAX_AUTONOMOUS_HOST_CLOSE_RETRIES = 3; const MAX_QUARANTINED_HOST_CLOSE_RETRIES = 3; @@ -124,6 +132,11 @@ export interface CodexAcpxDriverDependencies { maxBufferedEvents?: number; /** Internal test seam; production reserves fixed terminal-event capacity. */ terminalEventReserve?: number; + readRecoveryWorkspace?: (input: { + runtimeDirectory: string; + normalizedSessionId: string; + signal?: AbortSignal; + }) => Promise; } /** Codex-only HarnessDriver backed by the admitted ACPX runtime host. */ @@ -135,6 +148,9 @@ export class CodexAcpxDriver implements HarnessDriver { readonly #terminalEventReserve: number; readonly #cleanupOwners = new Set>(); readonly #quarantinedHostCleanups = new Set(); + readonly #readRecoveryWorkspace: NonNullable< + CodexAcpxDriverDependencies["readRecoveryWorkspace"] + >; constructor( options: CodexAcpxDriverOptions, @@ -168,6 +184,8 @@ export class CodexAcpxDriver implements HarnessDriver { this.#terminalEventReserve + TURN_START_EVENT_COUNT, Math.floor(dependencies.maxBufferedEvents ?? MAX_BUFFERED_EVENTS), ); + this.#readRecoveryWorkspace = + dependencies.readRecoveryWorkspace ?? readAcpxRecoveryWorkspace; } async descriptor(): Promise { @@ -182,11 +200,10 @@ export class CodexAcpxDriver implements HarnessDriver { }, capabilities: { ...descriptor.capabilities, - resume: false, + resume: true, runtimeRequestResolution: false, runtimeRequestHandoff: false, unsupported: [ - "resume", "steering", "runtimeRequestResolution", "runtimeRequestHandoff", @@ -218,6 +235,101 @@ export class CodexAcpxDriver implements HarnessDriver { await runAbortableDriverAdmission(input.signal, () => this.#retryQuarantinedHostCleanups(), ); + return await this.#open(input, null); + } + + async recoverSession( + snapshot: PersistedHarnessSession, + options: HarnessSessionRecoveryOptions = { + signal: new AbortController().signal, + }, + ): Promise { + try { + await runAbortableDriverAdmission( + options.signal, + () => this.#retryQuarantinedHostCleanups(), + ); + validateRecoverySnapshot(snapshot); + const terminalTurnIds = new Set( + (snapshot.terminalTurns ?? []).map(({ turnId }) => turnId), + ); + if ( + snapshot.activeTurnId && + !terminalTurnIds.has(snapshot.activeTurnId) + ) { + return { + recovered: false, + reason: "active Codex ACPX turn continuity is unavailable", + }; + } + const workspaceLease = await this.#readRecoveryWorkspaceForAdmission( + { + runtimeDirectory: this.#options.runtimeDirectory, + normalizedSessionId: snapshot.normalizedSessionId!, + signal: options.signal, + }, + options.signal, + ); + let recoveredSession: HarnessSession | null = null; + try { + recoveredSession = await this.#open( + { + runId: snapshot.runId!, + normalizedSessionId: snapshot.normalizedSessionId!, + workingDirectory: workspaceLease.path, + signal: options.signal, + }, + snapshot, + workspaceLease.assertHeld, + ); + await workspaceLease.close(); + return { recovered: true, session: recoveredSession }; + } catch (error) { + await workspaceLease.close().catch(() => undefined); + if (recoveredSession) { + await recoveredSession.close({ + reason: "ACPX recovery workspace lease cleanup failed", + force: true, + }).catch(() => undefined); + } + throw error; + } + } catch (error) { + return { recovered: false, reason: safeMessage(error) }; + } + } + + async #readRecoveryWorkspaceForAdmission( + input: Parameters< + NonNullable + >[0], + signal: AbortSignal | undefined, + ): Promise { + if (signal === undefined) return await this.#readRecoveryWorkspace(input); + signal.throwIfAborted(); + const pending = Promise.resolve().then(() => + this.#readRecoveryWorkspace(input), + ); + try { + return await raceDriverAdmissionWithAbort(pending, signal); + } catch (error) { + if (signal.aborted) { + this.#retainCleanup( + pending.then( + (lease) => lease.close(), + () => undefined, + ), + ); + } + throw error; + } + } + + async #open( + input: OpenHarnessSessionInput, + snapshot: PersistedHarnessSession | null, + assertWorkspaceHeld?: () => void, + ): Promise { let session: CodexAcpxSession | null = null; const host = await this.#openHostForAdmission( { @@ -231,6 +343,10 @@ export class CodexAcpxDriver implements HarnessDriver { environment: this.#options.environment, managedCodexCredentialSourcePath: this.#options.managedCodexCredentialSourcePath, + ...(assertWorkspaceHeld === undefined ? {} : { assertWorkspaceHeld }), + ...(snapshot?.providerIdentity?.kind === "acpx" + ? { expectedIdentity: snapshot.providerIdentity } + : {}), ...(input.signal === undefined ? {} : { signal: input.signal }), semanticTools: { tools: this.#options.dynamicTools ?? [], @@ -257,6 +373,7 @@ export class CodexAcpxDriver implements HarnessDriver { retainCleanup: (cleanup) => this.#retainCleanup(cleanup), quarantineCleanup: (hostToRetain, reason) => this.#quarantineHostCleanup(hostToRetain, reason), + snapshot, }); return session; } catch (error) { @@ -501,12 +618,21 @@ class CodexAcpxSession implements HarnessSession { readonly #transcript: Array<{ event: PrpEvent; bytes: number }> = []; readonly #terminalTurns = new Map(); readonly #sourceInstanceId: string; + readonly #providerRecoveryPolicy: NonNullable< + PersistedHarnessSession["providerRecoveryPolicy"] + >; #sourceSequence = 0; #activeTurnId: string | null = null; #semanticResult: PrpStructuredRunResult | null = null; #semanticFingerprint: string | null = null; #semanticCallId: string | null = null; #semanticTurnId: string | null = null; + #pendingSemanticTransfer: { + result: PrpStructuredRunResult; + fingerprint: string; + callId: string; + turnId: string; + } | null = null; #usage: Record | null = null; #assistantText = ""; #closed = false; @@ -538,6 +664,7 @@ class CodexAcpxSession implements HarnessSession { terminalEventReserve: number; retainCleanup: (cleanup: Promise) => void; quarantineCleanup: (host: CodexAcpxHost, reason: string) => void; + snapshot: PersistedHarnessSession | null; }) { const identity = input.host.identity(); if (identity.normalizedSessionId !== input.input.normalizedSessionId) { @@ -557,6 +684,23 @@ class CodexAcpxSession implements HarnessSession { "paperclip-acpx", input.input.normalizedSessionId, ); + this.#sourceSequence = input.snapshot?.lastSourceSequence ?? 0; + this.#activeTurnId = input.snapshot?.activeTurnId ?? null; + this.#providerRecoveryPolicy = + input.snapshot?.providerRecoveryPolicy ?? "same_session_only"; + const semantic = input.snapshot?.semanticResult; + if (semantic) { + this.#semanticResult = structuredClone(semantic.result); + this.#semanticFingerprint = semantic.fingerprint; + this.#semanticCallId = semantic.callId ?? null; + this.#semanticTurnId = semantic.turnId; + } + for (const terminal of input.snapshot?.terminalTurns ?? []) { + this.#terminalTurns.set(terminal.turnId, terminal.fingerprint); + } + if (this.#activeTurnId && this.#terminalTurns.has(this.#activeTurnId)) { + this.#activeTurnId = null; + } } ids() { @@ -678,22 +822,42 @@ class CodexAcpxSession implements HarnessSession { ) { throw new Error("A different semantic result is already committed"); } - if (this.#semanticFingerprint === null) { - if ( - !this.#emit("run.result.proposed", validation.result, { - turnId, - itemId: call.callId, - }) - ) { + const claimsLaterTurn = + this.#semanticFingerprint === fingerprint && + this.#semanticTurnId !== turnId; + const repeatsPendingTransfer = + claimsLaterTurn && + this.#pendingSemanticTransfer?.fingerprint === fingerprint && + this.#pendingSemanticTransfer.turnId === turnId; + if ( + this.#semanticFingerprint === null || + (claimsLaterTurn && !repeatsPendingTransfer) + ) { + if (!this.#emit("run.result.proposed", validation.result, { + turnId, + itemId: call.callId, + })) { throw new HarnessCapabilityUnavailableError( "run.result.proposed", "the event consumer must drain provider events before a semantic result can be accepted", ); } - this.#semanticResult = structuredClone(validation.result); - this.#semanticFingerprint = fingerprint; - this.#semanticCallId = call.callId; - this.#semanticTurnId = turnId; + if (claimsLaterTurn) { + // A reaffirming retry does not own the durable result until its + // provider turn completes successfully. A failed or interrupted + // retry must leave the last completed owner recoverable. + this.#pendingSemanticTransfer = { + result: structuredClone(validation.result), + fingerprint, + callId: call.callId, + turnId, + }; + } else { + this.#semanticResult = structuredClone(validation.result); + this.#semanticFingerprint = fingerprint; + this.#semanticCallId = call.callId; + this.#semanticTurnId = turnId; + } } return { accepted: true }; } @@ -750,6 +914,7 @@ class CodexAcpxSession implements HarnessSession { driverKind: "acpx_runtime", driverSessionId: identity.acpxRecordId, providerSessionId: identity.agentSessionId, + providerRecoveryPolicy: this.#providerRecoveryPolicy, runId: this.#input.runId, normalizedSessionId: this.#input.normalizedSessionId, activeTurnId: this.#activeTurnId, @@ -830,9 +995,18 @@ class CodexAcpxSession implements HarnessSession { pendingTerminal.payload, ); } else { + const reaffirmedSemanticResult = + this.#pendingSemanticTransfer?.turnId === closingTurnId + ? this.#pendingSemanticTransfer.fingerprint + : null; this.#publishTerminal( closingTurnId, - canonicalJson({ status: "interrupted" }), + canonicalJson({ + status: "interrupted", + ...(reaffirmedSemanticResult === null + ? {} + : { reaffirmedSemanticResult }), + }), "turn.interrupted", { status: "interrupted", stopReason: "session_closed" }, ); @@ -927,6 +1101,12 @@ class CodexAcpxSession implements HarnessSession { const result = await turn.result; if (this.#terminalTurns.has(turnId)) return; if (result.status === "completed") { + const completedSemanticFingerprint = + this.#pendingSemanticTransfer?.turnId === turnId + ? this.#pendingSemanticTransfer.fingerprint + : this.#semanticTurnId === turnId + ? this.#semanticFingerprint + : null; const finalText = this.#assistantText.trim(); if (finalText) { this.#emit( @@ -939,15 +1119,24 @@ class CodexAcpxSession implements HarnessSession { turnId, canonicalJson({ status: "completed", - semanticResult: this.#semanticFingerprint, + semanticResult: completedSemanticFingerprint, }), "turn.completed", { status: "completed", stopReason: result.stopReason ?? null }, ); } else if (result.status === "cancelled") { + const reaffirmedSemanticResult = + this.#pendingSemanticTransfer?.turnId === turnId + ? this.#pendingSemanticTransfer.fingerprint + : null; this.#publishTerminal( turnId, - canonicalJson({ status: "interrupted" }), + canonicalJson({ + status: "interrupted", + ...(reaffirmedSemanticResult === null + ? {} + : { reaffirmedSemanticResult }), + }), "turn.interrupted", { status: "interrupted", @@ -955,6 +1144,10 @@ class CodexAcpxSession implements HarnessSession { }, ); } else { + const reaffirmedSemanticResult = + this.#pendingSemanticTransfer?.turnId === turnId + ? this.#pendingSemanticTransfer.fingerprint + : null; this.#emit( "provider.notice.recorded", { @@ -971,7 +1164,12 @@ class CodexAcpxSession implements HarnessSession { ); this.#publishTerminal( turnId, - canonicalJson({ status: "failed" }), + canonicalJson({ + status: "failed", + ...(reaffirmedSemanticResult === null + ? {} + : { reaffirmedSemanticResult }), + }), "turn.failed", { status: "failed", @@ -986,16 +1184,34 @@ class CodexAcpxSession implements HarnessSession { if (this.#terminalTurns.has(turnId)) return; if (error instanceof TerminalEventCapacityError) throw error; if (this.#closed || this.#closingStarted) { + const reaffirmedSemanticResult = + this.#pendingSemanticTransfer?.turnId === turnId + ? this.#pendingSemanticTransfer.fingerprint + : null; this.#publishTerminal( turnId, - canonicalJson({ status: "interrupted" }), + canonicalJson({ + status: "interrupted", + ...(reaffirmedSemanticResult === null + ? {} + : { reaffirmedSemanticResult }), + }), "turn.interrupted", { status: "interrupted", stopReason: "session_closed" }, ); } else { + const reaffirmedSemanticResult = + this.#pendingSemanticTransfer?.turnId === turnId + ? this.#pendingSemanticTransfer.fingerprint + : null; this.#publishTerminal( turnId, - canonicalJson({ status: "failed" }), + canonicalJson({ + status: "failed", + ...(reaffirmedSemanticResult === null + ? {} + : { reaffirmedSemanticResult }), + }), "turn.failed", { status: "failed", error: { message: safeMessage(error) } }, ); @@ -1038,6 +1254,17 @@ class CodexAcpxSession implements HarnessSession { } this.#terminalTurns.set(turnId, fingerprint); this.#pendingTerminal = null; + if (this.#pendingSemanticTransfer?.turnId === turnId) { + if (eventType === "turn.completed") { + this.#semanticResult = structuredClone( + this.#pendingSemanticTransfer.result, + ); + this.#semanticFingerprint = this.#pendingSemanticTransfer.fingerprint; + this.#semanticCallId = this.#pendingSemanticTransfer.callId; + this.#semanticTurnId = turnId; + } + this.#pendingSemanticTransfer = null; + } if (this.#activeTurnId === turnId) this.#activeTurnId = null; } @@ -1190,6 +1417,208 @@ function canonicalJson(value: unknown): string { return JSON.stringify(value) ?? "undefined"; } +function validateRecoverySnapshot(snapshot: PersistedHarnessSession): void { + if ( + snapshot.driverKind !== "acpx_runtime" || + !snapshot.runId?.trim() || + !snapshot.normalizedSessionId?.trim() || + snapshot.providerIdentity?.kind !== "acpx" + ) { + throw new Error("persisted Codex ACPX session identity is incomplete"); + } + const identity = snapshot.providerIdentity; + if ( + !boundedIdentity(snapshot.runId) || + !boundedIdentity(snapshot.normalizedSessionId) || + identity.normalizedSessionId !== snapshot.normalizedSessionId || + identity.acpxRecordId !== snapshot.driverSessionId || + identity.agentSessionId !== snapshot.providerSessionId || + ![ + identity.normalizedSessionId, + identity.acpxRecordId, + identity.backendSessionId, + identity.agentSessionId, + identity.requestedModel, + identity.effectiveModel, + ].every(boundedIdentity) || + !/^sha256:[a-f0-9]{64}$/.test(identity.profileDigest) || + !/^sha256:[a-f0-9]{64}$/.test(identity.workspaceDigest) || + (identity.permissionMode !== undefined && + !["approve-all", "approve-reads", "deny-all"].includes( + identity.permissionMode, + )) + ) { + throw new Error("persisted Codex ACPX session identity is inconsistent"); + } + if ( + snapshot.providerRecoveryPolicy !== undefined && + snapshot.providerRecoveryPolicy !== "same_session_only" + ) { + throw new Error("persisted Codex ACPX recovery policy is unsupported"); + } + if ( + (snapshot.pendingRuntimeRequests?.length ?? 0) > 0 || + (snapshot.lineage?.length ?? 0) > 0 || + snapshot.goal != null + ) { + throw new Error("persisted Codex ACPX snapshot has unsupported state"); + } + if ( + snapshot.lastSourceSequence !== undefined && + (!Number.isSafeInteger(snapshot.lastSourceSequence) || + snapshot.lastSourceSequence < 0) + ) { + throw new Error("persisted Codex ACPX source sequence is invalid"); + } + if ( + snapshot.terminalTurns !== undefined && + !Array.isArray(snapshot.terminalTurns) + ) { + throw new Error("persisted Codex ACPX terminal history is invalid"); + } + const terminalTurns = snapshot.terminalTurns ?? []; + if (terminalTurns.length > MAX_RECOVERY_TERMINAL_TURNS) { + throw new Error("persisted Codex ACPX terminal history exceeds its limit"); + } + const terminalTurnIds = new Set(); + let terminalBytes = 0; + for (const terminal of terminalTurns) { + terminalBytes += + Buffer.byteLength(terminal.turnId ?? "") + + Buffer.byteLength(terminal.fingerprint ?? ""); + if ( + !boundedIdentity(terminal.turnId) || + !terminal.fingerprint || + Buffer.byteLength(terminal.fingerprint) > 256 * 1024 || + terminalBytes > MAX_RECOVERY_TERMINAL_BYTES || + terminalTurnIds.has(terminal.turnId) + ) { + throw new Error("persisted Codex ACPX terminal turn is invalid"); + } + terminalTurnIds.add(terminal.turnId); + } + if ( + snapshot.activeTurnId !== undefined && + snapshot.activeTurnId !== null && + !boundedIdentity(snapshot.activeTurnId) + ) { + throw new Error("persisted Codex ACPX active turn is invalid"); + } + const semantic = snapshot.semanticResult; + if (semantic) { + const validation = validatePrpStructuredRunResult(semantic.result); + if ( + !validation.ok || + semantic.fingerprint !== canonicalJson(validation.result) || + !boundedIdentity(semantic.turnId) || + (semantic.callId !== undefined && + semantic.callId !== null && + !boundedIdentity(semantic.callId)) + ) { + throw new Error("persisted Codex ACPX semantic result is invalid"); + } + const semanticTerminalIndex = terminalTurns.findIndex( + (terminal) => terminal.turnId === semantic.turnId, + ); + const semanticTerminal = terminalTurns[semanticTerminalIndex]; + if ( + !semanticTerminal || + !isCompletedSemanticTerminal( + semanticTerminal.fingerprint, + semantic.fingerprint, + ) + ) { + throw new Error( + "persisted Codex ACPX semantic result has no completed terminal turn", + ); + } + if (semanticTerminalIndex !== terminalTurns.length - 1) { + // A later failed or interrupted turn may have reaffirmed the same result, + // but it still supersedes the earlier settlement as the latest durable + // provider fact. Recovery must not finalize an earlier success after a + // newer attempt failed to complete. + throw new Error( + "persisted Codex ACPX semantic result is not the latest terminal settlement", + ); + } + if ( + snapshot.activeTurnId !== undefined && + snapshot.activeTurnId !== null + ) { + const activeTerminal = terminalTurns.find( + (terminal) => terminal.turnId === snapshot.activeTurnId, + ); + if ( + snapshot.activeTurnId !== semantic.turnId || + !activeTerminal || + !isCompletedSemanticTerminal( + activeTerminal.fingerprint, + semantic.fingerprint, + ) + ) { + throw new Error( + "persisted Codex ACPX active turn is not the completed semantic settlement", + ); + } + } + } else if (terminalTurns.length > 0) { + const latestTerminalTurnId = terminalTurns.at(-1)!.turnId; + const settlementTurnId = snapshot.activeTurnId ?? latestTerminalTurnId; + const settlement = terminalTurns.find( + (terminal) => terminal.turnId === settlementTurnId, + ); + if ( + settlementTurnId !== latestTerminalTurnId + || !settlement + || !isCompletedTerminal(settlement.fingerprint) + ) { + throw new Error( + "persisted Codex ACPX resultless recovery requires a completed terminal turn", + ); + } + } +} + +function isCompletedTerminal(terminalFingerprint: string): boolean { + try { + const value: unknown = JSON.parse(terminalFingerprint); + return typeof value === "object" + && value !== null + && !Array.isArray(value) + && (value as Record).status === "completed"; + } catch { + return false; + } +} + +function isCompletedSemanticTerminal( + terminalFingerprint: string, + semanticFingerprint: string, +): boolean { + try { + const value: unknown = JSON.parse(terminalFingerprint); + if (typeof value !== "object" || value === null || Array.isArray(value)) { + return false; + } + const terminal = value as Record; + return ( + terminal.status === "completed" && + terminal.semanticResult === semanticFingerprint + ); + } catch { + return false; + } +} + +function boundedIdentity(value: unknown): value is string { + return ( + typeof value === "string" && + value.length > 0 && + value.length <= 240 && + !/[\u0000-\u001f\u007f]/.test(value) + ); +} + function boundedRecord(value: unknown): Record { const serialized = JSON.stringify(value); if (!serialized || Buffer.byteLength(serialized) > 64 * 1024) { diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts index 0e89953bfe..ab310259a7 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts @@ -223,6 +223,37 @@ describe("Codex ACPX runtime adapter", () => { }, ); + it("revalidates a recovered workspace immediately before provider spawn", async () => { + const runtime = fakeRuntime(); + let runtimeOptions: AcpRuntimeOptions | undefined; + const command = fakeCommand(); + const workspaceSubstituted = new Error("recovered workspace substituted"); + const assertWorkspaceHeld = vi.fn(() => { + throw workspaceSubstituted; + }); + await openCodexAcpxRuntime( + { ...openOptions(command), assertWorkspaceHeld }, + { + createRegistry: () => registry(), + createStore: () => store(), + createRuntime: (options) => { + runtimeOptions = options; + return runtime; + }, + }, + ); + + expect(() => + runtimeOptions?.spawnAgent?.({ + command: "/attacker/replacement", + args: ["--stdio"], + options: {}, + }), + ).toThrow(workspaceSubstituted); + expect(assertWorkspaceHeld).toHaveBeenCalledOnce(); + expect(command.spawn).not.toHaveBeenCalled(); + }); + it("maps status, model selection, and state-preserving close", async () => { const runtime = fakeRuntime(); vi.mocked(runtime.getStatus!).mockResolvedValue({ diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts index 3c4508be88..e17903aab4 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -221,6 +221,7 @@ export async function openCodexAcpxRuntime( // A verified provider can create descendants that inherit its launch // credential. Give the provider a dedicated POSIX process group so // cleanup authority covers that complete credential-bearing tree. + options.assertWorkspaceHeld?.(); return children.add( options.command.spawn(input.args, { ...input.options, diff --git a/packages/paperclip-runner/src/drivers/acpx/recovery-identity.ts b/packages/paperclip-runner/src/drivers/acpx/recovery-identity.ts index b5cca23154..f10836f2a7 100644 --- a/packages/paperclip-runner/src/drivers/acpx/recovery-identity.ts +++ b/packages/paperclip-runner/src/drivers/acpx/recovery-identity.ts @@ -51,7 +51,7 @@ export async function createAcpxRecoveryBinding(input: { } const workspacePath = await resolveWorkspace(input.workingDirectory); const workspaceDigest = digest(workspacePath); - const runtimeRoot = await acpxRuntimeRoot( + const runtimeRoot = await resolveAcpxRuntimeRoot( input.runtimeDirectory, input.normalizedSessionId, ); @@ -235,7 +235,7 @@ async function resolveWorkspace(value: string): Promise { return workspacePath; } -async function acpxRuntimeRoot( +export async function resolveAcpxRuntimeRoot( runtimeDirectory: string, sessionId: string, ): Promise { diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts index 22cd50e916..a42074920f 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts @@ -172,6 +172,37 @@ describe("ACPX runtime host", () => { expect(fixture.commandClose).toHaveBeenCalledOnce(); }); + it("revalidates a pinned workspace at the runtime-open boundary", async () => { + const fixture = await hostFixture(); + const openRuntime = vi.fn(async () => runtimePort()); + const workspaceSubstituted = new Error("recovered workspace substituted"); + let assertions = 0; + const assertWorkspaceHeld = vi.fn(() => { + assertions += 1; + if (assertions === 2) throw workspaceSubstituted; + }); + + await expect( + AcpxRuntimeHost.open( + { + ...fixture.options, + agent: "codex", + model: "gpt-5.6-sol", + permissionMode: "approve-reads", + environment: { + PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}", + }, + assertWorkspaceHeld, + }, + fixture.dependencies({ openRuntime }), + ), + ).rejects.toBe(workspaceSubstituted); + + expect(assertWorkspaceHeld).toHaveBeenCalledTimes(2); + expect(openRuntime).not.toHaveBeenCalled(); + expect(fixture.commandClose).toHaveBeenCalledOnce(); + }); + it("owns an authenticated semantic bridge without persisting its secret", async () => { const fixture = await hostFixture(); const handler = vi.fn(async ({ tool }) => ({ tool, ok: true })); diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts index 81e02b42c4..44d38d4249 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts @@ -84,6 +84,8 @@ export interface AcpxRuntimePortOpenOptions { permissionPolicy: ReturnType; launchEnvironment: Readonly; systemInstructions: string; + /** Revalidate a pinned recovery workspace at the provider spawn boundary. */ + assertWorkspaceHeld?: () => void; /** Abort provider admission and clean any runtime that resolves too late. */ signal?: AbortSignal; mcpServers: readonly AcpxMcpServerBinding[]; @@ -134,6 +136,8 @@ export interface OpenAcpxRuntimeHostOptions { environment?: NodeJS.ProcessEnv; managedCodexCredentialSourcePath?: string; expectedIdentity?: AcpxExpectedSessionIdentity; + /** Revalidate a pinned recovery workspace through provider admission. */ + assertWorkspaceHeld?: () => void; /** Abort admission without admitting resources that resolve afterward. */ signal?: AbortSignal; semanticTools?: AcpxSemanticToolSession; @@ -222,6 +226,7 @@ export class AcpxRuntimeHost { if (options.expectedIdentity) { verifyExpectedAcpxIdentity(options.expectedIdentity, binding, null); } + options.assertWorkspaceHeld?.(); if ( options.agent !== "codex" && options.managedCodexCredentialSourcePath !== undefined @@ -317,8 +322,9 @@ export class AcpxRuntimeHost { : null; runtime = await acquireAbortableAdmissionResource({ signal: options.signal, - acquire: () => - dependencies.openRuntime({ + acquire: () => { + options.assertWorkspaceHeld?.(); + return dependencies.openRuntime({ command: command!, profile, cwd: binding.workspacePath, @@ -330,6 +336,9 @@ export class AcpxRuntimeHost { ), launchEnvironment: sandbox.launchEnvironment, systemInstructions: boundedInstructions(options.systemInstructions), + ...(options.assertWorkspaceHeld === undefined + ? {} + : { assertWorkspaceHeld: options.assertWorkspaceHeld }), ...(options.signal === undefined ? {} : { signal: options.signal }), mcpServers: toolBridge ? [ @@ -342,7 +351,8 @@ export class AcpxRuntimeHost { ] : [], retainFailedAdmissionCleanup, - }), + }); + }, resource: "runtime", releaseLate: (lateRuntime) => lateRuntime.close({ diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.test.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.test.ts index e50f8e2bca..9b97e54ba2 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.test.ts @@ -3,10 +3,13 @@ import { lstat, mkdir, mkdtemp, + open, readFile, + rename, rm, stat, symlink, + writeFile, } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; @@ -15,7 +18,10 @@ import { afterEach, describe, expect, it } from "vitest"; import { resolveQualifiedAcpxProfile } from "./qualified-profiles.js"; import { createAcpxRecoveryBinding } from "./recovery-identity.js"; -import { prepareAcpxRuntimeSandbox } from "./runtime-sandbox.js"; +import { + prepareAcpxRuntimeSandbox, + readAcpxRecoveryWorkspace, +} from "./runtime-sandbox.js"; const temporaryDirectories: string[] = []; @@ -80,6 +86,13 @@ describe("ACPX runtime sandbox", () => { expect(await readFile(sandbox.workspaceRecordPath, "utf8")).toBe( `${fixture.binding.workspacePath}\n`, ); + const recoveryWorkspace = await readAcpxRecoveryWorkspace({ + runtimeDirectory: join(fixture.root, "runtime"), + normalizedSessionId: `sandbox-${agent}`, + }); + expect(recoveryWorkspace.path).toBe(fixture.binding.workspacePath); + expect(() => recoveryWorkspace.assertHeld()).not.toThrow(); + await recoveryWorkspace.close(); expect((await lstat(sandbox.root)).isSymbolicLink()).toBe(false); if (process.platform !== "win32") { expect((await stat(sandbox.root)).mode & 0o777).toBe(0o700); @@ -152,6 +165,120 @@ describe("ACPX runtime sandbox", () => { ).rejects.toThrow(/real directory|escaped/); }, ); + + it("rejects a malformed workspace recovery record", async () => { + const fixture = await sandboxFixture("codex"); + const sandbox = await prepareAcpxRuntimeSandbox({ + binding: fixture.binding, + agent: "codex", + }); + const handle = await open(sandbox.workspaceRecordPath, "a"); + await handle.write("extra"); + await handle.close(); + + await expect( + readAcpxRecoveryWorkspace({ + runtimeDirectory: join(fixture.root, "runtime"), + normalizedSessionId: "sandbox-codex", + }), + ).rejects.toThrow("record is invalid"); + }); + + it.runIf(process.platform !== "win32")( + "does not follow a substituted workspace recovery record", + async () => { + const fixture = await sandboxFixture("codex"); + const sandbox = await prepareAcpxRuntimeSandbox({ + binding: fixture.binding, + agent: "codex", + }); + await rm(sandbox.workspaceRecordPath); + await symlink(fixture.binding.workspacePath, sandbox.workspaceRecordPath); + + await expect( + readAcpxRecoveryWorkspace({ + runtimeDirectory: join(fixture.root, "runtime"), + normalizedSessionId: "sandbox-codex", + }), + ).rejects.toThrow("record is unavailable"); + }, + ); + + it.runIf(process.platform !== "win32")( + "does not follow a substituted recovery session directory", + async () => { + const fixture = await sandboxFixture("codex"); + const sandbox = await prepareAcpxRuntimeSandbox({ + binding: fixture.binding, + agent: "codex", + }); + const outside = join(fixture.root, "outside-recovery"); + await mkdir(outside); + await rm(sandbox.root, { recursive: true }); + await symlink(outside, sandbox.root); + + await expect( + readAcpxRecoveryWorkspace({ + runtimeDirectory: join(fixture.root, "runtime"), + normalizedSessionId: "sandbox-codex", + }), + ).rejects.toThrow("runtime directory is unavailable"); + }, + ); + + it.runIf(process.platform !== "win32")( + "rejects a recovery session directory swapped after its handle is pinned", + async () => { + const fixture = await sandboxFixture("codex"); + const sandbox = await prepareAcpxRuntimeSandbox({ + binding: fixture.binding, + agent: "codex", + }); + const displacedRoot = `${sandbox.root}-displaced`; + + await expect( + readAcpxRecoveryWorkspace( + { + runtimeDirectory: join(fixture.root, "runtime"), + normalizedSessionId: "sandbox-codex", + }, + { + afterRuntimeRootPinned: async () => { + await rename(sandbox.root, displacedRoot); + await mkdir(sandbox.root); + await writeFile( + join(sandbox.root, "workspace"), + `${fixture.binding.workspacePath}\n`, + ); + }, + }, + ), + ).rejects.toThrow("workspace record is unavailable"); + }, + ); + + it.runIf(process.platform !== "win32")( + "pins the recovered workspace until provider admission", + async () => { + const fixture = await sandboxFixture("codex"); + await prepareAcpxRuntimeSandbox({ + binding: fixture.binding, + agent: "codex", + }); + const recoveryWorkspace = await readAcpxRecoveryWorkspace({ + runtimeDirectory: join(fixture.root, "runtime"), + normalizedSessionId: "sandbox-codex", + }); + const displacedWorkspace = `${fixture.binding.workspacePath}-displaced`; + await rename(fixture.binding.workspacePath, displacedWorkspace); + await mkdir(fixture.binding.workspacePath); + + expect(() => recoveryWorkspace.assertHeld()).toThrow( + "workspace changed before provider admission", + ); + await recoveryWorkspace.close(); + }, + ); }); async function sandboxFixture(agent: "pi" | "claude" | "codex") { diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts index 968ff4dea6..384f58544c 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts @@ -1,11 +1,20 @@ import { randomBytes } from "node:crypto"; -import { constants, type Stats } from "node:fs"; +import { + constants, + fstatSync, + lstatSync, + realpathSync, + type BigIntStats, + type Stats, +} from "node:fs"; import { lstat, mkdir, open, + readFile, realpath, rename, + stat, unlink, type FileHandle, } from "node:fs/promises"; @@ -20,11 +29,15 @@ import { import { createSanitizedAcpxSpawnInput } from "./environment.js"; import type { QualifiedAcpxAgent } from "./qualified-profiles.js"; -import type { AcpxRecoveryBinding } from "./recovery-identity.js"; +import { + resolveAcpxRuntimeRoot, + type AcpxRecoveryBinding, +} from "./recovery-identity.js"; const PRIVATE_DIRECTORY_MODE = 0o700; const PRIVATE_FILE_MODE = 0o600; const MAX_SANDBOX_ENVIRONMENT_BYTES = 512 * 1024; +const MAX_WORKSPACE_RECORD_BYTES = 64 * 1024; export interface AcpxRuntimeSandbox { root: string; @@ -39,6 +52,273 @@ export interface AcpxRuntimeSandbox { persistedEnvironment: Readonly; } +/** A recovered workspace pinned until the provider process is admitted. */ +export interface AcpxRecoveryWorkspaceLease { + readonly path: string; + assertHeld(): void; + close(): Promise; +} + +export interface AcpxRecoveryWorkspaceReadDependencies { + /** Internal seam for racing a parent-directory replacement in tests. */ + afterRuntimeRootPinned?: () => Promise; +} + +/** Read the private workspace binding used to reopen one exact ACPX session. */ +export async function readAcpxRecoveryWorkspace( + input: { + runtimeDirectory: string; + normalizedSessionId: string; + }, + dependencies: AcpxRecoveryWorkspaceReadDependencies = {}, +): Promise { + const runtimeRoot = await resolveAcpxRuntimeRoot( + input.runtimeDirectory, + input.normalizedSessionId, + ); + const namespace = dirname(runtimeRoot); + let physicalNamespace: string; + let physicalRuntimeRoot: string; + try { + const [namespaceMetadata, rootMetadata] = await Promise.all([ + lstat(namespace), + lstat(runtimeRoot), + ]); + if ( + namespaceMetadata.isSymbolicLink() || + !namespaceMetadata.isDirectory() || + rootMetadata.isSymbolicLink() || + !rootMetadata.isDirectory() + ) { + throw new Error("invalid recovery directory"); + } + [physicalNamespace, physicalRuntimeRoot] = await Promise.all([ + realpath(namespace), + realpath(runtimeRoot), + ]); + } catch { + throw new Error("ACPX recovery runtime directory is unavailable"); + } + if (!isInside(physicalNamespace, physicalRuntimeRoot)) { + throw new Error("ACPX recovery runtime directory escaped its namespace"); + } + + let namespaceHandle: FileHandle | null = null; + let rootHandle: FileHandle | null = null; + let workspaceHandle: FileHandle | null = null; + try { + namespaceHandle = await openPinnedDirectory(physicalNamespace); + rootHandle = await openPinnedDirectory(physicalRuntimeRoot); + assertPinnedDirectory( + physicalNamespace, + physicalNamespace, + namespaceHandle, + "ACPX recovery namespace changed during admission", + ); + assertPinnedDirectory( + physicalRuntimeRoot, + physicalRuntimeRoot, + rootHandle, + "ACPX recovery runtime directory changed during admission", + ); + await dependencies.afterRuntimeRootPinned?.(); + + const recordPath = join(physicalRuntimeRoot, "workspace"); + let recordHandle: FileHandle; + try { + assertPinnedDirectory( + physicalRuntimeRoot, + physicalRuntimeRoot, + rootHandle, + "ACPX recovery runtime directory changed before record open", + ); + recordHandle = await open( + recordPath, + constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0), + ); + } catch { + throw new Error("ACPX recovery workspace record is unavailable"); + } + try { + assertPinnedDirectory( + physicalRuntimeRoot, + physicalRuntimeRoot, + rootHandle, + "ACPX recovery runtime directory changed during record open", + ); + const before = await recordHandle.stat({ bigint: true }); + if ( + !before.isFile() || + before.nlink !== 1n || + before.size < 2n || + before.size > BigInt(MAX_WORKSPACE_RECORD_BYTES) + ) { + throw new Error("ACPX recovery workspace record is invalid"); + } + assertPinnedFile(recordPath, before); + const bytes = await readFile(recordHandle); + const after = await recordHandle.stat({ bigint: true }); + if ( + bytes.length !== Number(before.size) || + before.dev !== after.dev || + before.ino !== after.ino || + before.size !== after.size || + before.mtimeNs !== after.mtimeNs || + before.ctimeNs !== after.ctimeNs + ) { + throw new Error("ACPX recovery workspace record changed while read"); + } + assertPinnedFile(recordPath, after); + assertPinnedDirectory( + physicalRuntimeRoot, + physicalRuntimeRoot, + rootHandle, + "ACPX recovery runtime directory changed while record was read", + ); + + const workspace = bytes.toString("utf8").replace(/\n$/, ""); + if (!workspace || /[\u0000\r\n]/.test(workspace)) { + throw new Error("ACPX recovery workspace record is invalid"); + } + let physicalWorkspace: string; + try { + physicalWorkspace = await realpath(workspace); + workspaceHandle = await openPinnedDirectory(physicalWorkspace); + } catch { + throw new Error("ACPX recovery workspace is unavailable"); + } + if (physicalWorkspace === dirname(physicalWorkspace)) { + throw new Error("ACPX recovery workspace is not a non-root directory"); + } + if (!namespaceHandle || !rootHandle || !workspaceHandle) { + throw new Error("ACPX recovery workspace handles are unavailable"); + } + const pinnedNamespace = namespaceHandle; + const pinnedRoot = rootHandle; + const pinnedWorkspace = workspaceHandle; + let closed = false; + const lease: AcpxRecoveryWorkspaceLease = { + path: physicalWorkspace, + assertHeld() { + if (closed) throw new Error("ACPX recovery workspace lease is closed"); + assertPinnedDirectory( + physicalNamespace, + physicalNamespace, + pinnedNamespace, + "ACPX recovery namespace changed before provider admission", + ); + assertPinnedDirectory( + physicalRuntimeRoot, + physicalRuntimeRoot, + pinnedRoot, + "ACPX recovery runtime directory changed before provider admission", + ); + assertPinnedDirectory( + physicalWorkspace, + physicalWorkspace, + pinnedWorkspace, + "ACPX recovery workspace changed before provider admission", + ); + }, + async close() { + if (closed) return; + closed = true; + await closeRecoveryHandles([ + pinnedWorkspace, + pinnedRoot, + pinnedNamespace, + ]); + }, + }; + lease.assertHeld(); + namespaceHandle = null; + rootHandle = null; + workspaceHandle = null; + return lease; + } finally { + await recordHandle.close(); + } + } finally { + await closeRecoveryHandles([ + workspaceHandle, + rootHandle, + namespaceHandle, + ]); + } +} + +async function openPinnedDirectory(path: string): Promise { + return await open( + path, + constants.O_RDONLY | + (constants.O_DIRECTORY ?? 0) | + (constants.O_NOFOLLOW ?? 0), + ); +} + +function assertPinnedDirectory( + path: string, + expectedPhysicalPath: string, + handle: FileHandle, + message: string, +): void { + try { + const descriptor = fstatSync(handle.fd, { bigint: true }); + const entry = lstatSync(path, { bigint: true }); + if ( + !descriptor.isDirectory() || + entry.isSymbolicLink() || + !entry.isDirectory() || + !sameBigIntFile(descriptor, entry) || + realpathSync(path) !== expectedPhysicalPath + ) { + throw new Error(message); + } + } catch (error) { + if (error instanceof Error && error.message === message) throw error; + throw new Error(message); + } +} + +function assertPinnedFile(path: string, descriptor: BigIntStats): void { + let entry: BigIntStats; + try { + entry = lstatSync(path, { bigint: true }); + } catch { + throw new Error("ACPX recovery workspace record changed while read"); + } + if ( + entry.isSymbolicLink() || + !entry.isFile() || + !sameBigIntFile(descriptor, entry) + ) { + throw new Error("ACPX recovery workspace record changed while read"); + } +} + +function sameBigIntFile(left: BigIntStats, right: BigIntStats): boolean { + return left.dev === right.dev && left.ino === right.ino; +} + +async function closeRecoveryHandles( + handles: readonly (FileHandle | null)[], +): Promise { + const results = await Promise.allSettled( + handles + .filter((handle): handle is FileHandle => handle !== null) + .map((handle) => handle.close()), + ); + const errors = results + .filter( + (result): result is PromiseRejectedResult => + result.status === "rejected", + ) + .map((result) => result.reason); + if (errors.length > 0) { + throw new AggregateError(errors, "Failed to close ACPX recovery handles"); + } +} + /** Prepare the private filesystem and environment visible to an ACPX agent. */ export async function prepareAcpxRuntimeSandbox(input: { binding: AcpxRecoveryBinding; diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 3cb1e0521e..873f178918 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -151,6 +151,20 @@ function controlEvent( }; } +function canonicalTestJson(value: unknown): string { + if (Array.isArray(value)) { + return `[${value.map(canonicalTestJson).join(",")}]`; + } + if (typeof value === "object" && value !== null) { + const record = value as Record; + return `{${Object.keys(record) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonicalTestJson(record[key])}`) + .join(",")}}`; + } + return JSON.stringify(value) ?? "undefined"; +} + function runnerEvent( sourceSeq: number, eventType: PrpEvent["eventType"], @@ -4888,7 +4902,8 @@ describe("executeNativeSession recovery", () => { expect(completeRun).toHaveBeenCalledOnce(); }); - it("recovers a completed checkpoint and appends only a missing control terminal fact", async () => { + it("keeps a reconstructed semantic result on its matched terminal turn", async () => { + const semanticFingerprint = canonicalTestJson(result); const checkpoint: PersistedNativeSession = { backendKind: "mock", sessionId: "driver-recovery", @@ -4897,14 +4912,27 @@ describe("executeNativeSession recovery", () => { cursor: "4", semanticResult: result, terminal, - activeTurnId: "turn-recovery", + activeTurnId: null, terminalTurns: [ - { turnId: "turn-recovery", fingerprint: "terminal-fingerprint" }, + { + turnId: "turn-with-result", + fingerprint: JSON.stringify({ + status: "completed", + semanticResult: semanticFingerprint, + }), + }, + { + turnId: "turn-later-failed", + fingerprint: JSON.stringify({ status: "failed" }), + }, ], pendingRuntimeRequests: [], lineage: [], }; - const events = [controlEvent(1, "run.result.accepted", { result })]; + const events = [{ + ...controlEvent(1, "run.result.accepted", { result }), + turnId: "turn-with-result", + }]; const checkpoints: PersistedNativeSession[] = []; const completeRun = vi.fn(async () => undefined); const startTurn = vi.fn(async () => ({ turnId: "unexpected-turn" })); @@ -4997,7 +5025,15 @@ describe("executeNativeSession recovery", () => { "run.terminal", ]); expect(events.map((event) => event.sourceSeq)).toEqual([1, 2]); + expect(events.map((event) => event.turnId)).toEqual([ + "turn-with-result", + "turn-with-result", + ]); expect(completeRun).toHaveBeenCalledOnce(); + expect(completeRun).toHaveBeenCalledWith( + expect.objectContaining({ turnId: "turn-with-result" }), + expect.anything(), + ); expect(completed).toMatchObject({ nativeEventCount: 1, highestContiguousSourceSeq: 2, diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index cf608b941d..1d1239c63c 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -1540,7 +1540,9 @@ export async function executeNativeSession( ? { result: completionSnapshot.semanticResult, terminal: completionSnapshot.terminal, - turnId: completionSnapshot.activeTurnId ?? null, + turnId: + completionSnapshot.activeTurnId ?? + completedSemanticResultTurnId(completionSnapshot), } : null; if (!completed) { @@ -1947,3 +1949,30 @@ function canonicalJson(value: unknown): string { } return JSON.stringify(value) ?? "undefined"; } + +function completedSemanticResultTurnId( + snapshot: PersistedNativeSession, +): string | null { + if (snapshot.semanticResult === undefined || snapshot.semanticResult === null) { + return null; + } + const semanticFingerprint = canonicalJson(snapshot.semanticResult); + for (const terminal of [...(snapshot.terminalTurns ?? [])].reverse()) { + try { + const value: unknown = JSON.parse(terminal.fingerprint); + if ( + typeof value === "object" + && value !== null + && !Array.isArray(value) + && (value as Record).status === "completed" + && (value as Record).semanticResult + === semanticFingerprint + ) { + return terminal.turnId; + } + } catch { + // Legacy terminal fingerprints cannot prove semantic-result ownership. + } + } + return null; +}