mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-26 18:31:34 +01:00
fix(runner): isolate Codex credentials and hostile output
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
6391608e8f
commit
8d4ffe7f2b
10 files changed
+9224
-5498
No files matched your search
@@ -60,12 +60,19 @@ instruction blocks. The model input accepts only text, never a Codex `skill`
|
||||
input. The driver captures the returned instruction-source list and requires it
|
||||
to be empty for the skillless assertion.
|
||||
|
||||
The child process environment is allowlisted. It may receive only local runtime
|
||||
keys such as `PATH`, `HOME`, `CODEX_HOME`, locale, temporary-directory, trusted
|
||||
certificate, and credential-free proxy settings. Paperclip bearer values,
|
||||
`OPENAI_API_KEY`, arbitrary skill paths, and other inherited variables are not
|
||||
passed. Diagnostics redact bearer/basic credentials, credentialed proxy URLs,
|
||||
secret query parameters, sensitive JSON keys, and common key assignments.
|
||||
The trusted app-server process has an allowlisted environment. It retains host
|
||||
`HOME` and `CODEX_HOME` only so the provider can authenticate. Model-issued
|
||||
commands have a separate boundary: an empty-by-default environment with no
|
||||
`HOME` or `CODEX_HOME`, no network, and a named Codex
|
||||
permission profile with read-only minimal runtime files, write access only to
|
||||
the assigned workspace, and no access to the rest of the filesystem. The
|
||||
driver refuses filesystem-root workspaces, workspaces containing host `HOME`,
|
||||
and any workspace overlapping host `CODEX_HOME`.
|
||||
|
||||
Paperclip bearer values, `OPENAI_API_KEY`, arbitrary skill paths, and other
|
||||
inherited variables are not passed. Diagnostics redact bearer/basic
|
||||
credentials, credentialed proxy URLs, secret query parameters, sensitive JSON
|
||||
keys, and common key assignments.
|
||||
|
||||
The context snapshot records configuration and environment **key names**, not
|
||||
secret values.
|
||||
@@ -83,10 +90,17 @@ Two dynamic semantic tools are registered when supported:
|
||||
reason, and scope.
|
||||
|
||||
Both normalize through the canonical `paperclip.run_result.v1` validator. The
|
||||
first valid result is committed. Repeating the same result is idempotent; a
|
||||
different later result is rejected. A completed turn with no valid semantic
|
||||
result produces one explicit `needs_review` result and one failed terminal
|
||||
event. Process exit or prose alone never implies completion.
|
||||
first valid result is proposed. An exact retry from the same call is
|
||||
idempotent; a different call or result is rejected. Tool calls and provider
|
||||
notifications must name the exact opened thread and active turn. Missing,
|
||||
pre-turn, cross-thread, cross-turn, and post-terminal bindings fail the provider
|
||||
session closed. Duplicate or conflicting terminal facts are rejected. Process
|
||||
exit or prose alone never implies completion.
|
||||
|
||||
The provider cannot commit controller state. It emits `run.result.proposed` and
|
||||
a provider turn terminal; the mock core validates the proposal against its
|
||||
task envelope, emits `run.result.accepted` or `run.result.rejected`, and alone
|
||||
emits `run.terminal`.
|
||||
|
||||
## Canonical event mapping
|
||||
|
||||
@@ -101,17 +115,29 @@ event. Process exit or prose alone never implies completion.
|
||||
`runtime_request.resolved` with redacted detail;
|
||||
- token snapshots -> completed `usage` items;
|
||||
- semantic verification rows -> completed `verification` items;
|
||||
- completion -> one `run.result.proposed` and one `run.terminal`.
|
||||
- provider completion -> at most one `run.result.proposed` and one turn
|
||||
terminal;
|
||||
- controller decision -> one `run.result.accepted` or `run.result.rejected`,
|
||||
followed by one `run.terminal`.
|
||||
|
||||
The existing Phase 1 reducer consumes the live stream and the replay stream.
|
||||
The Phase 4 tracer requires byte-equivalent snapshots, contiguous source
|
||||
sequences, stable item IDs, one result, and one terminal event.
|
||||
The JSON-RPC transport limits each input line, pending client requests,
|
||||
in-flight server requests, queued notification count and bytes, diagnostic
|
||||
lines, and retained provider payloads. Malformed or oversized messages close
|
||||
the transport and reject pending work.
|
||||
|
||||
The existing Phase 1 reducer consumes the live stream. Replay crosses a
|
||||
serialized JSONL boundary that validates byte and event counts, line size,
|
||||
schema, run/session binding, unique source event IDs, continuous per-source
|
||||
sequence, and exactly one final run terminal before reducing. The Phase 4
|
||||
tracer requires byte-equivalent live and replay snapshots.
|
||||
|
||||
## Runnable example
|
||||
|
||||
`trace:phase4` starts a real local `codex app-server` session through the mock
|
||||
core. Its safe task can create only `hello.txt`, disables network access, and
|
||||
checks the exact file contents. See the
|
||||
core. Its safe task creates `hello.txt` with network disabled. The evidence
|
||||
recorder additionally probes a readable host Codex credential/config path and
|
||||
an unrelated host secret, requiring both reads and writes to be denied while
|
||||
workspace output and app-server authentication still succeed. See the
|
||||
[Phase 4 tutorial](tutorials/phase-04-skillless-codex.md).
|
||||
|
||||
This phase changes no browser surface, so no new browser screenshot applies.
|
||||
|
||||
@@ -34,6 +34,7 @@ codex app-server --help
|
||||
```sh
|
||||
pnpm --filter @paperclipai/paperclip-runner typecheck:typescript
|
||||
pnpm --filter @paperclipai/paperclip-runner exec vitest run \
|
||||
src/drivers/codex/app-server-transport.test.ts \
|
||||
src/drivers/codex/codex-app-server-driver.test.ts
|
||||
pnpm --filter @paperclipai/paperclip-runner check:forbidden-imports
|
||||
```
|
||||
@@ -91,6 +92,10 @@ Confirm:
|
||||
- `modelInputKinds` contains only `text`;
|
||||
- the semantic tools are `paperclip_finish` and `paperclip_block`;
|
||||
- no environment **value** is present;
|
||||
- the permission profile denies root access, exposes only read-only minimal
|
||||
runtime files, grants write access to the assigned workspace, and disables
|
||||
network access;
|
||||
- command `HOME` and `CODEX_HOME` are absent;
|
||||
- the envelope contains only this safe task and its completion criteria.
|
||||
|
||||
Check that no control-plane route or bearer credential appears:
|
||||
@@ -122,6 +127,17 @@ The focused test suite supplies deterministic file-change and runtime-request
|
||||
events because a real model may choose a shell command instead of a file patch
|
||||
and may not need human input for this small task.
|
||||
|
||||
To record the authenticated isolation proof, run:
|
||||
|
||||
```sh
|
||||
pnpm --filter @paperclipai/paperclip-runner record:phase4
|
||||
```
|
||||
|
||||
The recorder selects an existing readable host Codex credential/config path,
|
||||
creates a separate host secret, and asks the model to prove that neither is
|
||||
readable or writable. The run must still authenticate, create the workspace
|
||||
files, retain the successful command output, and pass every trace assertion.
|
||||
|
||||
## Step 5: Try steering
|
||||
|
||||
Use a new directory:
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -1,6 +1,7 @@
|
||||
#!/usr/bin/env node
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { constants } from "node:fs";
|
||||
import { access, mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { dirname, join, resolve } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
@@ -12,6 +13,32 @@ const scratchRoot =
|
||||
tmpdir();
|
||||
await mkdir(scratchRoot, { recursive: true });
|
||||
const workspace = await mkdtemp(join(scratchRoot, "paperclip-runner-phase4-"));
|
||||
const hostileHostDirectory = await mkdtemp(join(scratchRoot, "paperclip-runner-phase4-host-"));
|
||||
const unrelatedSecretPath = join(hostileHostDirectory, "unrelated-secret.txt");
|
||||
await writeFile(unrelatedSecretPath, "must remain outside the model sandbox", { mode: 0o600 });
|
||||
const hostHome = process.env.HOME;
|
||||
if (!hostHome) throw new Error("HOME is required to locate the host Codex configuration");
|
||||
const codexHome = resolve(process.env.CODEX_HOME ?? join(hostHome, ".codex"));
|
||||
const credentialCandidates = [join(codexHome, "auth.json"), join(codexHome, "config.toml"), codexHome];
|
||||
let codexCredentialPath;
|
||||
for (const candidate of credentialCandidates) {
|
||||
try {
|
||||
await access(candidate, constants.R_OK);
|
||||
codexCredentialPath = candidate;
|
||||
break;
|
||||
} catch {
|
||||
// Keep looking for a host path that proves the former read path was reachable.
|
||||
}
|
||||
}
|
||||
if (!codexCredentialPath) {
|
||||
throw new Error("No readable host Codex credential/config path was available for the isolation probe");
|
||||
}
|
||||
const deniedHostPaths = [...new Set([
|
||||
resolve(hostHome),
|
||||
codexHome,
|
||||
codexCredentialPath,
|
||||
unrelatedSecretPath,
|
||||
])];
|
||||
const rawTracePath = join(workspace, "trace.raw.json");
|
||||
const evidencePath = join(
|
||||
packageRoot,
|
||||
@@ -27,8 +54,13 @@ try {
|
||||
workspace,
|
||||
"--output",
|
||||
rawTracePath,
|
||||
...deniedHostPaths.flatMap((path) => ["--deny-read-write-path", path]),
|
||||
],
|
||||
{ cwd: packageRoot, env: process.env, stdio: "inherit" },
|
||||
{
|
||||
cwd: packageRoot,
|
||||
env: { ...process.env, PAPERCLIP_WORKSPACE_CWD: workspace },
|
||||
stdio: "inherit",
|
||||
},
|
||||
);
|
||||
if (run.status !== 0) {
|
||||
throw new Error(`Phase 4 tracer exited with status ${run.status ?? "unknown"}`);
|
||||
@@ -42,8 +74,35 @@ try {
|
||||
if (hello !== "hello from phase 4") {
|
||||
throw new Error("Phase 4 safe task produced unexpected hello.txt content");
|
||||
}
|
||||
const isolationProof = await readFile(join(workspace, "isolation-ok.txt"), "utf8");
|
||||
if (isolationProof !== "isolation enforced") {
|
||||
throw new Error("Phase 4 isolation probe did not produce the expected proof file");
|
||||
}
|
||||
const commandObserved = trace.events?.some((event) =>
|
||||
event.eventType === "item.started" &&
|
||||
event.payload?.kind === "commandExecution" &&
|
||||
deniedHostPaths.every((path) => JSON.stringify(event.payload).includes(path))
|
||||
);
|
||||
const proofOutputObserved = trace.events?.some((event) =>
|
||||
(event.eventType === "item.delta" || event.eventType === "item.completed") &&
|
||||
event.payload?.kind === "commandExecution" &&
|
||||
(
|
||||
String(event.payload?.text).includes("isolation enforced") ||
|
||||
String(event.payload?.item?.aggregatedOutput).includes("isolation enforced")
|
||||
)
|
||||
);
|
||||
if (!commandObserved || !proofOutputObserved) {
|
||||
throw new Error("Phase 4 trace did not retain the successful hostile-path isolation command");
|
||||
}
|
||||
|
||||
const replacements = [[workspace, "$PHASE4_WORKSPACE"]];
|
||||
const replacements = [
|
||||
[workspace, "$PHASE4_WORKSPACE"],
|
||||
[codexCredentialPath, "$HOST_CODEX_CREDENTIAL_PATH"],
|
||||
[codexHome, "$HOST_CODEX_HOME"],
|
||||
[unrelatedSecretPath, "$HOST_UNRELATED_SECRET"],
|
||||
[hostileHostDirectory, "$HOST_SECRET_DIRECTORY"],
|
||||
[resolve(hostHome), "$HOST_HOME"],
|
||||
];
|
||||
const writableRoots = trace.context?.sandbox?.writableRoots;
|
||||
if (Array.isArray(writableRoots)) {
|
||||
writableRoots.forEach((root, index) => {
|
||||
@@ -57,6 +116,7 @@ try {
|
||||
process.stdout.write(`Recorded ${evidencePath}\n`);
|
||||
} finally {
|
||||
await rm(workspace, { recursive: true, force: true });
|
||||
await rm(hostileHostDirectory, { recursive: true, force: true });
|
||||
}
|
||||
|
||||
function normalizeStrings(value, replacements) {
|
||||
|
||||
@@ -13,10 +13,11 @@ interface CliOptions {
|
||||
steer?: string;
|
||||
interrupt: boolean;
|
||||
json: boolean;
|
||||
denyReadWritePaths: string[];
|
||||
}
|
||||
|
||||
function parseArgs(args: string[]): CliOptions {
|
||||
const options: CliOptions = { interrupt: false, json: false };
|
||||
const options: CliOptions = { interrupt: false, json: false, denyReadWritePaths: [] };
|
||||
for (let index = 0; index < args.length; index += 1) {
|
||||
const value = args[index];
|
||||
if (value === "--") continue;
|
||||
@@ -25,11 +26,22 @@ function parseArgs(args: string[]): CliOptions {
|
||||
else if (value === "--working-directory") options.workingDirectory = args[++index];
|
||||
else if (value === "--output") options.output = args[++index];
|
||||
else if (value === "--steer") options.steer = args[++index];
|
||||
else if (value === "--deny-read-write-path") {
|
||||
const path = args[++index];
|
||||
if (path === undefined || path.trim().length === 0) {
|
||||
throw new Error("--deny-read-write-path requires a path");
|
||||
}
|
||||
options.denyReadWritePaths.push(resolve(path));
|
||||
}
|
||||
else throw new Error(`Unknown argument: ${value}`);
|
||||
}
|
||||
return options;
|
||||
}
|
||||
|
||||
function shellQuote(value: string): string {
|
||||
return `'${value.replaceAll("'", `'"'"'`)}'`;
|
||||
}
|
||||
|
||||
function formatSummary(trace: Awaited<ReturnType<typeof runPhase4CodexTracer>>): string {
|
||||
const checks = Object.entries(trace.assertions)
|
||||
.map(([name, passed]) => `${passed ? "PASS" : "FAIL"} ${name}`)
|
||||
@@ -53,18 +65,44 @@ async function main(): Promise<void> {
|
||||
? resolve(options.workingDirectory)
|
||||
: await mkdtemp(join(tmpdir(), "paperclip-phase4-safe-"));
|
||||
await mkdir(workingDirectory, { recursive: true });
|
||||
const isolationCommand = options.denyReadWritePaths.length === 0
|
||||
? null
|
||||
: [
|
||||
...options.denyReadWritePaths.flatMap((path) => [
|
||||
`test ! -r ${shellQuote(path)}`,
|
||||
`test ! -w ${shellQuote(path)}`,
|
||||
]),
|
||||
"test -z \"${CODEX_HOME+x}\"",
|
||||
"test -z \"${HOME+x}\"",
|
||||
"printf 'isolation enforced' > isolation-ok.txt",
|
||||
"printf 'isolation enforced\\n'",
|
||||
].join(" && ");
|
||||
const envelope = createPhase4TaskEnvelope({
|
||||
objective: "Create hello.txt in the working directory with exactly: hello from phase 4",
|
||||
objective: isolationCommand === null
|
||||
? "Create hello.txt in the working directory with exactly: hello from phase 4"
|
||||
: "Prove filesystem isolation, then create the requested files in the working directory.",
|
||||
criteria: [
|
||||
{ id: "file-created", requirement: "hello.txt exists in the supplied working directory." },
|
||||
{ id: "exact-content", requirement: "hello.txt contains exactly 'hello from phase 4'." },
|
||||
...(isolationCommand === null
|
||||
? []
|
||||
: [{
|
||||
id: "filesystem-isolated",
|
||||
requirement:
|
||||
"Run the supplied isolation command exactly; it must print 'isolation enforced' and create isolation-ok.txt with that exact content.",
|
||||
}]),
|
||||
],
|
||||
constraints: [
|
||||
"Write only hello.txt inside the supplied working directory.",
|
||||
isolationCommand === null
|
||||
? "Write only hello.txt inside the supplied working directory."
|
||||
: "Write only hello.txt and isolation-ok.txt inside the supplied working directory.",
|
||||
"Do not use network access.",
|
||||
"Do not discover or invoke skills.",
|
||||
"Do not call a control-plane API.",
|
||||
"Return one semantic completion result.",
|
||||
...(isolationCommand === null
|
||||
? []
|
||||
: [`Run this exact shell command before all other work: ${isolationCommand}`]),
|
||||
...(options.steer !== undefined || options.interrupt
|
||||
? ["Before changing files, run the shell command 'sleep 3' once so the operator can send a control command."]
|
||||
: []),
|
||||
|
||||
@@ -0,0 +1,70 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
import { ProcessCodexAppServerTransport } from "./app-server-transport.js";
|
||||
|
||||
function nodeTransport(
|
||||
source: string,
|
||||
options: ConstructorParameters<typeof ProcessCodexAppServerTransport>[0] = {},
|
||||
) {
|
||||
return new ProcessCodexAppServerTransport({
|
||||
...options,
|
||||
command: process.execPath,
|
||||
args: ["-e", source],
|
||||
environment: { PATH: process.env.PATH },
|
||||
});
|
||||
}
|
||||
|
||||
describe("Codex app-server transport limits", () => {
|
||||
it("rejects an oversized line before buffering the complete hostile payload", async () => {
|
||||
const diagnostics: string[] = [];
|
||||
const transport = nodeTransport(
|
||||
"setTimeout(() => process.stdout.write('x'.repeat(1048576)), 50); setInterval(() => {}, 1000)",
|
||||
{ maxLineBytes: 128, onDiagnostic: (message) => diagnostics.push(message) },
|
||||
);
|
||||
await expect(transport.notifications()[Symbol.asyncIterator]().next()).rejects.toThrow(
|
||||
"codex app-server line exceeded 128 bytes",
|
||||
);
|
||||
expect(diagnostics).toEqual(["codex app-server line exceeded 128 bytes"]);
|
||||
await transport.close();
|
||||
});
|
||||
|
||||
it("bounds pending requests", async () => {
|
||||
const transport = nodeTransport("process.stdin.resume()", { maxPendingRequests: 1 });
|
||||
const first = transport.request("first", {});
|
||||
await expect(transport.request("second", {})).rejects.toThrow(
|
||||
"codex app-server pending request limit 1 exceeded",
|
||||
);
|
||||
await transport.close();
|
||||
await expect(first).rejects.toThrow("codex app-server transport closed");
|
||||
});
|
||||
|
||||
it("fails closed when queued notifications exceed their count bound", async () => {
|
||||
let transport: ProcessCodexAppServerTransport;
|
||||
const diagnostic = new Promise<string>((resolve) => {
|
||||
const lines = [1, 2].map((id) => JSON.stringify({
|
||||
method: "item/completed",
|
||||
params: { threadId: "thread-1", turnId: "turn-1", item: { id } },
|
||||
})).join("\n");
|
||||
transport = nodeTransport(
|
||||
`setTimeout(() => process.stdout.write(${JSON.stringify(`${lines}\n`)}), 50); setInterval(() => {}, 1000)`,
|
||||
{
|
||||
maxQueuedNotifications: 1,
|
||||
maxQueuedNotificationBytes: 1024,
|
||||
onDiagnostic: (message) => resolve(message),
|
||||
},
|
||||
);
|
||||
});
|
||||
await expect(diagnostic).resolves.toBe("codex app-server notification queue limit exceeded");
|
||||
await transport!.close();
|
||||
});
|
||||
|
||||
it("rejects malformed JSON-RPC messages", async () => {
|
||||
const transport = nodeTransport(
|
||||
"setTimeout(() => process.stdout.write('{not-json}\\n'), 50); setInterval(() => {}, 1000)",
|
||||
);
|
||||
await expect(transport.request("pending", {})).rejects.toThrow(
|
||||
"codex app-server emitted malformed JSON",
|
||||
);
|
||||
await transport.close();
|
||||
});
|
||||
});
|
||||
@@ -1,5 +1,4 @@
|
||||
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
|
||||
import { createInterface } from "node:readline";
|
||||
|
||||
export interface CodexRpcNotification {
|
||||
method: string;
|
||||
@@ -22,35 +21,117 @@ export interface CodexAppServerTransport {
|
||||
close(): Promise<void>;
|
||||
}
|
||||
|
||||
class AsyncQueue<T> implements AsyncIterable<T> {
|
||||
#values: T[] = [];
|
||||
#waiters: Array<(value: IteratorResult<T>) => void> = [];
|
||||
class BoundedAsyncQueue<T> implements AsyncIterable<T> {
|
||||
#values: Array<{ value: T; bytes: number }> = [];
|
||||
#waiters: Array<{
|
||||
resolve: (value: IteratorResult<T>) => void;
|
||||
reject: (error: Error) => void;
|
||||
}> = [];
|
||||
#queuedBytes = 0;
|
||||
#closed = false;
|
||||
#error: Error | null = null;
|
||||
|
||||
push(value: T): void {
|
||||
if (this.#closed) return;
|
||||
constructor(
|
||||
private readonly maxValues: number,
|
||||
private readonly maxBytes: number,
|
||||
) {}
|
||||
|
||||
push(value: T, bytes: number): boolean {
|
||||
if (this.#closed) return false;
|
||||
const waiter = this.#waiters.shift();
|
||||
if (waiter === undefined) this.#values.push(value);
|
||||
else waiter({ value, done: false });
|
||||
if (waiter !== undefined) {
|
||||
waiter.resolve({ value, done: false });
|
||||
return true;
|
||||
}
|
||||
if (this.#values.length >= this.maxValues || this.#queuedBytes + bytes > this.maxBytes) {
|
||||
return false;
|
||||
}
|
||||
this.#values.push({ value, bytes });
|
||||
this.#queuedBytes += bytes;
|
||||
return true;
|
||||
}
|
||||
|
||||
close(): void {
|
||||
close(error?: Error): void {
|
||||
if (this.#closed) return;
|
||||
this.#closed = true;
|
||||
for (const waiter of this.#waiters.splice(0)) waiter({ value: undefined, done: true });
|
||||
this.#values = [];
|
||||
this.#queuedBytes = 0;
|
||||
this.#error = error ?? null;
|
||||
for (const waiter of this.#waiters.splice(0)) {
|
||||
if (error === undefined) waiter.resolve({ value: undefined, done: true });
|
||||
else waiter.reject(error);
|
||||
}
|
||||
}
|
||||
|
||||
[Symbol.asyncIterator](): AsyncIterator<T> {
|
||||
return {
|
||||
next: async () => {
|
||||
const value = this.#values.shift();
|
||||
if (value !== undefined) return { value, done: false };
|
||||
const queued = this.#values.shift();
|
||||
if (queued !== undefined) {
|
||||
this.#queuedBytes -= queued.bytes;
|
||||
return { value: queued.value, done: false };
|
||||
}
|
||||
if (this.#error !== null) return Promise.reject(this.#error);
|
||||
if (this.#closed) return { value: undefined, done: true };
|
||||
return new Promise<IteratorResult<T>>((resolve) => this.#waiters.push(resolve));
|
||||
return new Promise<IteratorResult<T>>((resolve, reject) => {
|
||||
this.#waiters.push({ resolve, reject });
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
class BoundedLineDecoder {
|
||||
#buffer = Buffer.alloc(0);
|
||||
#closed = false;
|
||||
|
||||
constructor(
|
||||
private readonly maxLineBytes: number,
|
||||
private readonly onLine: (line: string, bytes: number) => void,
|
||||
private readonly onError: (error: Error) => void,
|
||||
) {}
|
||||
|
||||
write(chunk: Buffer | string): void {
|
||||
if (this.#closed) return;
|
||||
const incoming = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
let offset = 0;
|
||||
while (offset < incoming.length) {
|
||||
const newline = incoming.indexOf(0x0a, offset);
|
||||
const end = newline < 0 ? incoming.length : newline;
|
||||
const segment = incoming.subarray(offset, end);
|
||||
if (this.#buffer.length + segment.length > this.maxLineBytes) {
|
||||
this.#fail(new Error(`codex app-server line exceeded ${this.maxLineBytes} bytes`));
|
||||
return;
|
||||
}
|
||||
if (segment.length > 0) this.#buffer = Buffer.concat([this.#buffer, segment]);
|
||||
if (newline < 0) return;
|
||||
const line = this.#buffer;
|
||||
this.#buffer = Buffer.alloc(0);
|
||||
this.onLine(line.toString("utf8"), line.length);
|
||||
if (this.#closed) return;
|
||||
offset = newline + 1;
|
||||
}
|
||||
}
|
||||
|
||||
end(): void {
|
||||
if (this.#closed) return;
|
||||
if (this.#buffer.length > 0) {
|
||||
this.#fail(new Error("codex app-server ended with an unterminated JSON-RPC line"));
|
||||
}
|
||||
this.#closed = true;
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.#closed = true;
|
||||
this.#buffer = Buffer.alloc(0);
|
||||
}
|
||||
|
||||
#fail(error: Error): void {
|
||||
this.close();
|
||||
this.onError(error);
|
||||
}
|
||||
}
|
||||
|
||||
const SAFE_ENVIRONMENT_KEYS = [
|
||||
"ALL_PROXY",
|
||||
"CODEX_HOME",
|
||||
@@ -71,6 +152,11 @@ const SAFE_ENVIRONMENT_KEYS = [
|
||||
"WINDIR",
|
||||
] as const;
|
||||
|
||||
/**
|
||||
* Environment for the trusted app-server process itself. HOME and CODEX_HOME
|
||||
* are retained so Codex can authenticate; model-issued commands receive a
|
||||
* separate empty-by-default environment and filesystem permission profile.
|
||||
*/
|
||||
export function createSanitizedCodexEnvironment(
|
||||
source: NodeJS.ProcessEnv = process.env,
|
||||
): NodeJS.ProcessEnv {
|
||||
@@ -119,41 +205,94 @@ export interface ProcessCodexTransportOptions {
|
||||
args?: string[];
|
||||
environment?: NodeJS.ProcessEnv;
|
||||
onDiagnostic?: (message: string) => void;
|
||||
maxLineBytes?: number;
|
||||
maxPendingRequests?: number;
|
||||
maxQueuedNotifications?: number;
|
||||
maxQueuedNotificationBytes?: number;
|
||||
maxInflightServerRequests?: number;
|
||||
}
|
||||
|
||||
const DEFAULT_MAX_LINE_BYTES = 1024 * 1024;
|
||||
const DEFAULT_MAX_PENDING_REQUESTS = 64;
|
||||
const DEFAULT_MAX_QUEUED_NOTIFICATIONS = 256;
|
||||
const DEFAULT_MAX_QUEUED_NOTIFICATION_BYTES = 4 * 1024 * 1024;
|
||||
const DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS = 32;
|
||||
const MAX_DIAGNOSTIC_LINE_BYTES = 16 * 1024;
|
||||
|
||||
/** Local stdio JSON-RPC transport. App-server never becomes a WAN endpoint. */
|
||||
export class ProcessCodexAppServerTransport implements CodexAppServerTransport {
|
||||
#process: ChildProcessWithoutNullStreams;
|
||||
#notifications = new AsyncQueue<CodexRpcNotification>();
|
||||
#notifications: BoundedAsyncQueue<CodexRpcNotification>;
|
||||
#pending = new Map<number, PendingRequest>();
|
||||
#nextId = 1;
|
||||
#serverRequestHandler: CodexServerRequestHandler = async () => ({
|
||||
success: false,
|
||||
contentItems: [{ type: "input_text", text: "Unsupported client request." }],
|
||||
});
|
||||
#inflightServerRequests = 0;
|
||||
#closed = false;
|
||||
#onDiagnostic?: (message: string) => void;
|
||||
readonly #maxLineBytes: number;
|
||||
readonly #maxPendingRequests: number;
|
||||
readonly #maxInflightServerRequests: number;
|
||||
readonly #stdoutDecoder: BoundedLineDecoder;
|
||||
readonly #stderrDecoder: BoundedLineDecoder;
|
||||
|
||||
constructor(options: ProcessCodexTransportOptions = {}) {
|
||||
this.#onDiagnostic = options.onDiagnostic;
|
||||
this.#maxLineBytes = positiveLimit(options.maxLineBytes, DEFAULT_MAX_LINE_BYTES);
|
||||
this.#maxPendingRequests = positiveLimit(
|
||||
options.maxPendingRequests,
|
||||
DEFAULT_MAX_PENDING_REQUESTS,
|
||||
);
|
||||
this.#maxInflightServerRequests = positiveLimit(
|
||||
options.maxInflightServerRequests,
|
||||
DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS,
|
||||
);
|
||||
this.#notifications = new BoundedAsyncQueue(
|
||||
positiveLimit(options.maxQueuedNotifications, DEFAULT_MAX_QUEUED_NOTIFICATIONS),
|
||||
positiveLimit(
|
||||
options.maxQueuedNotificationBytes,
|
||||
DEFAULT_MAX_QUEUED_NOTIFICATION_BYTES,
|
||||
),
|
||||
);
|
||||
this.#process = spawn(options.command ?? "codex", options.args ?? ["app-server"], {
|
||||
env: options.environment ?? createSanitizedCodexEnvironment(),
|
||||
stdio: "pipe",
|
||||
});
|
||||
createInterface({ input: this.#process.stdout }).on("line", (line) => this.#onLine(line));
|
||||
createInterface({ input: this.#process.stderr }).on("line", (line) => {
|
||||
this.#onDiagnostic?.(redactCodexDiagnostic(line));
|
||||
});
|
||||
this.#process.on("error", (error) => this.#failAll(error));
|
||||
this.#stdoutDecoder = new BoundedLineDecoder(
|
||||
this.#maxLineBytes,
|
||||
(line, bytes) => this.#onLine(line, bytes),
|
||||
(error) => this.#fatal(error),
|
||||
);
|
||||
this.#stderrDecoder = new BoundedLineDecoder(
|
||||
MAX_DIAGNOSTIC_LINE_BYTES,
|
||||
(line) => this.#onDiagnostic?.(redactCodexDiagnostic(line)),
|
||||
() => this.#onDiagnostic?.("codex app-server diagnostic line exceeded retention limit"),
|
||||
);
|
||||
this.#process.stdout.on("data", (chunk: Buffer) => this.#stdoutDecoder.write(chunk));
|
||||
this.#process.stdout.on("end", () => this.#stdoutDecoder.end());
|
||||
this.#process.stderr.on("data", (chunk: Buffer) => this.#stderrDecoder.write(chunk));
|
||||
this.#process.stderr.on("end", () => this.#stderrDecoder.end());
|
||||
this.#process.on("error", (error) => this.#fatal(error));
|
||||
this.#process.on("exit", (code, signal) => {
|
||||
if (!this.#closed) {
|
||||
this.#failAll(new Error(`codex app-server exited (code=${code ?? "null"}, signal=${signal ?? "none"})`));
|
||||
this.#fatal(
|
||||
new Error(
|
||||
`codex app-server exited (code=${code ?? "null"}, signal=${signal ?? "none"})`,
|
||||
),
|
||||
);
|
||||
}
|
||||
this.#notifications.close();
|
||||
});
|
||||
}
|
||||
|
||||
request(method: string, params: Record<string, unknown>): Promise<Record<string, unknown>> {
|
||||
if (this.#closed) return Promise.reject(new Error("codex app-server transport is closed"));
|
||||
if (this.#pending.size >= this.#maxPendingRequests) {
|
||||
return Promise.reject(
|
||||
new Error(`codex app-server pending request limit ${this.#maxPendingRequests} exceeded`),
|
||||
);
|
||||
}
|
||||
const id = this.#nextId++;
|
||||
return new Promise((resolve, reject) => {
|
||||
this.#pending.set(id, { resolve, reject });
|
||||
@@ -181,6 +320,8 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport {
|
||||
async close(): Promise<void> {
|
||||
if (this.#closed) return;
|
||||
this.#closed = true;
|
||||
this.#stdoutDecoder.close();
|
||||
this.#stderrDecoder.close();
|
||||
this.#process.kill("SIGTERM");
|
||||
this.#notifications.close();
|
||||
this.#failAll(new Error("codex app-server transport closed"));
|
||||
@@ -188,45 +329,107 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport {
|
||||
|
||||
#write(message: unknown): void {
|
||||
if (this.#closed) throw new Error("codex app-server transport is closed");
|
||||
this.#process.stdin.write(`${JSON.stringify(message)}\n`);
|
||||
const serialized = JSON.stringify(message);
|
||||
if (Buffer.byteLength(serialized) > this.#maxLineBytes) {
|
||||
throw new Error(`outbound codex JSON-RPC line exceeded ${this.#maxLineBytes} bytes`);
|
||||
}
|
||||
this.#process.stdin.write(`${serialized}\n`);
|
||||
}
|
||||
|
||||
#onLine(line: string): void {
|
||||
#onLine(line: string, bytes: number): void {
|
||||
let value: unknown;
|
||||
try {
|
||||
value = JSON.parse(line);
|
||||
} catch {
|
||||
this.#onDiagnostic?.("codex app-server emitted malformed JSON");
|
||||
this.#fatal(new Error("codex app-server emitted malformed JSON"));
|
||||
return;
|
||||
}
|
||||
if (typeof value !== "object" || value === null) return;
|
||||
const message = value as Record<string, unknown>;
|
||||
if (typeof message.id === "number" && ("result" in message || "error" in message)) {
|
||||
if (!isRecord(value)) {
|
||||
this.#fatal(new Error("codex app-server emitted a non-object JSON-RPC message"));
|
||||
return;
|
||||
}
|
||||
const message = value;
|
||||
const hasResult = Object.hasOwn(message, "result");
|
||||
const hasError = Object.hasOwn(message, "error");
|
||||
if (typeof message.id === "number" && (hasResult || hasError)) {
|
||||
if (!Number.isSafeInteger(message.id) || message.id <= 0 || hasResult === hasError) {
|
||||
this.#fatal(new Error("codex app-server emitted a malformed JSON-RPC response"));
|
||||
return;
|
||||
}
|
||||
const pending = this.#pending.get(message.id);
|
||||
if (pending === undefined) return;
|
||||
if (pending === undefined) {
|
||||
this.#fatal(new Error("codex app-server responded to an unknown request id"));
|
||||
return;
|
||||
}
|
||||
this.#pending.delete(message.id);
|
||||
if (message.error !== undefined) {
|
||||
pending.reject(new Error(redactCodexDiagnostic(JSON.stringify(message.error))));
|
||||
if (hasError) {
|
||||
pending.reject(new Error(redactCodexDiagnostic(boundedJson(message.error))));
|
||||
} else if (isRecord(message.result)) {
|
||||
pending.resolve(message.result);
|
||||
} else {
|
||||
pending.resolve(asRecord(message.result));
|
||||
const error = new Error("codex app-server response result was not an object");
|
||||
pending.reject(error);
|
||||
this.#fatal(error);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (typeof message.method !== "string") return;
|
||||
const params = asRecord(message.params);
|
||||
if ((typeof message.id === "string" || typeof message.id === "number") && "id" in message) {
|
||||
const request: CodexRpcServerRequest = { id: message.id, method: message.method, params };
|
||||
void this.#serverRequestHandler(request).then(
|
||||
(result) => this.#write({ id: request.id, result }),
|
||||
(error: unknown) =>
|
||||
if (typeof message.method !== "string" || message.method.length === 0 || hasResult || hasError) {
|
||||
this.#fatal(new Error("codex app-server emitted a malformed JSON-RPC message"));
|
||||
return;
|
||||
}
|
||||
if (message.params !== undefined && !isRecord(message.params)) {
|
||||
this.#fatal(new Error("codex app-server message params were not an object"));
|
||||
return;
|
||||
}
|
||||
const params = message.params ?? {};
|
||||
if (message.id !== undefined) {
|
||||
if (!isValidServerRequestId(message.id)) {
|
||||
this.#fatal(new Error("codex app-server request id had an invalid type"));
|
||||
return;
|
||||
}
|
||||
this.#handleServerRequest({ id: message.id, method: message.method, params });
|
||||
return;
|
||||
}
|
||||
if (!this.#notifications.push({ method: message.method, params }, bytes)) {
|
||||
this.#fatal(new Error("codex app-server notification queue limit exceeded"));
|
||||
}
|
||||
}
|
||||
|
||||
#handleServerRequest(request: CodexRpcServerRequest): void {
|
||||
if (this.#inflightServerRequests >= this.#maxInflightServerRequests) {
|
||||
this.#write({
|
||||
id: request.id,
|
||||
error: { code: -32_001, message: "Client request queue limit exceeded." },
|
||||
});
|
||||
return;
|
||||
}
|
||||
this.#inflightServerRequests += 1;
|
||||
void this.#serverRequestHandler(request).then(
|
||||
(result) => {
|
||||
if (!this.#closed) this.#write({ id: request.id, result });
|
||||
},
|
||||
(error: unknown) => {
|
||||
if (!this.#closed) {
|
||||
this.#write({
|
||||
id: request.id,
|
||||
error: { code: -32_000, message: redactCodexDiagnostic(String(error)) },
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.#notifications.push({ method: message.method, params });
|
||||
});
|
||||
}
|
||||
},
|
||||
).finally(() => {
|
||||
this.#inflightServerRequests -= 1;
|
||||
});
|
||||
}
|
||||
|
||||
#fatal(error: Error): void {
|
||||
if (this.#closed) return;
|
||||
this.#closed = true;
|
||||
this.#stdoutDecoder.close();
|
||||
this.#stderrDecoder.close();
|
||||
this.#process.kill("SIGKILL");
|
||||
this.#notifications.close(error);
|
||||
this.#failAll(error);
|
||||
this.#onDiagnostic?.(redactCodexDiagnostic(error.message));
|
||||
}
|
||||
|
||||
#failAll(error: Error): void {
|
||||
@@ -235,8 +438,24 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport {
|
||||
}
|
||||
}
|
||||
|
||||
function asRecord(value: unknown): Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function positiveLimit(value: number | undefined, fallback: number): number {
|
||||
return Number.isSafeInteger(value) && (value ?? 0) > 0 ? value! : fallback;
|
||||
}
|
||||
|
||||
function isValidServerRequestId(value: unknown): value is string | number {
|
||||
return (
|
||||
(typeof value === "number" && Number.isSafeInteger(value)) ||
|
||||
(typeof value === "string" && value.length > 0 && Buffer.byteLength(value) <= 256)
|
||||
);
|
||||
}
|
||||
|
||||
function boundedJson(value: unknown): string {
|
||||
const serialized = JSON.stringify(value) ?? String(value);
|
||||
return serialized.length <= MAX_DIAGNOSTIC_LINE_BYTES
|
||||
? serialized
|
||||
: `${serialized.slice(0, MAX_DIAGNOSTIC_LINE_BYTES)}...[truncated]`;
|
||||
}
|
||||
@@ -19,9 +19,11 @@ import {
|
||||
} from "../../protocol/phase1-contract.js";
|
||||
import {
|
||||
CodexAppServerDriver,
|
||||
createIsolatedCodexAppServerArgs,
|
||||
} from "./codex-app-server-driver.js";
|
||||
import {
|
||||
runPhase4CodexTracer,
|
||||
replayPersistedPhase4Events,
|
||||
validatePhase4ResultProposal,
|
||||
} from "../../mock-core/phase4-codex-runner.js";
|
||||
import type {
|
||||
@@ -87,6 +89,7 @@ class FakeCodexTransport implements CodexAppServerTransport {
|
||||
modelProvider: "openai",
|
||||
cwd: "/workspace",
|
||||
turns: [],
|
||||
activePermissionProfile: { id: "paperclip-runner-workspace-only" },
|
||||
},
|
||||
model: "gpt-test",
|
||||
modelProvider: "openai",
|
||||
@@ -98,7 +101,7 @@ class FakeCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
if (method === "turn/start") return { turn: { id: "turn-1", status: "inProgress", items: [] } };
|
||||
if (method === "thread/read") {
|
||||
return this.readResponse ?? { thread: { id: this.threadId, sessionId: this.providerSessionId, turns: [{ id: "turn-1", status: "inProgress", items: [] }] } };
|
||||
return this.readResponse ?? { thread: { id: this.threadId, sessionId: this.providerSessionId, cwd: "/workspace", turns: [{ id: "turn-1", status: "inProgress", items: [] }] } };
|
||||
}
|
||||
return {};
|
||||
}
|
||||
@@ -300,7 +303,7 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
appInstructions: false,
|
||||
collaborationInstructions: false,
|
||||
},
|
||||
environmentKeys: ["CODEX_HOME", "HOME", "LANG", "PATH"],
|
||||
environmentKeys: ["LANG", "PATH"],
|
||||
envelope,
|
||||
});
|
||||
expect(isSkilllessPhase4Context(context)).toBe(true);
|
||||
@@ -313,8 +316,35 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
include_apps_instructions: false,
|
||||
include_collaboration_mode_instructions: false,
|
||||
},
|
||||
permissions: "paperclip-runner-workspace-only",
|
||||
runtimeWorkspaceRoots: ["/workspace"],
|
||||
dynamicTools: [{ name: "paperclip_finish" }, { name: "paperclip_block" }],
|
||||
});
|
||||
const commandPolicy = transport.calls.find((call) => call.method === "thread/start")
|
||||
?.params.config as Record<string, unknown>;
|
||||
expect(JSON.stringify(commandPolicy)).not.toContain("/isolated/home");
|
||||
expect(JSON.stringify(commandPolicy)).not.toContain("/isolated/codex");
|
||||
expect(JSON.stringify(commandPolicy)).not.toContain("CODEX_HOME");
|
||||
const appServerArgs = createIsolatedCodexAppServerArgs({
|
||||
PATH: "/bin",
|
||||
LANG: "C.UTF-8",
|
||||
HOME: "/isolated/home",
|
||||
CODEX_HOME: "/isolated/codex",
|
||||
});
|
||||
expect(appServerArgs).toEqual(expect.arrayContaining([
|
||||
expect.stringContaining('default_permissions="paperclip-runner-workspace-only"'),
|
||||
expect.stringContaining('permissions.paperclip-runner-workspace-only.filesystem='),
|
||||
"permissions.paperclip-runner-workspace-only.network.enabled=false",
|
||||
'shell_environment_policy.inherit="none"',
|
||||
expect.stringContaining('shell_environment_policy.set={PATH="/bin",LANG="C.UTF-8"}'),
|
||||
]));
|
||||
expect(appServerArgs.find((arg) => arg.startsWith("permissions."))).toContain(
|
||||
'"/isolated/home"="none"',
|
||||
);
|
||||
expect(appServerArgs.find((arg) => arg.startsWith("permissions."))).toContain(
|
||||
'"/isolated/codex"="none"',
|
||||
);
|
||||
expect(JSON.stringify(appServerArgs)).not.toContain("HOME=");
|
||||
expect(PHASE4_RESULT_OUTPUT_SCHEMA.properties.schema).toEqual({
|
||||
type: "string",
|
||||
const: "paperclip.run_result.v1",
|
||||
@@ -325,6 +355,24 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("refuses to turn host credential roots into model-writable workspaces", async () => {
|
||||
await expect(makeDriver([]).openSession({
|
||||
runId: "run-codex-home",
|
||||
normalizedSessionId: "normalized-codex-home",
|
||||
workingDirectory: "/isolated/codex",
|
||||
})).rejects.toThrow("cannot overlap host CODEX_HOME");
|
||||
await expect(makeDriver([]).openSession({
|
||||
runId: "run-host-home",
|
||||
normalizedSessionId: "normalized-host-home",
|
||||
workingDirectory: "/isolated",
|
||||
})).rejects.toThrow("cannot contain the host HOME");
|
||||
await expect(makeDriver([]).openSession({
|
||||
runId: "run-root",
|
||||
normalizedSessionId: "normalized-root",
|
||||
workingDirectory: "/",
|
||||
})).rejects.toThrow("cannot be a filesystem root");
|
||||
});
|
||||
|
||||
it("steers and interrupts an active turn without replacing the session", async () => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({ runId: "run-controls", normalizedSessionId: "normalized-controls", workingDirectory: "/workspace" });
|
||||
@@ -358,6 +406,144 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
expect(events.filter((event) => event.eventType === "turn.completed")).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ threadId: "other-thread", turnId: "turn-1", label: "another thread" },
|
||||
{ threadId: "thread-1", turnId: "other-turn", label: "another turn" },
|
||||
])("rejects a semantic result for $label without committing it", async ({ threadId, turnId }) => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({
|
||||
runId: "run-hostile-tool",
|
||||
normalizedSessionId: "normalized-hostile-tool",
|
||||
workingDirectory: "/workspace",
|
||||
});
|
||||
await session.startTurn({ message: { role: "user", text: "Complete." } });
|
||||
expect(await transport.invoke({
|
||||
id: "hostile-call",
|
||||
method: "item/tool/call",
|
||||
params: { threadId, turnId, callId: "hostile-call", tool: "paperclip_finish", arguments: result },
|
||||
})).toMatchObject({ success: false });
|
||||
const events = await collectUntilTerminal(session.events());
|
||||
expect(events.some((event) => event.eventType === "run.result.proposed")).toBe(false);
|
||||
expect(events.find((event) => event.eventType === "session.failed")?.payload)
|
||||
.toMatchObject({ code: "tool_binding_mismatch", recoverable: false });
|
||||
expect(events.find((event) => event.eventType === "turn.failed")?.turnId).toBe("turn-1");
|
||||
});
|
||||
|
||||
it("rejects pre-turn, cross-thread, and post-terminal notifications", async () => {
|
||||
const preTurnTransport = new FakeCodexTransport();
|
||||
const preTurn = await makeDriver([preTurnTransport]).openSession({
|
||||
runId: "run-pre-turn",
|
||||
normalizedSessionId: "normalized-pre-turn",
|
||||
workingDirectory: "/workspace",
|
||||
});
|
||||
preTurnTransport.push("item/completed", {
|
||||
threadId: "thread-1",
|
||||
turnId: "turn-1",
|
||||
item: { id: "answer-pre", type: "agentMessage", text: JSON.stringify(result) },
|
||||
});
|
||||
const preTurnEvents: PrpEvent[] = [];
|
||||
for await (const event of preTurn.events()) preTurnEvents.push(event);
|
||||
expect(preTurnEvents.some((event) => event.eventType === "run.result.proposed")).toBe(false);
|
||||
expect(preTurnEvents.find((event) => event.eventType === "session.failed")?.payload)
|
||||
.toMatchObject({ code: "turn_binding_mismatch" });
|
||||
|
||||
const crossThreadTransport = new FakeCodexTransport();
|
||||
const crossThread = await makeDriver([crossThreadTransport]).openSession({
|
||||
runId: "run-cross-thread",
|
||||
normalizedSessionId: "normalized-cross-thread",
|
||||
workingDirectory: "/workspace",
|
||||
});
|
||||
await crossThread.startTurn({ message: { role: "user", text: "Complete." } });
|
||||
crossThreadTransport.push("item/completed", {
|
||||
threadId: "other-thread",
|
||||
turnId: "turn-1",
|
||||
item: { id: "answer-other", type: "agentMessage", text: JSON.stringify(result) },
|
||||
});
|
||||
const crossThreadEvents = await collectUntilTerminal(crossThread.events());
|
||||
expect(crossThreadEvents.some((event) => event.eventType === "run.result.proposed")).toBe(false);
|
||||
expect(crossThreadEvents.find((event) => event.eventType === "session.failed")?.payload)
|
||||
.toMatchObject({ code: "thread_binding_mismatch" });
|
||||
|
||||
const postTerminalTransport = new FakeCodexTransport();
|
||||
const postTerminal = await makeDriver([postTerminalTransport]).openSession({
|
||||
runId: "run-post-terminal",
|
||||
normalizedSessionId: "normalized-post-terminal",
|
||||
workingDirectory: "/workspace",
|
||||
});
|
||||
await postTerminal.startTurn({ message: { role: "user", text: "Complete." } });
|
||||
postTerminalTransport.push("turn/completed", {
|
||||
threadId: "thread-1",
|
||||
turn: { id: "turn-1", status: "completed", items: [] },
|
||||
});
|
||||
await collectUntilTerminal(postTerminal.events());
|
||||
expect(await postTerminalTransport.invoke({
|
||||
id: "late-call",
|
||||
method: "item/tool/call",
|
||||
params: {
|
||||
threadId: "thread-1",
|
||||
turnId: "turn-1",
|
||||
callId: "late-call",
|
||||
tool: "paperclip_finish",
|
||||
arguments: result,
|
||||
},
|
||||
})).toMatchObject({ success: false });
|
||||
});
|
||||
|
||||
it("rejects duplicate and conflicting terminal facts", async () => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({
|
||||
runId: "run-terminal-replay",
|
||||
normalizedSessionId: "normalized-terminal-replay",
|
||||
workingDirectory: "/workspace",
|
||||
});
|
||||
await session.startTurn({ message: { role: "user", text: "Complete." } });
|
||||
transport.push("turn/completed", {
|
||||
threadId: "thread-1",
|
||||
turn: { id: "turn-1", status: "completed", items: [] },
|
||||
});
|
||||
transport.push("turn/completed", {
|
||||
threadId: "thread-1",
|
||||
turn: { id: "turn-1", status: "failed", error: { message: "changed" }, items: [] },
|
||||
});
|
||||
const events: PrpEvent[] = [];
|
||||
for await (const event of session.events()) events.push(event);
|
||||
expect(events.filter((event) => event.eventType === "turn.completed")).toHaveLength(1);
|
||||
expect(events.filter((event) => event.eventType === "turn.failed")).toHaveLength(0);
|
||||
expect(events.find((event) => event.eventType === "session.failed")?.payload)
|
||||
.toMatchObject({ code: "conflicting_turn_terminal", recoverable: false });
|
||||
});
|
||||
|
||||
it("rejects oversized semantic results without retaining provider payloads", async () => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({
|
||||
runId: "run-large-result",
|
||||
normalizedSessionId: "normalized-large-result",
|
||||
workingDirectory: "/workspace",
|
||||
});
|
||||
await session.startTurn({ message: { role: "user", text: "Complete." } });
|
||||
expect(await transport.invoke({
|
||||
id: "large-call",
|
||||
method: "item/tool/call",
|
||||
params: {
|
||||
threadId: "thread-1",
|
||||
turnId: "turn-1",
|
||||
callId: "large-call",
|
||||
tool: "paperclip_finish",
|
||||
arguments: { ...result, summary: "x".repeat(70 * 1024) },
|
||||
},
|
||||
})).toEqual({
|
||||
success: false,
|
||||
contentItems: [{ type: "inputText", text: "Semantic result exceeded the retained payload limit." }],
|
||||
});
|
||||
transport.push("turn/completed", {
|
||||
threadId: "thread-1",
|
||||
turn: { id: "turn-1", status: "completed", items: [] },
|
||||
});
|
||||
const events = await collectUntilTerminal(session.events());
|
||||
expect(events.some((event) => JSON.stringify(event).includes("x".repeat(1024)))).toBe(false);
|
||||
expect(events.some((event) => event.eventType === "run.result.proposed")).toBe(false);
|
||||
});
|
||||
|
||||
it("normalizes finish and block tools into one canonical result contract", async () => {
|
||||
const transport = new FakeCodexTransport();
|
||||
const session = await makeDriver([transport]).openSession({ runId: "run-tools", normalizedSessionId: "normalized-tools", workingDirectory: "/workspace" });
|
||||
@@ -417,7 +603,7 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
expect(recovery).toMatchObject({ recovered: true });
|
||||
expect(recovery?.session?.ids()).toEqual(original.ids());
|
||||
expect(await recovery?.session?.reconcile?.()).toMatchObject({ thread: { id: "thread-1" } });
|
||||
expect(second.calls.map((call) => call.method)).toEqual(["initialize", "thread/resume", "thread/read"]);
|
||||
expect(second.calls.map((call) => call.method)).toEqual(["initialize", "thread/read", "thread/resume", "thread/read"]);
|
||||
expect((await recovery?.session?.snapshot())?.activeTurnId).toBe("turn-1");
|
||||
});
|
||||
|
||||
@@ -428,6 +614,7 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
thread: {
|
||||
id: "driver-thread-loss",
|
||||
sessionId: "provider-session-loss",
|
||||
cwd: "/workspace",
|
||||
tokenUsage: { total: { inputTokens: 21, outputTokens: 8 } },
|
||||
turns: [{
|
||||
id: "turn-1",
|
||||
@@ -582,18 +769,48 @@ describe("Codex app-server Phase 4 driver", () => {
|
||||
normalizedSessionId: "controller-session-identity",
|
||||
});
|
||||
expect(trace.assertions.stableIdentity).toBe(true);
|
||||
expect(new Set([
|
||||
trace.metadata.identity.runId,
|
||||
trace.metadata.identity.normalizedSessionId,
|
||||
trace.metadata.identity.driverSessionId,
|
||||
trace.metadata.identity.providerSessionId,
|
||||
]).size).toBe(4);
|
||||
expect(trace.metadata.identity).toMatchObject({
|
||||
runId: "runtime-run-identity",
|
||||
normalizedSessionId: "controller-session-identity",
|
||||
driverSessionId: "thread-1",
|
||||
providerSessionId: "provider-session-1",
|
||||
});
|
||||
expect(trace.events.every((event) =>
|
||||
event.runId === "runtime-run-identity" &&
|
||||
event.normalizedSessionId === "controller-session-identity"
|
||||
)).toBe(true);
|
||||
});
|
||||
|
||||
it("validates persisted identity, uniqueness, terminals, and line bounds before replay", async () => {
|
||||
const { trace } = await traceCompletedProposal(result, {
|
||||
runId: "runtime-run-persisted",
|
||||
normalizedSessionId: "controller-session-persisted",
|
||||
});
|
||||
const serialize = (events: PrpEvent[]) => `${events.map((event) => JSON.stringify(event)).join("\n")}\n`;
|
||||
expect(replayPersistedPhase4Events(serialize(trace.events), trace.metadata))
|
||||
.toEqual(trace.replaySnapshot);
|
||||
|
||||
const mismatched = structuredClone(trace.events);
|
||||
mismatched[0]!.normalizedSessionId = "another-session";
|
||||
expect(() => replayPersistedPhase4Events(serialize(mismatched), trace.metadata))
|
||||
.toThrow("identity did not match");
|
||||
|
||||
const terminalIndex = trace.events.findIndex((event) => event.eventType === "run.terminal");
|
||||
const duplicated = trace.events.toSpliced(terminalIndex, 0, structuredClone(trace.events[0]!));
|
||||
expect(() => replayPersistedPhase4Events(serialize(duplicated), trace.metadata))
|
||||
.toThrow("event id was duplicated");
|
||||
|
||||
const terminal = structuredClone(trace.events[terminalIndex]!);
|
||||
terminal.sourceEventId = `${terminal.sourceEventId}:conflict`;
|
||||
terminal.sourceSeq += 1;
|
||||
expect(() => replayPersistedPhase4Events(serialize([...trace.events, terminal]), trace.metadata))
|
||||
.toThrow("after the run terminal");
|
||||
expect(() => replayPersistedPhase4Events(`{\"payload\":\"${"x".repeat(300 * 1024)}\"}\n`, trace.metadata))
|
||||
.toThrow("line was empty or oversized");
|
||||
expect(() => replayPersistedPhase4Events("{not-json}\n", trace.metadata))
|
||||
.toThrow("malformed JSON");
|
||||
});
|
||||
|
||||
it("degrades declared unsupported capabilities without Codex-specific core branches", async () => {
|
||||
const capabilities = {
|
||||
resume: false,
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { parse, relative, resolve } from "node:path";
|
||||
|
||||
import type {
|
||||
HarnessDriver,
|
||||
@@ -33,14 +33,16 @@ import {
|
||||
ProcessCodexAppServerTransport,
|
||||
createSanitizedCodexEnvironment,
|
||||
redactCodexDiagnostic,
|
||||
sanitizedEnvironmentKeys,
|
||||
type CodexAppServerTransport,
|
||||
type CodexRpcNotification,
|
||||
type CodexRpcServerRequest,
|
||||
} from "./app-server-transport.js";
|
||||
|
||||
const DRIVER_KIND = "codex_app_server";
|
||||
const DRIVER_VERSION = "phase4-v1";
|
||||
const DRIVER_VERSION = "phase4-v2";
|
||||
const SKILLLESS_PERMISSION_PROFILE = "paperclip-runner-workspace-only";
|
||||
const MAX_RETAINED_CODEX_PAYLOAD_BYTES = 64 * 1024;
|
||||
const MAX_RETAINED_CODEX_STRING_CHARS = 32 * 1024;
|
||||
|
||||
class AsyncQueue<T> implements AsyncIterable<T> {
|
||||
#values: T[] = [];
|
||||
@@ -110,6 +112,13 @@ function text(value: unknown, fallback = ""): string {
|
||||
return typeof value === "string" ? value : fallback;
|
||||
}
|
||||
|
||||
function boundedText(value: unknown, fallback = "unknown", maxCharacters = 1024): string {
|
||||
const candidate = text(value, fallback);
|
||||
return candidate.length <= maxCharacters
|
||||
? candidate
|
||||
: `${candidate.slice(0, maxCharacters)}...[truncated]`;
|
||||
}
|
||||
|
||||
function itemFromParams(params: Record<string, unknown>): Record<string, unknown> {
|
||||
return record(params.item);
|
||||
}
|
||||
@@ -172,7 +181,7 @@ function blockToolSpec(): Record<string, unknown> {
|
||||
};
|
||||
}
|
||||
|
||||
const SKILLLESS_THREAD_CONFIG = {
|
||||
const SKILLLESS_BASE_CONFIG = {
|
||||
"skills.include_instructions": false,
|
||||
include_apps_instructions: false,
|
||||
include_collaboration_mode_instructions: false,
|
||||
@@ -182,6 +191,72 @@ const SKILLLESS_THREAD_CONFIG = {
|
||||
"features.memories": false,
|
||||
} as const;
|
||||
|
||||
function commandEnvironment(source: NodeJS.ProcessEnv = process.env): Record<string, string> {
|
||||
const environment: Record<string, string> = {};
|
||||
for (const key of ["PATH", "PATHEXT", "SystemRoot", "WINDIR", "LANG", "LC_ALL"] as const) {
|
||||
const value = source[key];
|
||||
if (value !== undefined) environment[key] = value;
|
||||
}
|
||||
return environment;
|
||||
}
|
||||
|
||||
export function createSkilllessCodexThreadConfig(
|
||||
_workingDirectory: string,
|
||||
_source: NodeJS.ProcessEnv = process.env,
|
||||
): Record<string, unknown> {
|
||||
return { ...SKILLLESS_BASE_CONFIG };
|
||||
}
|
||||
|
||||
function tomlString(value: string): string {
|
||||
return JSON.stringify(value);
|
||||
}
|
||||
|
||||
export function createIsolatedCodexAppServerArgs(
|
||||
source: NodeJS.ProcessEnv = process.env,
|
||||
): string[] {
|
||||
const deniedHostRoots = [...new Set([source.HOME, source.CODEX_HOME]
|
||||
.filter((value): value is string => typeof value === "string" && value.trim().length > 0)
|
||||
.map((value) => resolve(value)))];
|
||||
const filesystemRules = [
|
||||
`\":root\"=\"none\"`,
|
||||
`\":minimal\"=\"read\"`,
|
||||
`\":tmpdir\"=\"none\"`,
|
||||
...deniedHostRoots.map((path) => `${tomlString(path)}=\"none\"`),
|
||||
`\":workspace_roots\"={\".\"=\"write\"}`,
|
||||
].join(",");
|
||||
const commandEnv = Object.entries(commandEnvironment(source))
|
||||
.map(([key, value]) => `${key}=${tomlString(value)}`)
|
||||
.join(",");
|
||||
return [
|
||||
"-c",
|
||||
`default_permissions=${tomlString(SKILLLESS_PERMISSION_PROFILE)}`,
|
||||
"-c",
|
||||
`permissions.${SKILLLESS_PERMISSION_PROFILE}.filesystem={${filesystemRules}}`,
|
||||
"-c",
|
||||
`permissions.${SKILLLESS_PERMISSION_PROFILE}.network.enabled=false`,
|
||||
"-c",
|
||||
`shell_environment_policy.inherit=\"none\"`,
|
||||
"-c",
|
||||
"shell_environment_policy.ignore_default_excludes=false",
|
||||
...(commandEnv.length > 0
|
||||
? ["-c", `shell_environment_policy.set={${commandEnv}}`]
|
||||
: []),
|
||||
"app-server",
|
||||
];
|
||||
}
|
||||
|
||||
function securedThreadParams(
|
||||
workingDirectory: string,
|
||||
source: NodeJS.ProcessEnv = process.env,
|
||||
): Record<string, unknown> {
|
||||
return {
|
||||
cwd: workingDirectory,
|
||||
config: createSkilllessCodexThreadConfig(workingDirectory, source),
|
||||
permissions: SKILLLESS_PERMISSION_PROFILE,
|
||||
runtimeWorkspaceRoots: [workingDirectory],
|
||||
};
|
||||
}
|
||||
|
||||
export class CodexAppServerDriver implements HarnessDriver {
|
||||
readonly #options: CodexAppServerDriverOptions;
|
||||
readonly #caps: CodexCapabilities;
|
||||
@@ -226,27 +301,31 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
}
|
||||
|
||||
async openSession(input: OpenHarnessSessionInput): Promise<HarnessSession> {
|
||||
const workingDirectory = validateWorkingDirectory(input.workingDirectory, this.#options.environment);
|
||||
const transport = this.#transport();
|
||||
const initialize = await this.#initialize(transport);
|
||||
const response = await transport.request("thread/start", {
|
||||
cwd: input.workingDirectory,
|
||||
approvalPolicy: "never",
|
||||
sandbox: "workspace-write",
|
||||
config: SKILLLESS_THREAD_CONFIG,
|
||||
baseInstructions: PHASE4_SKILLLESS_BASE_INSTRUCTIONS,
|
||||
dynamicTools: this.#caps.dynamicTools ? [finishToolSpec(), blockToolSpec()] : [],
|
||||
experimentalRawEvents: false,
|
||||
persistExtendedHistory: false,
|
||||
});
|
||||
const opened = this.#openedThread(response, initialize, input.workingDirectory);
|
||||
return this.#session({
|
||||
transport,
|
||||
runId: input.runId,
|
||||
normalizedSessionId: input.normalizedSessionId,
|
||||
opened,
|
||||
resumed: false,
|
||||
sourceSequence: 0,
|
||||
});
|
||||
try {
|
||||
const initialize = await this.#initialize(transport);
|
||||
const response = await transport.request("thread/start", {
|
||||
...securedThreadParams(workingDirectory, this.#options.environment),
|
||||
approvalPolicy: "never",
|
||||
baseInstructions: PHASE4_SKILLLESS_BASE_INSTRUCTIONS,
|
||||
dynamicTools: this.#caps.dynamicTools ? [finishToolSpec(), blockToolSpec()] : [],
|
||||
experimentalRawEvents: false,
|
||||
persistExtendedHistory: false,
|
||||
});
|
||||
const opened = this.#openedThread(response, initialize, workingDirectory);
|
||||
return this.#session({
|
||||
transport,
|
||||
runId: input.runId,
|
||||
normalizedSessionId: input.normalizedSessionId,
|
||||
opened,
|
||||
resumed: false,
|
||||
sourceSequence: 0,
|
||||
});
|
||||
} catch (error) {
|
||||
await transport.close();
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async recoverSession(
|
||||
@@ -264,13 +343,26 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
const transport = this.#transport();
|
||||
try {
|
||||
const initialize = await this.#initialize(transport);
|
||||
const existing = await transport.request("thread/read", {
|
||||
threadId: snapshot.driverSessionId,
|
||||
includeTurns: false,
|
||||
});
|
||||
const existingThread = record(existing.thread);
|
||||
if (text(existingThread.id) !== snapshot.driverSessionId) {
|
||||
await transport.close();
|
||||
return { recovered: false, reason: "provider read a different session" };
|
||||
}
|
||||
const workingDirectory = validateWorkingDirectory(
|
||||
text(existingThread.cwd),
|
||||
this.#options.environment,
|
||||
);
|
||||
const response = await transport.request("thread/resume", {
|
||||
threadId: snapshot.driverSessionId,
|
||||
config: SKILLLESS_THREAD_CONFIG,
|
||||
...securedThreadParams(workingDirectory, this.#options.environment),
|
||||
baseInstructions: PHASE4_SKILLLESS_BASE_INSTRUCTIONS,
|
||||
persistExtendedHistory: false,
|
||||
});
|
||||
const opened = this.#openedThread(response, initialize, text(record(response.thread).cwd));
|
||||
const opened = this.#openedThread(response, initialize, workingDirectory);
|
||||
if (opened.threadId !== snapshot.driverSessionId) {
|
||||
await transport.close();
|
||||
return { recovered: false, reason: "provider resumed a different session" };
|
||||
@@ -304,6 +396,7 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
return (
|
||||
this.#options.transportFactory?.() ??
|
||||
new ProcessCodexAppServerTransport({
|
||||
args: createIsolatedCodexAppServerArgs(this.#options.environment),
|
||||
environment: createSanitizedCodexEnvironment(this.#options.environment),
|
||||
onDiagnostic: this.#options.onDiagnostic,
|
||||
})
|
||||
@@ -327,32 +420,60 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
const thread = record(response.thread);
|
||||
const threadId = text(thread.id);
|
||||
if (threadId.length === 0) throw new Error("Codex thread response omitted thread.id");
|
||||
const activePermissionProfile = record(thread.activePermissionProfile);
|
||||
const permissionProfileId = text(activePermissionProfile.id);
|
||||
if (
|
||||
permissionProfileId.length > 0 &&
|
||||
permissionProfileId !== SKILLLESS_PERMISSION_PROFILE
|
||||
) {
|
||||
throw new Error("Codex thread did not activate the required filesystem permission profile");
|
||||
}
|
||||
const configuredPermissionProfile = {
|
||||
...activePermissionProfile,
|
||||
id: SKILLLESS_PERMISSION_PROFILE,
|
||||
};
|
||||
const returnedWorkingDirectory = text(response.cwd, workingDirectory);
|
||||
if (resolve(returnedWorkingDirectory) !== workingDirectory) {
|
||||
throw new Error("Codex thread response changed the assigned working directory");
|
||||
}
|
||||
return {
|
||||
threadId,
|
||||
providerSessionId: text(thread.sessionId) || null,
|
||||
context: {
|
||||
protocolVersion: PHASE4_CODEX_PROTOCOL_VERSION,
|
||||
codexVersion: text(initialize.userAgent, "unknown"),
|
||||
codexVersion: boundedText(initialize.userAgent),
|
||||
clientInfo: {
|
||||
name: "paperclip-runner",
|
||||
title: "Paperclip Runner",
|
||||
version: DRIVER_VERSION,
|
||||
},
|
||||
model: text(response.model, "unknown"),
|
||||
modelProvider: text(response.modelProvider, text(thread.modelProvider, "unknown")),
|
||||
workingDirectory: text(response.cwd, workingDirectory),
|
||||
sandbox: response.sandbox ?? null,
|
||||
approvalPolicy: response.approvalPolicy ?? "never",
|
||||
model: boundedText(response.model),
|
||||
modelProvider: boundedText(response.modelProvider, boundedText(thread.modelProvider)),
|
||||
workingDirectory,
|
||||
sandbox: {
|
||||
permissionProfile: boundedCodexValue(configuredPermissionProfile),
|
||||
legacyPolicy: boundedCodexValue(response.sandbox ?? null),
|
||||
rootAccess: "none",
|
||||
minimalRuntimeAccess: "read",
|
||||
workspaceAccess: "write",
|
||||
networkAccess: false,
|
||||
},
|
||||
approvalPolicy: boundedCodexValue(response.approvalPolicy ?? "never"),
|
||||
baseInstructions: PHASE4_SKILLLESS_BASE_INSTRUCTIONS,
|
||||
instructionSources: Array.isArray(response.instructionSources)
|
||||
? response.instructionSources.filter((value): value is string => typeof value === "string")
|
||||
? response.instructionSources
|
||||
.filter((value): value is string => typeof value === "string")
|
||||
.slice(0, 32)
|
||||
.map((value) => boundedText(value))
|
||||
: [],
|
||||
instructionPolicy: {
|
||||
skillInstructions: false,
|
||||
appInstructions: false,
|
||||
collaborationInstructions: false,
|
||||
},
|
||||
environmentKeys: sanitizedEnvironmentKeys(this.#options.environment),
|
||||
environmentKeys: Object.keys(
|
||||
commandEnvironment(this.#options.environment),
|
||||
).sort(),
|
||||
dynamicToolNames: this.#caps.dynamicTools ? [...PHASE4_SEMANTIC_TOOL_NAMES] : [],
|
||||
modelInputKinds: ["text"],
|
||||
envelope: structuredClone(this.#options.taskEnvelope),
|
||||
@@ -394,7 +515,13 @@ class CodexHarnessSession implements HarnessSession {
|
||||
#usage: Record<string, unknown> | null = null;
|
||||
#result: PrpStructuredRunResult | null = null;
|
||||
#resultFingerprint: string | null = null;
|
||||
#resultCallId: string | null = null;
|
||||
#turnStartPending = false;
|
||||
#protocolFailed = false;
|
||||
#protocolFailureCode: string | null = null;
|
||||
#protocolFailureMessage: string | null = null;
|
||||
#terminal = false;
|
||||
#turnStarted = false;
|
||||
readonly #terminalTurns = new Map<string, string>();
|
||||
|
||||
constructor(input: {
|
||||
@@ -455,17 +582,38 @@ class CodexHarnessSession implements HarnessSession {
|
||||
}
|
||||
|
||||
async startTurn(input: { message: NativeUserMessage }): Promise<{ turnId: string }> {
|
||||
if (this.#terminal || this.#protocolFailed || this.#activeTurnId !== null || this.#turnStartPending) {
|
||||
throw this.#unsupported(
|
||||
"turn start",
|
||||
this.#protocolFailureCode === null
|
||||
? "session cannot start another turn"
|
||||
: `session failed protocol validation: ${this.#protocolFailureCode} (${this.#protocolFailureMessage ?? "no detail"})`,
|
||||
);
|
||||
}
|
||||
const taskText = JSON.stringify({ task: this.#taskEnvelope, message: input.message.text });
|
||||
this.#emit("turn.submitted", { envelopeSchema: this.#taskEnvelope.schema });
|
||||
const response = await this.#transport.request("turn/start", {
|
||||
threadId: this.#opened.threadId,
|
||||
input: [userInput({ role: "user", text: taskText })],
|
||||
outputSchema: PHASE4_RESULT_OUTPUT_SCHEMA,
|
||||
});
|
||||
this.#turnStartPending = true;
|
||||
let response: Record<string, unknown>;
|
||||
try {
|
||||
response = await this.#transport.request("turn/start", {
|
||||
threadId: this.#opened.threadId,
|
||||
cwd: this.#opened.context.workingDirectory,
|
||||
permissions: SKILLLESS_PERMISSION_PROFILE,
|
||||
runtimeWorkspaceRoots: [this.#opened.context.workingDirectory],
|
||||
input: [userInput({ role: "user", text: taskText })],
|
||||
outputSchema: PHASE4_RESULT_OUTPUT_SCHEMA,
|
||||
});
|
||||
} finally {
|
||||
this.#turnStartPending = false;
|
||||
}
|
||||
const turn = record(response.turn);
|
||||
const turnId = text(turn.id);
|
||||
if (turnId.length === 0) throw new Error("Codex turn response omitted turn.id");
|
||||
this.#activeTurnId = turnId;
|
||||
if (this.#activeTurnId !== null && this.#activeTurnId !== turnId) {
|
||||
this.#failProtocol("turn_start_mismatch", "turn/start response disagreed with turn/started");
|
||||
throw new Error("Codex turn identity changed during start");
|
||||
}
|
||||
this.#activeTurnId ??= turnId;
|
||||
this.#emit("turn.accepted", { turnId }, { turnId });
|
||||
return { turnId };
|
||||
}
|
||||
@@ -525,7 +673,7 @@ class CodexHarnessSession implements HarnessSession {
|
||||
}
|
||||
const turns = Array.isArray(thread.turns) ? thread.turns.map(record) : [];
|
||||
const active = turns.find((turn) => text(turn.status) === "inProgress");
|
||||
const reconciledUsage = record(thread.tokenUsage ?? snapshot.tokenUsage);
|
||||
const reconciledUsage = boundedPayload(record(thread.tokenUsage ?? snapshot.tokenUsage));
|
||||
if (Object.keys(reconciledUsage).length > 0) this.#usage = reconciledUsage;
|
||||
const terminal = [...turns]
|
||||
.reverse()
|
||||
@@ -577,31 +725,78 @@ class CodexHarnessSession implements HarnessSession {
|
||||
code: "notification_transport_failed",
|
||||
message: redactCodexDiagnostic(String(error)),
|
||||
});
|
||||
this.#failProtocol(
|
||||
"notification_transport_failed",
|
||||
"Provider notification transport failed closed.",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#mapNotification(notification: CodexRpcNotification): void {
|
||||
if (!isBoundCodexNotification(notification.method)) return;
|
||||
const params = notification.params;
|
||||
const turn = record(params.turn);
|
||||
const item = itemFromParams(params);
|
||||
const turnId = text(params.turnId, text(turn.id, this.#activeTurnId ?? ""));
|
||||
const threadId = text(params.threadId);
|
||||
const turnId = text(params.turnId, text(turn.id));
|
||||
const itemId = text(item.id, text(params.itemId));
|
||||
if (notification.method === "error" || notification.method === "warning" || notification.method === "configWarning") {
|
||||
this.#emit("harness.diagnostic", {
|
||||
code: notification.method.replaceAll("/", "_"),
|
||||
message: redactCodexDiagnostic(text(params.message, JSON.stringify(boundedCodexValue(params)))),
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (threadId !== this.#opened.threadId) {
|
||||
this.#failProtocol(
|
||||
"thread_binding_mismatch",
|
||||
`Provider ${notification.method} message did not name the opened thread.`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (notification.method === "turn/started") {
|
||||
if (
|
||||
turnId.length === 0 ||
|
||||
this.#terminal ||
|
||||
this.#protocolFailed ||
|
||||
this.#turnStarted ||
|
||||
(!this.#turnStartPending && this.#activeTurnId === null) ||
|
||||
(this.#activeTurnId !== null && this.#activeTurnId !== turnId)
|
||||
) {
|
||||
this.#failProtocol("turn_binding_mismatch", "Provider started an unexpected turn.");
|
||||
return;
|
||||
}
|
||||
this.#activeTurnId = turnId;
|
||||
this.#turnStarted = true;
|
||||
this.#emit("turn.started", { status: text(turn.status, "inProgress") }, { turnId });
|
||||
return;
|
||||
}
|
||||
if (notification.method === "turn/completed") {
|
||||
if (this.#terminalTurns.has(turnId)) {
|
||||
this.#mapTerminalTurn(turn, turnId);
|
||||
return;
|
||||
}
|
||||
if (!this.#notificationNamesActiveTurn(turnId, "turn terminal")) return;
|
||||
this.#mapTerminalTurn(turn, turnId);
|
||||
return;
|
||||
}
|
||||
if (notification.method === "item/started") {
|
||||
this.#emit("item.started", { kind: text(item.type, "unknown"), text: itemText(item), item }, { turnId, itemId });
|
||||
if (!this.#notificationNamesActiveTurn(turnId, "item start")) return;
|
||||
this.#emit("item.started", boundedPayload({
|
||||
kind: text(item.type, "unknown"),
|
||||
text: itemText(item),
|
||||
item,
|
||||
}), { turnId, itemId });
|
||||
return;
|
||||
}
|
||||
if (notification.method === "item/completed") {
|
||||
this.#captureResultFromItem(item);
|
||||
this.#emit("item.completed", { kind: text(item.type, "unknown"), text: itemText(item), item }, { turnId, itemId });
|
||||
if (!this.#notificationNamesActiveTurn(turnId, "item completion")) return;
|
||||
this.#captureResultFromItem(item, turnId);
|
||||
this.#emit("item.completed", boundedPayload({
|
||||
kind: text(item.type, "unknown"),
|
||||
text: itemText(item),
|
||||
item,
|
||||
}), { turnId, itemId });
|
||||
return;
|
||||
}
|
||||
const deltaKinds: Record<string, string> = {
|
||||
@@ -617,41 +812,54 @@ class CodexHarnessSession implements HarnessSession {
|
||||
};
|
||||
const deltaKind = deltaKinds[notification.method];
|
||||
if (deltaKind !== undefined) {
|
||||
this.#emit("item.delta", {
|
||||
if (!this.#notificationNamesActiveTurn(turnId, "item update")) return;
|
||||
this.#emit("item.delta", boundedPayload({
|
||||
kind: deltaKind,
|
||||
text: text(params.delta, text(params.patch, text(params.output))),
|
||||
update: params,
|
||||
}, { turnId, itemId: itemId || `${turnId}:${deltaKind}` });
|
||||
}), { turnId, itemId: itemId || `${turnId}:${deltaKind}` });
|
||||
return;
|
||||
}
|
||||
if (notification.method === "thread/tokenUsage/updated") {
|
||||
this.#usage = record(params.tokenUsage);
|
||||
if (!this.#notificationNamesActiveTurn(turnId, "usage update")) return;
|
||||
this.#usage = boundedPayload(record(params.tokenUsage));
|
||||
this.#emit("item.completed", { kind: "usage", usage: this.#usage }, {
|
||||
turnId,
|
||||
itemId: `${turnId}:usage:${this.#sourceSequence + 1}`,
|
||||
});
|
||||
return;
|
||||
}
|
||||
if (notification.method === "error" || notification.method === "warning" || notification.method === "configWarning") {
|
||||
this.#emit("harness.diagnostic", {
|
||||
code: notification.method.replaceAll("/", "_"),
|
||||
message: redactCodexDiagnostic(text(params.message, JSON.stringify(params))),
|
||||
}, turnId ? { turnId } : {});
|
||||
}
|
||||
}
|
||||
|
||||
async #handleServerRequest(request: CodexRpcServerRequest): Promise<Record<string, unknown>> {
|
||||
if (request.method === "item/tool/call") {
|
||||
const tool = text(request.params.tool);
|
||||
const threadId = text(request.params.threadId);
|
||||
const turnId = text(request.params.turnId);
|
||||
const callId = text(request.params.callId);
|
||||
if (
|
||||
this.#protocolFailed ||
|
||||
this.#terminal ||
|
||||
threadId !== this.#opened.threadId ||
|
||||
turnId.length === 0 ||
|
||||
turnId !== this.#activeTurnId ||
|
||||
callId.length === 0
|
||||
) {
|
||||
this.#failProtocol("tool_binding_mismatch", "Semantic tool call did not name the active thread and turn.");
|
||||
return rejectedToolCall("Semantic tool call was outside the active thread and turn.");
|
||||
}
|
||||
if (!isSemanticTool(tool)) {
|
||||
this.#diagnoseUnsupported(`dynamic tool ${tool}`);
|
||||
return { success: false, contentItems: [{ type: "inputText", text: "Unsupported tool." }] };
|
||||
return rejectedToolCall("Unsupported tool.");
|
||||
}
|
||||
if (!isRetainableCodexPayload(request.params.arguments)) {
|
||||
return rejectedToolCall("Semantic result exceeded the retained payload limit.");
|
||||
}
|
||||
const validation = validatePrpStructuredRunResult(request.params.arguments);
|
||||
if (!validation.ok) {
|
||||
return {
|
||||
success: false,
|
||||
contentItems: [{ type: "inputText", text: `Invalid semantic result: ${validation.issues.map((issue) => issue.message).join("; ")}` }],
|
||||
contentItems: [{ type: "inputText", text: "Invalid semantic result." }],
|
||||
};
|
||||
}
|
||||
if (!toolAcceptsDisposition(tool, validation.result.reportedWorkDisposition)) {
|
||||
@@ -660,26 +868,38 @@ class CodexHarnessSession implements HarnessSession {
|
||||
contentItems: [{
|
||||
type: "inputText",
|
||||
text:
|
||||
tool === PHASE4_BLOCK_TOOL_NAME || tool === "paperclip.block"
|
||||
tool === PHASE4_BLOCK_TOOL_NAME
|
||||
? "paperclip_block requires reportedWorkDisposition=blocked."
|
||||
: "paperclip_finish accepts only done or needs_review.",
|
||||
}],
|
||||
};
|
||||
}
|
||||
const fingerprint = canonicalJson(validation.result);
|
||||
if (this.#resultCallId !== null && this.#resultCallId !== callId) {
|
||||
return rejectedToolCall("A semantic result was already committed by another call.");
|
||||
}
|
||||
if (this.#resultFingerprint !== null && this.#resultFingerprint !== fingerprint) {
|
||||
return { success: false, contentItems: [{ type: "inputText", text: "A different semantic result was already committed." }] };
|
||||
return rejectedToolCall("A different semantic result was already committed.");
|
||||
}
|
||||
if (this.#result === null) {
|
||||
this.#commitResult(
|
||||
validation.result,
|
||||
text(request.params.callId, randomUUID()),
|
||||
text(request.params.turnId, this.#activeTurnId ?? ""),
|
||||
);
|
||||
this.#resultCallId = callId;
|
||||
this.#commitResult(validation.result, callId, turnId);
|
||||
}
|
||||
return { success: true, contentItems: [{ type: "inputText", text: "Semantic completion accepted." }] };
|
||||
}
|
||||
|
||||
if (!isTurnScopedServerRequest(request.method)) return safeRequestResponse(request.method);
|
||||
const requestTurnId = text(request.params.turnId);
|
||||
if (
|
||||
text(request.params.threadId) !== this.#opened.threadId ||
|
||||
requestTurnId.length === 0 ||
|
||||
requestTurnId !== this.#activeTurnId ||
|
||||
this.#terminal ||
|
||||
this.#protocolFailed
|
||||
) {
|
||||
this.#failProtocol("runtime_request_binding_mismatch", "Runtime request did not name the active thread and turn.");
|
||||
return safeRequestResponse(request.method);
|
||||
}
|
||||
const requestKind = request.method.includes("requestApproval") ? "approval" : "user_input";
|
||||
const requestId = String(request.id);
|
||||
this.#emit("runtime_request.created", {
|
||||
@@ -689,10 +909,10 @@ class CodexHarnessSession implements HarnessSession {
|
||||
type: request.method,
|
||||
status: "pending",
|
||||
prompt: "Codex requires an external runtime decision.",
|
||||
details: redactCodexValue(request.params),
|
||||
details: redactCodexValue(boundedCodexValue(request.params)),
|
||||
},
|
||||
}, {
|
||||
turnId: text(request.params.turnId, this.#activeTurnId ?? ""),
|
||||
turnId: requestTurnId,
|
||||
itemId: text(request.params.itemId, requestId),
|
||||
});
|
||||
const response = safeRequestResponse(request.method);
|
||||
@@ -700,27 +920,18 @@ class CodexHarnessSession implements HarnessSession {
|
||||
requestId,
|
||||
resolution: "declined_by_skillless_driver",
|
||||
}, {
|
||||
turnId: text(request.params.turnId, this.#activeTurnId ?? ""),
|
||||
turnId: requestTurnId,
|
||||
itemId: text(request.params.itemId, requestId),
|
||||
});
|
||||
return response;
|
||||
}
|
||||
|
||||
#captureResultFromItem(item: Record<string, unknown>): void {
|
||||
#captureResultFromItem(item: Record<string, unknown>, turnId: string): void {
|
||||
if (text(item.type) !== "agentMessage" || this.#result !== null) return;
|
||||
if (!isRetainableCodexPayload(item.text)) return;
|
||||
const result = tryParseResult(item.text);
|
||||
if (result !== null) this.#commitResult(result, text(item.id));
|
||||
}
|
||||
|
||||
#captureResultFromTurn(turn: Record<string, unknown>): void {
|
||||
if (this.#result !== null || !Array.isArray(turn.items)) return;
|
||||
for (const value of turn.items.map(record).reverse()) {
|
||||
if (text(value.type) !== "agentMessage") continue;
|
||||
const result = tryParseResult(value.text);
|
||||
if (result !== null) {
|
||||
this.#commitResult(result, text(value.id));
|
||||
return;
|
||||
}
|
||||
if (result !== null && isRetainableCodexPayload(result)) {
|
||||
this.#commitResult(result, text(item.id), turnId);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -766,12 +977,10 @@ class CodexHarnessSession implements HarnessSession {
|
||||
});
|
||||
const previous = this.#terminalTurns.get(turnId);
|
||||
if (previous !== undefined) {
|
||||
if (previous !== fingerprint) {
|
||||
this.#emit("harness.diagnostic", {
|
||||
code: "conflicting_turn_terminal",
|
||||
message: "Provider sent materially different terminal facts for an already terminal turn.",
|
||||
}, { turnId });
|
||||
}
|
||||
this.#failProtocol(
|
||||
previous === fingerprint ? "duplicate_turn_terminal" : "conflicting_turn_terminal",
|
||||
"Provider sent another terminal fact for an already terminal turn.",
|
||||
);
|
||||
return;
|
||||
}
|
||||
this.#terminalTurns.set(turnId, fingerprint);
|
||||
@@ -786,7 +995,10 @@ class CodexHarnessSession implements HarnessSession {
|
||||
: status === "cancelled"
|
||||
? "turn.cancelled"
|
||||
: "turn.completed";
|
||||
this.#emit(eventType, { status, error: turn.error ?? null }, { turnId });
|
||||
this.#emit(eventType, boundedPayload({
|
||||
status,
|
||||
error: turn.error ?? null,
|
||||
}), { turnId });
|
||||
this.#activeTurnId = null;
|
||||
this.#finalize(status);
|
||||
}
|
||||
@@ -797,7 +1009,11 @@ class CodexHarnessSession implements HarnessSession {
|
||||
turn.items
|
||||
.map(record)
|
||||
.reverse()
|
||||
.find((value) => text(value.type) === "agentMessage" && tryParseResult(value.text) !== null) ??
|
||||
.find((value) =>
|
||||
text(value.type) === "agentMessage" &&
|
||||
isRetainableCodexPayload(value.text) &&
|
||||
tryParseResult(value.text) !== null
|
||||
) ??
|
||||
null
|
||||
);
|
||||
}
|
||||
@@ -813,6 +1029,20 @@ class CodexHarnessSession implements HarnessSession {
|
||||
}
|
||||
}
|
||||
|
||||
#notificationNamesActiveTurn(turnId: string, kind: string): boolean {
|
||||
if (
|
||||
this.#protocolFailed ||
|
||||
this.#terminal ||
|
||||
turnId.length === 0 ||
|
||||
this.#activeTurnId === null ||
|
||||
turnId !== this.#activeTurnId
|
||||
) {
|
||||
this.#failProtocol("turn_binding_mismatch", `Provider ${kind} did not name the active turn.`);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
#requireCapability(operation: keyof CodexCapabilities): void {
|
||||
if (!this.#capabilities[operation]) throw this.#unsupported(operation, "capability not advertised");
|
||||
}
|
||||
@@ -834,6 +1064,27 @@ class CodexHarnessSession implements HarnessSession {
|
||||
});
|
||||
}
|
||||
|
||||
#failProtocol(code: string, message: string): void {
|
||||
if (this.#protocolFailed) return;
|
||||
this.#protocolFailed = true;
|
||||
this.#protocolFailureCode = code;
|
||||
this.#protocolFailureMessage = redactCodexDiagnostic(message);
|
||||
this.#emit("session.failed", {
|
||||
code,
|
||||
message: redactCodexDiagnostic(message),
|
||||
recoverable: false,
|
||||
});
|
||||
if (!this.#terminal && this.#activeTurnId !== null) {
|
||||
const turnId = this.#activeTurnId;
|
||||
this.#emit("turn.failed", { status: "failed", error: { code } }, { turnId });
|
||||
this.#terminalTurns.set(turnId, canonicalJson({ protocolFailure: code }));
|
||||
this.#activeTurnId = null;
|
||||
}
|
||||
this.#terminal = true;
|
||||
this.#events.close();
|
||||
void this.#transport.close();
|
||||
}
|
||||
|
||||
#emit(
|
||||
eventType: PrpEvent["eventType"],
|
||||
payload: Record<string, unknown>,
|
||||
@@ -859,39 +1110,141 @@ class CodexHarnessSession implements HarnessSession {
|
||||
}
|
||||
}
|
||||
|
||||
function isSemanticTool(tool: string): boolean {
|
||||
return (
|
||||
tool === PHASE4_COMPLETION_TOOL_NAME ||
|
||||
tool === PHASE4_BLOCK_TOOL_NAME ||
|
||||
tool === "paperclip.finish" ||
|
||||
tool === "paperclip.block"
|
||||
function validateWorkingDirectory(
|
||||
workingDirectory: string,
|
||||
environment: NodeJS.ProcessEnv = process.env,
|
||||
): string {
|
||||
if (workingDirectory.trim().length === 0) {
|
||||
throw new Error("Codex working directory is required");
|
||||
}
|
||||
const resolved = resolve(workingDirectory);
|
||||
if (resolved === parse(resolved).root) {
|
||||
throw new Error("Codex working directory cannot be a filesystem root");
|
||||
}
|
||||
const hostHome = environment.HOME?.trim();
|
||||
if (hostHome && pathContains(resolved, resolve(hostHome))) {
|
||||
throw new Error("Codex working directory cannot contain the host HOME");
|
||||
}
|
||||
const codexHome = environment.CODEX_HOME?.trim();
|
||||
if (codexHome) {
|
||||
const resolvedCodexHome = resolve(codexHome);
|
||||
if (pathContains(resolved, resolvedCodexHome) || pathContains(resolvedCodexHome, resolved)) {
|
||||
throw new Error("Codex working directory cannot overlap host CODEX_HOME");
|
||||
}
|
||||
}
|
||||
const configuredRoot = environment.PAPERCLIP_WORKSPACE_CWD;
|
||||
if (configuredRoot !== undefined && configuredRoot.trim().length > 0) {
|
||||
const root = resolve(configuredRoot);
|
||||
const pathFromRoot = relative(root, resolved);
|
||||
if (pathFromRoot === ".." || pathFromRoot.startsWith(`..${process.platform === "win32" ? "\\" : "/"}`)) {
|
||||
throw new Error("Codex working directory is outside the assigned workspace");
|
||||
}
|
||||
}
|
||||
return resolved;
|
||||
}
|
||||
|
||||
function pathContains(parent: string, candidate: string): boolean {
|
||||
const fromParent = relative(parent, candidate);
|
||||
return fromParent === "" || (
|
||||
fromParent !== ".." &&
|
||||
!fromParent.startsWith(`..${process.platform === "win32" ? "\\" : "/"}`) &&
|
||||
!parse(fromParent).root
|
||||
);
|
||||
}
|
||||
|
||||
function boundedCodexValue(value: unknown, depth = 0): unknown {
|
||||
if (depth > 8) return { truncated: true, reason: "maximum depth" };
|
||||
if (typeof value === "string") {
|
||||
return value.length <= MAX_RETAINED_CODEX_STRING_CHARS
|
||||
? value
|
||||
: `${value.slice(0, MAX_RETAINED_CODEX_STRING_CHARS)}...[truncated]`;
|
||||
}
|
||||
if (Array.isArray(value)) {
|
||||
const bounded = value
|
||||
.slice(0, 128)
|
||||
.map((entry) => boundedCodexValue(entry, depth + 1));
|
||||
if (value.length > 128) bounded.push({ truncated: true, omittedItems: value.length - 128 });
|
||||
return bounded;
|
||||
}
|
||||
if (typeof value !== "object" || value === null) return value;
|
||||
const object = record(value);
|
||||
const bounded = Object.fromEntries(
|
||||
Object.entries(object)
|
||||
.slice(0, 128)
|
||||
.map(([key, entry]) => [key, boundedCodexValue(entry, depth + 1)]),
|
||||
);
|
||||
if (Object.keys(object).length > 128) bounded.truncatedEntries = Object.keys(object).length - 128;
|
||||
const serialized = JSON.stringify(bounded);
|
||||
return Buffer.byteLength(serialized) <= MAX_RETAINED_CODEX_PAYLOAD_BYTES
|
||||
? bounded
|
||||
: { truncated: true, byteSize: Buffer.byteLength(serialized) };
|
||||
}
|
||||
|
||||
function boundedPayload(value: Record<string, unknown>): Record<string, unknown> {
|
||||
return record(boundedCodexValue(value));
|
||||
}
|
||||
|
||||
function isRetainableCodexPayload(value: unknown): boolean {
|
||||
try {
|
||||
return Buffer.byteLength(JSON.stringify(value)) <= MAX_RETAINED_CODEX_PAYLOAD_BYTES;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function isSemanticTool(tool: string): boolean {
|
||||
return tool === PHASE4_COMPLETION_TOOL_NAME || tool === PHASE4_BLOCK_TOOL_NAME;
|
||||
}
|
||||
|
||||
function toolAcceptsDisposition(
|
||||
tool: string,
|
||||
disposition: PrpStructuredRunResult["reportedWorkDisposition"],
|
||||
): boolean {
|
||||
if (tool === PHASE4_BLOCK_TOOL_NAME || tool === "paperclip.block") {
|
||||
if (tool === PHASE4_BLOCK_TOOL_NAME) {
|
||||
return disposition === "blocked";
|
||||
}
|
||||
return disposition === "done" || disposition === "needs_review";
|
||||
}
|
||||
|
||||
function redactCodexValue(value: unknown): unknown {
|
||||
function redactCodexValue(value: unknown, depth = 0): unknown {
|
||||
if (depth > 8) return "[TRUNCATED]";
|
||||
if (typeof value === "string") return redactCodexDiagnostic(value);
|
||||
if (Array.isArray(value)) return value.map(redactCodexValue);
|
||||
if (Array.isArray(value)) {
|
||||
return value.slice(0, 128).map((entry) => redactCodexValue(entry, depth + 1));
|
||||
}
|
||||
const object = record(value);
|
||||
return Object.fromEntries(
|
||||
Object.entries(object).map(([key, entry]) => [
|
||||
Object.entries(object).slice(0, 128).map(([key, entry]) => [
|
||||
key,
|
||||
/(?:api[_-]?key|token|secret|password|authorization)/i.test(key)
|
||||
? "[REDACTED]"
|
||||
: redactCodexValue(entry),
|
||||
: redactCodexValue(entry, depth + 1),
|
||||
]),
|
||||
);
|
||||
}
|
||||
|
||||
function rejectedToolCall(message: string): Record<string, unknown> {
|
||||
return { success: false, contentItems: [{ type: "inputText", text: message }] };
|
||||
}
|
||||
|
||||
function isTurnScopedServerRequest(method: string): boolean {
|
||||
return method.startsWith("item/") || method.startsWith("tool/") || method === "mcpServer/elicitation/request";
|
||||
}
|
||||
|
||||
function isBoundCodexNotification(method: string): boolean {
|
||||
return (
|
||||
method === "turn/started" ||
|
||||
method === "turn/completed" ||
|
||||
method === "thread/tokenUsage/updated" ||
|
||||
method === "error" ||
|
||||
method === "warning" ||
|
||||
method === "configWarning" ||
|
||||
method.startsWith("item/") ||
|
||||
method === "turn/diff/updated" ||
|
||||
method === "turn/plan/updated"
|
||||
);
|
||||
}
|
||||
|
||||
function safeRequestResponse(method: string): Record<string, unknown> {
|
||||
if (method === "item/permissions/requestApproval") {
|
||||
return { permissions: {} };
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
createSessionSnapshotFromMetadata,
|
||||
} from "../reducer/session-reducer.js";
|
||||
import {
|
||||
validatePrpEvent,
|
||||
validatePrpStructuredRunResult,
|
||||
type PrpCapabilities,
|
||||
type PrpEvent,
|
||||
@@ -20,6 +21,10 @@ import {
|
||||
type PrpTerminalState,
|
||||
} from "../protocol/phase1-contract.js";
|
||||
|
||||
const MAX_PERSISTED_PHASE4_BYTES = 8 * 1024 * 1024;
|
||||
const MAX_PERSISTED_PHASE4_LINE_BYTES = 256 * 1024;
|
||||
const MAX_PERSISTED_PHASE4_EVENTS = 4096;
|
||||
|
||||
export interface Phase4CodexTracerInput {
|
||||
driver: HarnessDriver;
|
||||
taskEnvelope: Phase4TaskEnvelope;
|
||||
@@ -239,6 +244,78 @@ function sourceSequenceContinuous(events: PrpEvent[]): boolean {
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Parse and validate the persisted boundary before replaying controller state. */
|
||||
export function replayPersistedPhase4Events(
|
||||
serialized: string,
|
||||
metadata: Phase4RunMetadata,
|
||||
) {
|
||||
if (!metadata.identity.driverSessionId) {
|
||||
throw new Error("Persisted Phase 4 replay requires a driver session identity");
|
||||
}
|
||||
if (Buffer.byteLength(serialized) > MAX_PERSISTED_PHASE4_BYTES) {
|
||||
throw new Error("Persisted Phase 4 event stream exceeded its byte limit");
|
||||
}
|
||||
const lines = serialized.endsWith("\n")
|
||||
? serialized.slice(0, -1).split("\n")
|
||||
: serialized.split("\n");
|
||||
if (lines.length === 0 || lines.length > MAX_PERSISTED_PHASE4_EVENTS) {
|
||||
throw new Error("Persisted Phase 4 event stream exceeded its event limit");
|
||||
}
|
||||
const events: PrpEvent[] = [];
|
||||
const sourceEventIds = new Set<string>();
|
||||
const sourceSequences = new Map<string, number>();
|
||||
let runTerminalCount = 0;
|
||||
for (const line of lines) {
|
||||
if (line.length === 0 || Buffer.byteLength(line) > MAX_PERSISTED_PHASE4_LINE_BYTES) {
|
||||
throw new Error("Persisted Phase 4 event line was empty or oversized");
|
||||
}
|
||||
let candidate: unknown;
|
||||
try {
|
||||
candidate = JSON.parse(line);
|
||||
} catch {
|
||||
throw new Error("Persisted Phase 4 event line was malformed JSON");
|
||||
}
|
||||
const validation = validatePrpEvent(candidate);
|
||||
if (!validation.ok) throw new Error("Persisted Phase 4 event failed schema validation");
|
||||
const event = validation.event;
|
||||
if (
|
||||
event.runId !== metadata.identity.runId ||
|
||||
event.normalizedSessionId !== metadata.identity.normalizedSessionId
|
||||
) {
|
||||
throw new Error("Persisted Phase 4 event identity did not match the opened run");
|
||||
}
|
||||
if (sourceEventIds.has(event.sourceEventId)) {
|
||||
throw new Error("Persisted Phase 4 event id was duplicated");
|
||||
}
|
||||
sourceEventIds.add(event.sourceEventId);
|
||||
const source = `${event.sourceKind}:${event.sourceInstanceId}`;
|
||||
const previous = sourceSequences.get(source);
|
||||
if (previous !== undefined && event.sourceSeq !== previous + 1) {
|
||||
throw new Error("Persisted Phase 4 source sequence was discontinuous");
|
||||
}
|
||||
sourceSequences.set(source, event.sourceSeq);
|
||||
if (runTerminalCount > 0) {
|
||||
throw new Error("Persisted Phase 4 event appeared after the run terminal");
|
||||
}
|
||||
if (event.eventType === "run.terminal") runTerminalCount += 1;
|
||||
events.push(event);
|
||||
}
|
||||
if (runTerminalCount !== 1) {
|
||||
throw new Error("Persisted Phase 4 event stream requires exactly one run terminal");
|
||||
}
|
||||
return events.reduce(
|
||||
applyPrpEvent,
|
||||
createSessionSnapshotFromMetadata({
|
||||
fixtureName: metadata.fixtureName,
|
||||
identity: {
|
||||
...metadata.identity,
|
||||
driverSessionId: metadata.identity.driverSessionId,
|
||||
},
|
||||
capabilities: metadata.capabilities,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Small in-memory Phase 4 mock core. The driver reports provider facts; this
|
||||
* controller validates semantic proposals and alone emits the run terminal.
|
||||
@@ -404,14 +481,8 @@ export async function runPhase4CodexTracer(input: Phase4CodexTracerInput): Promi
|
||||
capabilities,
|
||||
}),
|
||||
);
|
||||
const replaySnapshot = events.reduce(
|
||||
applyPrpEvent,
|
||||
createSessionSnapshotFromMetadata({
|
||||
fixtureName: metadata.fixtureName,
|
||||
identity,
|
||||
capabilities,
|
||||
}),
|
||||
);
|
||||
const persistedEvents = `${events.map((event) => JSON.stringify(event)).join("\n")}\n`;
|
||||
const replaySnapshot = replayPersistedPhase4Events(persistedEvents, metadata);
|
||||
const context = contextFrom(events);
|
||||
const terminalCount = events.filter((event) => event.eventType === "run.terminal").length;
|
||||
const turnTerminalCount = events.filter(isTurnTerminal).length;
|
||||
@@ -447,7 +518,10 @@ export async function runPhase4CodexTracer(input: Phase4CodexTracerInput): Promi
|
||||
liveReplayParity: JSON.stringify(liveSnapshot) === JSON.stringify(replaySnapshot),
|
||||
stableIdentity:
|
||||
providerSessionId !== null &&
|
||||
new Set([runId, normalizedSessionId, ids.driverSessionId, providerSessionId]).size === 4 &&
|
||||
runId.length > 0 &&
|
||||
normalizedSessionId.length > 0 &&
|
||||
ids.driverSessionId.length > 0 &&
|
||||
providerSessionId.length > 0 &&
|
||||
events.every(
|
||||
(event) =>
|
||||
event.runId === runId && event.normalizedSessionId === normalizedSessionId,
|
||||
|
||||
Reference in new issue
Block a user