feat(runner): recover settled Codex ACPX sessions (#12407)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The runner package keeps provider sessions durable across process restarts. > - The merged Codex ACPX backend can admit a new qualified session through the package factory. > - It cannot yet recover a settled session from a durable checkpoint. > - Recovery must prove the same provider identity, runtime root, workspace, and latest terminal settlement before it resumes. > - ACPX cannot yet prove continuity for a provider turn that was active when the process stopped. > - This pull request recovers only settled Codex ACPX checkpoints and rejects ambiguous or stale state. > - The benefit is fail-closed restart recovery without repeated work, silent session replacement, or recovery of an older semantic result. ## Linked Issues or Issue Description Refs #12406 This pull request builds on the Codex ACPX backend route merged in #12406. It adds only settled-session recovery and the package-local durability and filesystem checks that recovery requires. ## What Changed - Add settled-session recovery to the Codex ACPX harness driver and advertise resume support for that bounded path. - Keep provider recovery on the same persisted ACPX session. Do not permit a replacement provider session. - Require persisted run, normalized session, provider, profile, workspace, permission, result, terminal, and recovery-policy identities to agree. - Reopen the provider with the exact persisted ACPX identity and reject provider or qualification drift. - Restore the last source sequence, terminal history, semantic result, provider recovery policy, and normalized terminal fact before new work can start. - Clear a stale active-turn marker only when durable terminal history proves that exact turn settled. - Reject active-turn recovery when durable history cannot prove provider-turn continuity. - Require a semantic result to belong to the latest completed terminal settlement. - Reject an older completed semantic result when a later identical retry failed or was interrupted. - Reject resultless recovery unless the latest terminal turn completed successfully. - Bound terminal history and semantic-result recovery by entry count, aggregate bytes, fingerprint bytes, node count, and nesting depth before cloning or serializing persisted data. - Read the private workspace record through pinned namespace, runtime-root, record, and workspace identities. - Reject symbolic links, substituted directories, replaced records, invalid files, root workspaces, and workspace changes during admission. - Keep the recovery workspace lease pinned and revalidate it at the host and provider-spawn boundaries. - Retain and close a workspace lease that resolves after cancellation. Close a partially recovered provider session if lease cleanup fails. - Stage a valid isolated Codex credential in the runtime-open-boundary regression so the test reaches and proves the intended workspace revalidation fence. - Add focused recovery tests for settled checkpoints, terminal ownership, stale semantic results, active turns, identity drift, bounded persisted data, cancellation, directory substitution, record replacement, and spawn-boundary workspace replacement. ## Verification - Replay base: `30ef14edd4e7290d9eac43ca7b7835611933cc74` (`master` after #12406 merged). - Exact replay head: `3d1b16d5cda2dd72f83785c921a025cc57aab8d6`. - Stable patch ID for the exact replay delta: `10742d7db00f801b5723e7a8582c637261ceb0dd`. The source-stack combined patch was `4d7c9de72a0a30f6e33c935c092b745fe414c588`; the replay preserves the newer unconditional provider process-group containment from `master` while adding recovery workspace revalidation before spawn, plus the exact-head credential fixture correction exposed by GitHub Actions. - The exact pull request delta contains exactly 14 files: - `packages/paperclip-runner/src/backends/harness-driver-backend.ts` - `packages/paperclip-runner/src/backends/harness-driver-backend.test.ts` - `packages/paperclip-runner/src/backends/native-backend-factory.test.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts` - `packages/paperclip-runner/src/drivers/acpx/recovery-identity.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-host.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-sandbox.test.ts` - `packages/paperclip-runner/src/native-session-runtime.ts` - `packages/paperclip-runner/src/native-session-runtime.test.ts` - The exact combined delta is 2,279 additions and 73 deletions. - This delta does not change dependencies, `pnpm-lock.yaml`, workspace configuration, workflows, migrations, server selection, UI behavior, or public package exports. - Focused GitHub recovery tests: **PASS**. The exact-head Build job passed the harness backend, native runtime, Codex ACPX driver, runtime host, runtime sandbox, runtime adapter, and backend factory suites as part of all 751 TypeScript runner tests. - GitHub Actions: **PASS** for exact head `3d1b16d5cda2dd72f83785c921a025cc57aab8d6`. Every applicable job passed. The Storybook visual regression skipped intentionally because this package-only delta does not touch UI or Storybook paths. - Security checks: **PASS** for the exact head. Superagent, Snyk, both Socket checks, and contributor trust completed successfully. - Greptile: **5/5** for the exact head, with no open P1/P2 findings, recommendations, or follow-ups. - No local test result is claimed. GitHub Actions is the authoritative verification environment for the replayed revision. ## Risks This change has medium package-local risk because it admits persisted state back into a provider session. Recovery fails closed when an active provider turn has no matching terminal fingerprint, when a semantic result is not owned by the latest completed settlement, when any durable identity changes, or when the workspace cannot remain pinned through provider spawn. Recovery never selects a replacement provider session. Bounds prevent oversized or deeply nested persisted state from consuming unbounded work before validation. The workspace lease prevents path substitution during recovery admission. Existing direct adapters and the Codex app-server driver do not use this ACPX recovery route. This change does not add a migration, dependency, lockfile update, workflow, server route, UI surface, public export, or production rollout flag. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI Codex with GPT-5.6, extended reasoning, repository tool use, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
30ef14edd4
commit
96421b0663
|
|
@ -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<PersistedHarnessSession> {
|
||||
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<string, unknown>;
|
||||
return `{${Object.keys(record)
|
||||
.sort()
|
||||
.map((key) => `${JSON.stringify(key)}:${canonicalTestJson(record[key])}`)
|
||||
.join(",")}}`;
|
||||
}
|
||||
return JSON.stringify(value) ?? "undefined";
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<object>();
|
||||
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<string, unknown>) {
|
||||
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<string, unknown>)[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,
|
||||
|
|
|
|||
|
|
@ -134,7 +134,7 @@ describe("native backend factory", () => {
|
|||
name: "acpx_runtime",
|
||||
version: "0.13.1",
|
||||
capabilities: {
|
||||
resume: false,
|
||||
resume: true,
|
||||
interruption: true,
|
||||
dynamicTools: true,
|
||||
},
|
||||
|
|
|
|||
|
|
@ -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<AcpxRecoveryWorkspaceLease>();
|
||||
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<ReturnType<typeof fakeHost>>();
|
||||
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<CodexAcpxDriverDependencies["openHost"]>;
|
||||
readRecoveryWorkspace?: NonNullable<
|
||||
CodexAcpxDriverDependencies["readRecoveryWorkspace"]
|
||||
>;
|
||||
} = {},
|
||||
): {
|
||||
driver: CodexAcpxDriver;
|
||||
host: ReturnType<typeof fakeHost>;
|
||||
openHost: ReturnType<typeof vi.fn>;
|
||||
readRecoveryWorkspace: ReturnType<typeof vi.fn>;
|
||||
hostOptions: OpenAcpxRuntimeHostOptions | null;
|
||||
finishTurn(result: Awaited<AcpxRuntimeTurn["result"]>): void;
|
||||
} {
|
||||
const result = deferred<Awaited<AcpxRuntimeTurn["result"]>>();
|
||||
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<typeof deferred<Awaited<AcpxRuntimeTurn["result"]>>> | null = null;
|
||||
const createTurn = (): AcpxRuntimeTurn => {
|
||||
activeResult = deferred<Awaited<AcpxRuntimeTurn["result"]>>();
|
||||
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<typeof vi.fn>;
|
||||
close: ReturnType<typeof vi.fn>;
|
||||
} {
|
||||
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();
|
||||
|
|
|
|||
|
|
@ -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<AcpxRecoveryWorkspaceLease>;
|
||||
}
|
||||
|
||||
/** 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<Promise<void>>();
|
||||
readonly #quarantinedHostCleanups = new Set<QuarantinedHostCleanup>();
|
||||
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<HarnessDriverDescriptor> {
|
||||
|
|
@ -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<HarnessSessionRecoveryResult> {
|
||||
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<CodexAcpxDriverDependencies["readRecoveryWorkspace"]>
|
||||
>[0],
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<AcpxRecoveryWorkspaceLease> {
|
||||
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<HarnessSession> {
|
||||
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<string, string>();
|
||||
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<string, unknown> | null = null;
|
||||
#assistantText = "";
|
||||
#closed = false;
|
||||
|
|
@ -538,6 +664,7 @@ class CodexAcpxSession implements HarnessSession {
|
|||
terminalEventReserve: number;
|
||||
retainCleanup: (cleanup: Promise<void>) => 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<string>();
|
||||
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<string, unknown>).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<string, unknown>;
|
||||
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<string, unknown> {
|
||||
const serialized = JSON.stringify(value);
|
||||
if (!serialized || Buffer.byteLength(serialized) > 64 * 1024) {
|
||||
|
|
|
|||
|
|
@ -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({
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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<string> {
|
|||
return workspacePath;
|
||||
}
|
||||
|
||||
async function acpxRuntimeRoot(
|
||||
export async function resolveAcpxRuntimeRoot(
|
||||
runtimeDirectory: string,
|
||||
sessionId: string,
|
||||
): Promise<string> {
|
||||
|
|
|
|||
|
|
@ -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 }));
|
||||
|
|
|
|||
|
|
@ -84,6 +84,8 @@ export interface AcpxRuntimePortOpenOptions {
|
|||
permissionPolicy: ReturnType<typeof acpxRuntimePermissionPolicy>;
|
||||
launchEnvironment: Readonly<NodeJS.ProcessEnv>;
|
||||
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({
|
||||
|
|
|
|||
|
|
@ -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") {
|
||||
|
|
|
|||
|
|
@ -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<NodeJS.ProcessEnv>;
|
||||
}
|
||||
|
||||
/** A recovered workspace pinned until the provider process is admitted. */
|
||||
export interface AcpxRecoveryWorkspaceLease {
|
||||
readonly path: string;
|
||||
assertHeld(): void;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
export interface AcpxRecoveryWorkspaceReadDependencies {
|
||||
/** Internal seam for racing a parent-directory replacement in tests. */
|
||||
afterRuntimeRootPinned?: () => Promise<void>;
|
||||
}
|
||||
|
||||
/** Read the private workspace binding used to reopen one exact ACPX session. */
|
||||
export async function readAcpxRecoveryWorkspace(
|
||||
input: {
|
||||
runtimeDirectory: string;
|
||||
normalizedSessionId: string;
|
||||
},
|
||||
dependencies: AcpxRecoveryWorkspaceReadDependencies = {},
|
||||
): Promise<AcpxRecoveryWorkspaceLease> {
|
||||
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<FileHandle> {
|
||||
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<void> {
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>;
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>).status === "completed"
|
||||
&& (value as Record<string, unknown>).semanticResult
|
||||
=== semanticFingerprint
|
||||
) {
|
||||
return terminal.turnId;
|
||||
}
|
||||
} catch {
|
||||
// Legacy terminal fingerprints cannot prove semantic-result ownership.
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue