From aa87487a2ccf95a634167bd7a319eaaf37216aee Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 10 Aug 2026 03:16:57 +0000 Subject: [PATCH] feat(runner): add live Phase 7 Codex tool loop Start package-local runnerd and Codex app-server processes, bind the authorized semantic catalog to the mock ControlPlanePort, persist resumable sessions and interactions, and enforce bounded stop/reset cleanup. Co-Authored-By: Paperclip --- .gitignore | 1 + packages/paperclip-runner/README.md | 8 +- packages/paperclip-runner/docs/index.md | 1 + .../docs/phase-07-live-runnerd-codex.md | 84 ++ packages/paperclip-runner/package.json | 2 + .../runner-core/src/bin/paperclip-runnerd.rs | 81 +- .../paperclip-runner/src/cli/phase7-live.ts | 115 +++ .../src/drivers/codex/app-server-transport.ts | 88 +- packages/paperclip-runner/src/index.ts | 1 + packages/paperclip-runner/src/phase7/index.ts | 2 + .../src/phase7/live-session.test.ts | 325 +++++++ .../src/phase7/live-session.ts | 873 ++++++++++++++++++ .../src/phase7/runnerd-codex-transport.ts | 106 +++ 13 files changed, 1677 insertions(+), 10 deletions(-) create mode 100644 packages/paperclip-runner/docs/phase-07-live-runnerd-codex.md create mode 100644 packages/paperclip-runner/src/cli/phase7-live.ts create mode 100644 packages/paperclip-runner/src/phase7/index.ts create mode 100644 packages/paperclip-runner/src/phase7/live-session.test.ts create mode 100644 packages/paperclip-runner/src/phase7/live-session.ts create mode 100644 packages/paperclip-runner/src/phase7/runnerd-codex-transport.ts diff --git a/.gitignore b/.gitignore index f7f6ae75b1f..3f4ade4d65e 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ node_modules/ **/node_modules **/node_modules/ dist/ +**/target/ ui/storybook-static/ .env *.tsbuildinfo diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 20762d7d4d5..ecaa9c06113 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -34,7 +34,10 @@ committed coordinator makes replay a no-op. Phase 7D adds the stable semantic-tool catalog, actor/task/mode/scenario policy, invocation-time dispatcher checks, typed denials, redacted authorization records, and one provider-neutral contract shared by fake and live Codex -bindings. The implementation remains mock-only and package-local. +bindings. Phase 7E starts a real package-local runnerd and Codex app-server, +routes allowed semantic calls through the mock `ControlPlanePort`, resumes the +same provider thread, and performs bounded process-group cleanup. The data plane +remains mock-only and package-local. ## Phase 0–5 quick start @@ -123,6 +126,8 @@ and JSON content; see the protocol-server tutorial for direct `curl` examples. | `check:phase-07-contract` | Verify the Phase 7A skill, MCP, and eval traceability contract. | | `generate:semantic-contracts` | Compile and regenerate the provider-neutral Phase 7D tool contract. | | `check:semantic-contracts` | Fail when the checked-in Phase 7D tool contract is stale. | +| `test:phase7` | Prove the Phase 7 mock adapter, semantic policy, live session persistence, interaction resume, stop, and reset contracts. | +| `trace:phase7` | Run a real runnerd and Codex app-server semantic-tool smoke with process and network evidence. | | `test:browser:phase5` | Exercise both consumers with the fake driver, keyboard/a11y checks, reconnect/replay, measurements, and screenshots. | | `record:phase5:codex` | Run both public consumers against a safe real Codex session and capture live screenshots. | | `browser:dev` | Start the standalone live/replay/recovery browser devtool. | @@ -150,6 +155,7 @@ and JSON content; see the protocol-server tutorial for direct `curl` examples. - [Phase 5 browser SDK reference](docs/phase-05-sdk.md) - [Phase 5 component decision record](docs/design/phase-5-component-decisions.md) - [Phase 7 semantic catalog and authorization](docs/phase-07-semantic-catalog.md) +- [Phase 7 live runnerd and Codex loop](docs/phase-07-live-runnerd-codex.md) - [PRP compatibility/versioning policy](docs/protocol-compatibility.md) - [Cumulative end-to-end tutorial](docs/tutorials/end-to-end.md) - [Journal guide](docs/journal.md) diff --git a/packages/paperclip-runner/docs/index.md b/packages/paperclip-runner/docs/index.md index 3ea3fac9f54..3d4f0f76699 100644 --- a/packages/paperclip-runner/docs/index.md +++ b/packages/paperclip-runner/docs/index.md @@ -28,6 +28,7 @@ Each tutorial starts with an explanation of what the phase is and what the phase - [Phase 5 browser SDK and reference console](phase-05-sdk.md) - [Phase 6 thin Paperclip adapter](phase-06-thin-paperclip-adapter.md) - [Phase 7 semantic catalog and authorization](phase-07-semantic-catalog.md) +- [Phase 7 live runnerd and Codex loop](phase-07-live-runnerd-codex.md) - [Engineering journal guide](journal.md) - [Dated shadcn/ui and AI Elements compatibility note](research/2026-08-07-ui-library-compatibility.md) - [Phase 4b live-console interaction map](design/phase-4b-interaction-map.md) diff --git a/packages/paperclip-runner/docs/phase-07-live-runnerd-codex.md b/packages/paperclip-runner/docs/phase-07-live-runnerd-codex.md new file mode 100644 index 00000000000..f14dc31a078 --- /dev/null +++ b/packages/paperclip-runner/docs/phase-07-live-runnerd-codex.md @@ -0,0 +1,84 @@ +# Phase 7 live runnerd and Codex loop + +Phase 7E binds the provider-neutral semantic catalog to a real package-local +`paperclip-runnerd` process and a real Codex app-server session. Paperclip data +remains deterministic mock state behind `ControlPlanePort`; no request reaches +the Paperclip API. + +## Process and authority boundary + +The process chain is: + +```text +Phase7LiveSession -> paperclip-runnerd -> codex app-server + -> Phase7SemanticDispatcher -> ControlPlanePort mock +``` + +`paperclip-runnerd` owns the Codex child and proxies newline-delimited JSON-RPC +over stdio. The transport starts a dedicated Unix process group. Normal close, +stop/reset cleanup, and fatal protocol errors terminate that group with a +bounded TERM/KILL sequence so a Codex child is not abandoned. + +Only the allowlisted Codex host environment is copied. `PAPERCLIP_*` variables, +provider credentials other than Codex's server-side home, and credentialed +proxy URLs are not passed to runnerd or Codex. Model-issued commands still use +the separate skillless, network-disabled workspace permission profile. + +## Stable session API + +Import the package root and construct `Phase7LiveSessionService`: + +```ts +import { Phase7LiveSessionService } from "@paperclipai/paperclip-runner"; + +const service = new Phase7LiveSessionService({ store }); +const session = await service.create({ workingDirectory }); +const turn = await session.sendMessage("Read the mock task and report progress."); +await session.interrupt(); // only when a turn is active +await service.stop(session.id); +``` + +The handoff surface for later tracks is: + +- `create(input)` starts runnerd, Codex, one mock run, and one dynamic-tool thread. +- `sendMessage(text)` supports repeated turns on the same provider thread. +- `pendingInteractions()` and `resolveInteraction(input)` preserve typed human + interactions and return their results to that same thread. +- `reconnect(sessionId)` closes the old process group, starts a fresh runnerd and + Codex app-server, then reads and resumes the persisted provider thread. +- `restore(sessionId)` recreates mock state, transcript, authority, authorization + records, pending interactions, and the provider thread from a stored snapshot. +- `interrupt(reason)` cancels only the active turn and retains session authority. +- `service.stop(sessionId)` clears authority and reaps the process group; + `reset` also deletes the old snapshot and restores the original clean mock + seed under new run/session authority. + +Supply a durable implementation of `Phase7LiveSessionStore` for refresh and +server-restart survival. The included in-memory store is intended for tests and +single-process consumers. + +Every snapshot includes bounded transcript/evidence, serialized mock state, +semantic authorization records, runner/Codex PIDs and exit state, and explicit +network evidence. Tool calls are admitted only when their thread and turn match +the active Codex turn. Their typed `Phase7SemanticToolResult` is serialized into +the app-server response, allowing Codex to use the resulting state revision in +its next response. + +## Verification + +Run the deterministic contract suite: + +```sh +pnpm --filter @paperclipai/paperclip-runner test:phase7 +``` + +Run a real runnerd and Codex app-server smoke: + +```sh +pnpm --filter @paperclipai/paperclip-runner trace:phase7 -- --json +``` + +The smoke requires an authenticated local Codex installation. It checks a real +semantic tool mutation, typed-result response, same-thread second turn, process +ownership/cleanup, cleared authority, and zero Paperclip network/child-env +exposure. diff --git a/packages/paperclip-runner/package.json b/packages/paperclip-runner/package.json index b89ffbab062..b0bc5633dc1 100644 --- a/packages/paperclip-runner/package.json +++ b/packages/paperclip-runner/package.json @@ -65,6 +65,8 @@ "trace:phase3": "pnpm run build:typescript && pnpm run build:phase2-binaries && node dist/cli/phase3-recovery.js", "trace:phase4": "pnpm run build:typescript && node dist/cli/phase4-codex.js", "trace:phase6": "pnpm run build:typescript && node dist/cli/phase6-paperclip.js", + "trace:phase7": "pnpm run build:typescript && cargo build --manifest-path runner/Cargo.toml --locked -p paperclip-runner-core --bin paperclip-runnerd && node dist/cli/phase7-live.js", + "test:phase7": "vitest run src/phase7/live-session.test.ts src/semantic-tools/semantic-tools.test.ts src/mock-core/phase7-mock-control-plane-adapter.test.ts", "demo:phase4b": "pnpm run build:typescript && node scripts/phase4b-demo-server.mjs", "record:phase4b": "pnpm run build:typescript && node scripts/record-phase4b-evidence.mjs", "record:phase4": "pnpm run build:typescript && node scripts/record-phase4-evidence.mjs", diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs index f69993d6a9e..18fe39f096c 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/paperclip-runnerd.rs @@ -1,5 +1,5 @@ use std::path::PathBuf; -use std::process::ExitCode; +use std::process::{Command, ExitCode, Stdio}; use std::time::Duration; use paperclip_runner_core::phase2::{run_local_runner, Phase2Error, RunnerConfig}; @@ -43,6 +43,82 @@ fn duration_value(args: &[String], name: &str, default: u64) -> Result Result, Phase2Error> { + let mut values = Vec::new(); + let mut index = 0; + while index < args.len() { + if args[index] == name { + let value = args + .get(index + 1) + .ok_or_else(|| Phase2Error::invalid(format!("missing value for {name}")))?; + values.push(value.clone()); + index += 2; + } else { + index += 1; + } + } + Ok(values) +} + +/// Phase 7 keeps runnerd as the package-local process owner while the existing +/// Codex app-server remains the provider protocol implementation. Stdio stays +/// byte-for-byte JSON-RPC; runnerd writes process evidence only to stderr. +fn run_codex_app_server_proxy(args: &[String]) -> Result<(), Phase2Error> { + let command_name = value(args, "--codex-command")?; + let codex_args = repeated_values(args, "--codex-arg")?; + let mut command = Command::new(&command_name); + command + .args(&codex_args) + .env_clear() + .stdin(Stdio::inherit()) + .stdout(Stdio::inherit()) + .stderr(Stdio::inherit()); + for key in [ + "ALL_PROXY", + "CODEX_HOME", + "HOME", + "HTTP_PROXY", + "HTTPS_PROXY", + "LANG", + "LC_ALL", + "NO_PROXY", + "NODE_EXTRA_CA_CERTS", + "PATH", + "PATHEXT", + "SSL_CERT_FILE", + "SystemRoot", + "TEMP", + "TMP", + "TMPDIR", + "WINDIR", + ] { + if let Some(value) = std::env::var_os(key) { + command.env(key, value); + } + } + let mut child = command.spawn().map_err(|error| { + Phase2Error::invalid(format!("failed to start Codex app-server: {error}")) + })?; + eprintln!( + "paperclip-runnerd: phase7 codex proxy started runner_pid={} codex_pid={}", + std::process::id(), + child.id() + ); + let status = child.wait().map_err(|error| { + Phase2Error::invalid(format!("failed to wait for Codex app-server: {error}")) + })?; + eprintln!( + "paperclip-runnerd: phase7 codex proxy exited success={} code={}", + status.success(), + status.code().map_or_else(|| "signal".to_owned(), |code| code.to_string()) + ); + if status.success() { + Ok(()) + } else { + Err(Phase2Error::invalid("Codex app-server exited unsuccessfully")) + } +} + fn run_phase3( args: &[String], bootstrap_ticket: Option, @@ -82,6 +158,9 @@ fn run_phase3( fn run(bootstrap_ticket: Option) -> Result<(), Phase2Error> { let args = std::env::args().skip(1).collect::>(); + if args.iter().any(|argument| argument == "--codex-app-server-proxy") { + return run_codex_app_server_proxy(&args); + } if args.iter().any(|argument| argument == "--connect-url") { return run_phase3(&args, bootstrap_ticket); } diff --git a/packages/paperclip-runner/src/cli/phase7-live.ts b/packages/paperclip-runner/src/cli/phase7-live.ts new file mode 100644 index 00000000000..4b361fad769 --- /dev/null +++ b/packages/paperclip-runner/src/cli/phase7-live.ts @@ -0,0 +1,115 @@ +#!/usr/bin/env node +import { resolve } from "node:path"; + +import { + Phase7LiveSessionService, + type Phase7LiveSession, +} from "../phase7/live-session.js"; + +interface CliOptions { + json: boolean; + workingDirectory: string; +} + +function parseArgs(args: string[]): CliOptions { + const options: CliOptions = { json: false, workingDirectory: process.cwd() }; + for (let index = 0; index < args.length; index += 1) { + const value = args[index]; + if (value === "--") continue; + if (value === "--json") options.json = true; + else if (value === "--working-directory") { + const directory = args[++index]; + if (!directory) throw new Error("--working-directory requires a path"); + options.workingDirectory = resolve(directory); + } else { + throw new Error(`Unknown argument: ${value}`); + } + } + return options; +} + +function operationRevision(session: Phase7LiveSession): number | null { + const result = session.snapshot().evidence.findLast((entry) => entry.kind === "tool_result")?.data.result; + if (typeof result !== "object" || result === null || Array.isArray(result)) return null; + return typeof result.stateRevision === "number" ? result.stateRevision : null; +} + +async function main(): Promise { + const options = parseArgs(process.argv.slice(2)); + const service = new Phase7LiveSessionService({ + transportOptions: { + onDiagnostic: (message) => process.stderr.write(`[phase7 runnerd] ${message}\n`), + }, + }); + let session: Phase7LiveSession | null = null; + try { + session = await service.create({ + workingDirectory: options.workingDirectory, + turnTimeoutMs: 180_000, + }); + const first = await session.sendMessage([ + "Call report_progress exactly once with idempotencyKey phase7-live-smoke-1", + "and body 'Real Codex changed mock state through ControlPlanePort.'.", + "After the tool returns, reply with the exact stateRevision from its typed result.", + "Do not call any other tool.", + ].join(" ")); + const providerThreadId = first.snapshot.providerThreadId; + const firstRunnerPid = first.snapshot.process?.runnerPid ?? null; + const revision = operationRevision(session); + await session.reconnect(); + const reconnected = session.snapshot(); + const second = await session.sendMessage( + "Without calling another tool, confirm that the prior typed result changed mock state and repeat its stateRevision.", + ); + const beforeShutdown = session.snapshot(); + const comment = session.mockState().comments.at(-1); + await service.shutdown(session.id, "Phase 7 live smoke complete"); + const afterShutdown = session.snapshot(); + const assertions = { + realRunnerdStarted: beforeShutdown.process?.runnerPid !== null, + realCodexStarted: beforeShutdown.process?.codexPid !== null, + sameProviderThread: second.snapshot.providerThreadId === providerThreadId, + reconnectStartedNewRunner: + firstRunnerPid !== null && reconnected.process?.runnerPid !== firstRunnerPid, + reconnectReapedOldRunner: reconnected.evidence.some((entry) => + entry.kind === "process" && + entry.data.action === "transport_closed" && + entry.data.reason === "reconnect" && + entry.data.runnerExited === true), + mockStateChanged: comment?.body === "Real Codex changed mock state through ControlPlanePort.", + typedResultChangedResponse: + revision !== null && first.assistantText.includes(String(revision)), + resumedResponseUsesTypedResult: + revision !== null && second.assistantText.includes(String(revision)), + noRealPaperclipRequest: beforeShutdown.networkEvidence.realPaperclipRequests === 0, + noPaperclipAuthorityInChild: + beforeShutdown.networkEvidence.childPaperclipEnvironmentKeys.length === 0, + authorityCleared: afterShutdown.authority.active === false, + runnerExited: afterShutdown.process?.runnerExited === true, + }; + const output = { + schema: "paperclip.phase7.live-smoke.v1", + providerThreadId, + firstTurnId: first.turnId, + secondTurnId: second.turnId, + stateRevision: revision, + firstRunnerPid, + process: afterShutdown.process, + networkEvidence: afterShutdown.networkEvidence, + assertions, + }; + process.stdout.write(options.json + ? `${JSON.stringify(output, null, 2)}\n` + : `${Object.entries(assertions).map(([key, value]) => `${value ? "PASS" : "FAIL"} ${key}`).join("\n")}\n`); + if (Object.values(assertions).some((passed) => !passed)) process.exitCode = 1; + } finally { + if (session !== null && session.snapshot().status !== "closed") { + await service.shutdown(session.id, "Phase 7 live smoke cleanup"); + } + } +} + +void main().catch((error: unknown) => { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + process.exitCode = 1; +}); diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index 286ba131c8f..541d4905ec8 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -19,6 +19,15 @@ export interface CodexAppServerTransport { notifications(): AsyncIterable; setServerRequestHandler(handler: CodexServerRequestHandler): void; close(): Promise; + processInfo?(): CodexTransportProcessInfo; +} + +export interface CodexTransportProcessInfo { + pid: number | null; + processGroupId: number | null; + exited: boolean; + exitCode: number | null; + signal: NodeJS.Signals | null; } class BoundedAsyncQueue implements AsyncIterable { @@ -260,6 +269,10 @@ export interface ProcessCodexTransportOptions { maxQueuedNotifications?: number; maxQueuedNotificationBytes?: number; maxInflightServerRequests?: number; + /** Starts a dedicated process group so runnerd and its Codex child are reaped together. */ + processGroup?: boolean; + closeGraceMs?: number; + onProcess?: (info: CodexTransportProcessInfo) => void; } const DEFAULT_MAX_LINE_BYTES = 1024 * 1024; @@ -281,6 +294,14 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { }); #inflightServerRequests = 0; #closed = false; + #exited = false; + #exitCode: number | null = null; + #exitSignal: NodeJS.Signals | null = null; + #resolveExit!: () => void; + readonly #exitPromise: Promise; + readonly #processGroup: boolean; + readonly #closeGraceMs: number; + readonly #onProcess?: (info: CodexTransportProcessInfo) => void; #onDiagnostic?: (message: string) => void; readonly #maxLineBytes: number; readonly #maxPendingRequests: number; @@ -299,6 +320,12 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { options.maxInflightServerRequests, DEFAULT_MAX_INFLIGHT_SERVER_REQUESTS, ); + this.#processGroup = options.processGroup === true && process.platform !== "win32"; + this.#closeGraceMs = positiveLimit(options.closeGraceMs, 1_000); + this.#onProcess = options.onProcess; + this.#exitPromise = new Promise((resolve) => { + this.#resolveExit = resolve; + }); this.#notifications = new BoundedAsyncQueue( positiveLimit(options.maxQueuedNotifications, DEFAULT_MAX_QUEUED_NOTIFICATIONS), positiveLimit( @@ -309,7 +336,9 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { this.#process = spawn(options.command ?? "codex", options.args ?? ["app-server"], { env: options.environment ?? createSanitizedCodexEnvironment(), stdio: "pipe", + detached: this.#processGroup, }); + this.#onProcess?.(this.processInfo()); this.#stdoutDecoder = new BoundedLineDecoder( this.#maxLineBytes, (line, bytes) => this.#onLine(line, bytes), @@ -326,6 +355,11 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { this.#process.stderr.on("end", () => this.#stderrDecoder.end()); this.#process.on("error", (error) => this.#fatal(error)); this.#process.on("exit", (code, signal) => { + this.#exited = true; + this.#exitCode = code; + this.#exitSignal = signal; + this.#resolveExit(); + this.#onProcess?.(this.processInfo()); if (!this.#closed) { this.#fatal( new Error( @@ -368,13 +402,51 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { } async close(): Promise { - 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")); + const firstClose = !this.#closed; + if (firstClose) { + this.#closed = true; + this.#stdoutDecoder.close(); + this.#stderrDecoder.close(); + this.#notifications.close(); + this.#failAll(new Error("codex app-server transport closed")); + } + if (!this.#exited) { + if (firstClose) this.#signal("SIGTERM"); + const exited = await Promise.race([ + this.#exitPromise.then(() => true), + new Promise((resolve) => setTimeout(() => resolve(false), this.#closeGraceMs)), + ]); + if (!exited && !this.#exited) { + this.#signal("SIGKILL"); + await Promise.race([ + this.#exitPromise, + new Promise((resolve) => setTimeout(resolve, this.#closeGraceMs)), + ]); + } + } + } + + processInfo(): CodexTransportProcessInfo { + return { + pid: this.#process.pid ?? null, + processGroupId: this.#processGroup ? this.#process.pid ?? null : null, + exited: this.#exited, + exitCode: this.#exitCode, + signal: this.#exitSignal, + }; + } + + #signal(signal: NodeJS.Signals): void { + const pid = this.#process.pid; + if (this.#processGroup && pid !== undefined) { + try { + process.kill(-pid, signal); + return; + } catch { + // The group may already be gone; the direct child kill below is idempotent. + } + } + this.#process.kill(signal); } #write(message: unknown): void { @@ -479,7 +551,7 @@ export class ProcessCodexAppServerTransport implements CodexAppServerTransport { this.#closed = true; this.#stdoutDecoder.close(); this.#stderrDecoder.close(); - this.#process.kill("SIGKILL"); + this.#signal("SIGKILL"); this.#notifications.close(error); this.#failAll(error); this.#onDiagnostic?.(redactCodexDiagnostic(error.message)); diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index d08b3688a49..544c667095b 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -30,3 +30,4 @@ export * from "./tracer/phase0-runner.js"; export * from "./tracer/phase1-replay.js"; export * from "./tracer/phase2-live.js"; export * from "./semantic-tools/index.js"; +export * from "./phase7/index.js"; diff --git a/packages/paperclip-runner/src/phase7/index.ts b/packages/paperclip-runner/src/phase7/index.ts new file mode 100644 index 00000000000..f12d7f47f52 --- /dev/null +++ b/packages/paperclip-runner/src/phase7/index.ts @@ -0,0 +1,2 @@ +export * from "./live-session.js"; +export * from "./runnerd-codex-transport.js"; diff --git a/packages/paperclip-runner/src/phase7/live-session.test.ts b/packages/paperclip-runner/src/phase7/live-session.test.ts new file mode 100644 index 00000000000..8d3258d29a3 --- /dev/null +++ b/packages/paperclip-runner/src/phase7/live-session.test.ts @@ -0,0 +1,325 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { + CodexAppServerTransport, + CodexRpcNotification, + CodexRpcServerRequest, + CodexServerRequestHandler, +} from "../drivers/codex/app-server-transport.js"; +import { + InMemoryPhase7LiveSessionStore, + Phase7LiveSessionService, + type Phase7LiveTransportFactory, +} from "./live-session.js"; + +class AsyncNotifications implements AsyncIterable { + #values: CodexRpcNotification[] = []; + #waiters: Array<(value: IteratorResult) => void> = []; + #closed = false; + + push(value: CodexRpcNotification): void { + const waiter = this.#waiters.shift(); + if (waiter) waiter({ value, done: false }); + else this.#values.push(value); + } + + close(): void { + this.#closed = true; + for (const waiter of this.#waiters.splice(0)) waiter({ value: undefined, done: true }); + } + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: async () => { + const value = this.#values.shift(); + if (value) return { value, done: false }; + if (this.#closed) return { value: undefined, done: true }; + return new Promise((resolve) => this.#waiters.push(resolve)); + }, + }; + } +} + +interface FakeProviderState { + threadId: string; + providerSessionId: string; + nextTurn: number; + lastToolResult: string; + transports: FakePhase7CodexTransport[]; +} + +class FakePhase7CodexTransport implements CodexAppServerTransport { + readonly notificationsQueue = new AsyncNotifications(); + readonly requests: Array<{ method: string; params: Record }> = []; + #handler: CodexServerRequestHandler = async () => ({}); + #closed = false; + #activeTurnId: string | null = null; + + constructor( + readonly state: FakeProviderState, + readonly onClose: () => void = () => undefined, + ) { + state.transports.push(this); + } + + async request(method: string, params: Record): Promise> { + this.requests.push({ method, params: structuredClone(params) }); + if (method === "initialize") return { user: { sessionId: this.state.providerSessionId } }; + if (method === "thread/start") { + expect(Array.isArray(params.dynamicTools)).toBe(true); + return { thread: { id: this.state.threadId, sessionId: this.state.providerSessionId } }; + } + if (method === "thread/read") { + return { thread: { id: this.state.threadId, sessionId: this.state.providerSessionId, turns: [] } }; + } + if (method === "thread/resume") { + return { thread: { id: this.state.threadId, sessionId: this.state.providerSessionId } }; + } + if (method === "turn/start") { + const turnId = `turn-${++this.state.nextTurn}`; + this.#activeTurnId = turnId; + const input = Array.isArray(params.input) ? params.input[0] as Record : {}; + const message = String(input.text ?? ""); + void this.#runTurn(turnId, message); + return { turn: { id: turnId, status: "inProgress" } }; + } + if (method === "turn/interrupt") { + const turnId = String(params.turnId); + this.notificationsQueue.push({ + method: "turn/completed", + params: { threadId: this.state.threadId, turn: { id: turnId, status: "interrupted" } }, + }); + this.#activeTurnId = null; + return {}; + } + throw new Error(`unsupported fake Codex method ${method}`); + } + + notify(): void {} + + notifications(): AsyncIterable { + return this.notificationsQueue; + } + + setServerRequestHandler(handler: CodexServerRequestHandler): void { + this.#handler = handler; + } + + async close(): Promise { + this.#closed = true; + this.notificationsQueue.close(); + this.onClose(); + } + + processInfo() { + return { + pid: 4100 + this.state.transports.indexOf(this), + processGroupId: 4100 + this.state.transports.indexOf(this), + exited: this.#closed, + exitCode: this.#closed ? 0 : null, + signal: null, + }; + } + + async #runTurn(turnId: string, message: string): Promise { + await Promise.resolve(); + this.notificationsQueue.push({ + method: "turn/started", + params: { threadId: this.state.threadId, turn: { id: turnId, status: "inProgress" } }, + }); + await new Promise((resolve) => setTimeout(resolve, 0)); + let assistantText: string; + if (message.includes("progress")) { + const result = await this.#toolCall({ + id: `request-${turnId}`, + method: "item/tool/call", + params: { + threadId: this.state.threadId, + turnId, + callId: `call-${turnId}`, + tool: "report_progress", + arguments: { + idempotencyKey: `progress-${turnId}`, + body: "Progress persisted through the live Codex tool loop.", + }, + }, + }); + this.state.lastToolResult = String((result.contentItems as Array>)[0]?.text); + assistantText = `The typed result changed my response: ${this.state.lastToolResult}`; + } else if (message.includes("semantic-interaction-result")) { + assistantText = `Resumed the same provider thread with ${message}`; + } else if (message.includes("question")) { + const result = await this.#toolCall({ + id: `request-${turnId}`, + method: "item/tool/call", + params: { + threadId: this.state.threadId, + turnId, + callId: `call-${turnId}`, + tool: "request_human_input", + arguments: { + idempotencyKey: `question-${turnId}`, + interactionKind: "questions", + title: "Choose the mock path", + prompt: "Which path should the mock agent take?", + payload: { fields: [{ id: "path", label: "Path" }] }, + continuationPolicy: "wake_assignee", + }, + }, + }); + this.state.lastToolResult = String((result.contentItems as Array>)[0]?.text); + assistantText = "I created a durable question and will use its typed result when resumed."; + } else if (message.includes("long")) { + return; + } else { + assistantText = `Same thread remembers ${this.state.lastToolResult}`; + } + this.notificationsQueue.push({ + method: "item/completed", + params: { + threadId: this.state.threadId, + turnId, + item: { id: `message-${turnId}`, type: "agentMessage", text: assistantText }, + }, + }); + this.notificationsQueue.push({ + method: "turn/completed", + params: { threadId: this.state.threadId, turn: { id: turnId, status: "completed" } }, + }); + this.#activeTurnId = null; + } + + #toolCall(request: CodexRpcServerRequest): Promise> { + return this.#handler(request); + } +} + +function fakeTransportFactory(state: FakeProviderState): Phase7LiveTransportFactory { + return (options) => { + const evidence = { + runnerPid: 4100 + state.transports.length, + runnerProcessGroupId: 4100 + state.transports.length, + codexPid: 5100 + state.transports.length, + runnerExited: false, + runnerExitCode: null, + runnerSignal: null, + childEnvironmentKeys: ["CODEX_HOME", "HOME", "PATH"], + diagnostics: ["paperclip-runnerd: phase7 codex proxy started"], + }; + const transport = new FakePhase7CodexTransport(state, () => { + evidence.runnerExited = true; + options.onEvidence?.(evidence); + }); + options.onEvidence?.(evidence); + return { transport, evidence: () => ({ ...evidence, runnerExited: transport.processInfo().exited }) }; + }; +} + +function providerState(): FakeProviderState { + return { + threadId: "codex-thread-phase7", + providerSessionId: "codex-provider-phase7", + nextTurn: 0, + lastToolResult: "nothing yet", + transports: [], + }; +} + +describe("Phase 7 live runnerd and Codex session", () => { + it("returns typed mock results to the same multi-turn Codex thread without Paperclip network calls", async () => { + const state = providerState(); + const fetchSpy = vi.spyOn(globalThis, "fetch"); + const service = new Phase7LiveSessionService({ transportFactory: fakeTransportFactory(state) }); + const session = await service.create({ + runId: "run-live-tool-loop", + sessionId: "session-live-tool-loop", + }); + + const first = await session.sendMessage("Please report progress on this mock issue."); + const second = await session.sendMessage("What did the prior tool result say?"); + + expect(first.assistantText).toContain("stateRevision"); + expect(second.assistantText).toContain("stateRevision"); + expect(first.snapshot.providerThreadId).toBe(second.snapshot.providerThreadId); + expect(session.mockState().comments.at(-1)?.body).toBe( + "Progress persisted through the live Codex tool loop.", + ); + expect(first.snapshot.evidence.some((entry) => entry.kind === "tool_call")).toBe(true); + expect(first.snapshot.evidence.some((entry) => entry.kind === "tool_result")).toBe(true); + expect(first.snapshot.networkEvidence).toEqual({ + realPaperclipRequests: 0, + childPaperclipEnvironmentKeys: [], + }); + expect(fetchSpy).not.toHaveBeenCalled(); + fetchSpy.mockRestore(); + await service.shutdown(session.id); + }); + + it("restores transcript, mock state, and pending interactions before resuming the same Codex thread", async () => { + const state = providerState(); + const store = new InMemoryPhase7LiveSessionStore(); + const firstService = new Phase7LiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + const first = await firstService.create({ + runId: "run-live-resume", + sessionId: "session-live-resume", + capabilities: ["control_plane:interactions"], + }); + await first.sendMessage("Ask one structured question."); + const pending = first.pendingInteractions()[0]!; + + const resumedService = new Phase7LiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + const resumed = await resumedService.restore(first.id); + expect(resumed.snapshot().providerThreadId).toBe("codex-thread-phase7"); + expect(resumed.pendingInteractions()).toHaveLength(1); + expect(resumed.snapshot().transcript).toEqual(first.snapshot().transcript); + + const continuation = await resumed.resolveInteraction({ + interactionId: pending.id, + outcome: "answered", + result: { path: "safe" }, + }); + expect(continuation.assistantText).toContain("semantic-interaction-result"); + expect(resumed.pendingInteractions()).toHaveLength(0); + expect(resumed.mockState().interactions[0]?.result).toEqual({ path: "safe" }); + expect(state.transports[1]?.requests.map((entry) => entry.method)).toContain("thread/resume"); + await resumedService.shutdown(resumed.id); + await firstService.shutdown(first.id); + }); + + it("stops active work and reset rotates authority while restoring clean mock state", async () => { + const state = providerState(); + const store = new InMemoryPhase7LiveSessionStore(); + const service = new Phase7LiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + const session = await service.create({ + runId: "run-live-reset", + sessionId: "session-live-reset", + }); + const longTurn = session.sendMessage("Start a long turn."); + await new Promise((resolve) => setTimeout(resolve, 1)); + await session.interrupt("bounded test interrupt"); + await expect(longTurn).resolves.toMatchObject({ status: "interrupted" }); + await session.sendMessage("Please report progress before reset."); + + const replacement = await service.reset(session.id); + expect(replacement.id).not.toBe(session.id); + expect(replacement.snapshot().authority.runId).not.toBe("run-live-reset"); + expect(replacement.mockState().comments).toHaveLength(0); + expect(replacement.snapshot().authority.active).toBe(true); + expect(session.snapshot().authority.active).toBe(false); + expect((await store.load(session.id))).toBeNull(); + expect(state.transports[0]?.processInfo().exited).toBe(true); + const stopped = await service.stop(replacement.id, "bounded test stop"); + expect(stopped.authority.active).toBe(false); + expect(stopped.process?.runnerExited).toBe(true); + expect((await store.load(replacement.id))?.status).toBe("closed"); + }); +}); diff --git a/packages/paperclip-runner/src/phase7/live-session.ts b/packages/paperclip-runner/src/phase7/live-session.ts new file mode 100644 index 00000000000..d0ac60a1b09 --- /dev/null +++ b/packages/paperclip-runner/src/phase7/live-session.ts @@ -0,0 +1,873 @@ +import { randomUUID } from "node:crypto"; + +import type { + CodexAppServerTransport, + CodexRpcNotification, + CodexRpcServerRequest, +} from "../drivers/codex/app-server-transport.js"; +import { redactCodexDiagnostic } from "../drivers/codex/app-server-transport.js"; +import { createSkilllessCodexThreadConfig } from "../drivers/codex/codex-app-server-driver.js"; +import type { + Phase7FixtureInteraction, + Phase7FixtureSeed, + Phase7FixtureState, + Phase7InteractionKind, + Phase7JsonValue, + Phase7OpenFixtureRunInput, +} from "../mock-core/phase7-control-plane-types.js"; +import { Phase7MockControlPlaneAdapter } from "../mock-core/phase7-mock-control-plane-adapter.js"; +import { Phase7SemanticDispatcher } from "../semantic-tools/dispatcher.js"; +import type { + Phase7SemanticAuthorizationRecord, + Phase7SemanticScenarioPolicy, + Phase7SemanticToolDefinition, + Phase7SemanticToolResult, +} from "../semantic-tools/types.js"; +import { + createPhase7RunnerdCodexTransport, + type Phase7RunnerdCodexTransport, + type Phase7RunnerdCodexTransportOptions, + type Phase7RunnerdProcessEvidence, +} from "./runnerd-codex-transport.js"; + +const LIVE_SESSION_SCHEMA = "paperclip.phase7.live-session.v1" as const; +const LIVE_BASE_INSTRUCTIONS = [ + "You are operating one mock Paperclip issue through typed semantic tools.", + "Use only the tools exposed in this thread; never call a Paperclip REST API.", + "Treat every tool result as authoritative mock state and use it in your next response.", + "The user conversation and provider are real, but all Paperclip records are mock records.", + "Do not discover skills, credentials, endpoints, or hidden control-plane capabilities.", +].join(" "); +const CODEX_PERMISSION_PROFILE = "paperclip-runner-workspace-only"; + +export type Phase7LiveSessionStatus = + | "starting" + | "idle" + | "running" + | "stopping" + | "closed" + | "failed"; + +export interface Phase7LiveTranscriptEntry { + id: string; + role: "user" | "assistant" | "system"; + text: string; + turnId: string | null; + at: string; +} + +export interface Phase7LiveEvidenceEntry { + id: string; + kind: + | "session" + | "provider_event" + | "tool_exposure" + | "tool_call" + | "tool_result" + | "interaction" + | "process" + | "diagnostic" + | "cleanup"; + at: string; + turnId: string | null; + data: Record; +} + +export interface Phase7LiveAuthoritySnapshot { + active: boolean; + runId: string; + companyId: string; + actorId: string; + taskId: string; + sessionId: string; + scenarioId: string; + capabilities: string[]; + explicitClaims: string[]; +} + +export interface Phase7LiveSessionConfigSnapshot { + workingDirectory: string; + scenario: Phase7SemanticScenarioPolicy; + capabilities: string[]; + explicitClaims: string[]; + seedState: string; + turnTimeoutMs: number; +} + +export interface Phase7LiveSessionSnapshot { + schema: typeof LIVE_SESSION_SCHEMA; + revision: number; + sessionId: string; + providerThreadId: string; + providerSessionId: string | null; + status: Phase7LiveSessionStatus; + activeTurnId: string | null; + createdAt: string; + updatedAt: string; + authority: Phase7LiveAuthoritySnapshot; + config: Phase7LiveSessionConfigSnapshot; + mockState: string; + transcript: Phase7LiveTranscriptEntry[]; + evidence: Phase7LiveEvidenceEntry[]; + authorizationRecords: Phase7SemanticAuthorizationRecord[]; + process: Phase7RunnerdProcessEvidence | null; + networkEvidence: { + realPaperclipRequests: 0; + childPaperclipEnvironmentKeys: string[]; + }; +} + +export interface CreatePhase7LiveSessionInput { + seed?: Phase7FixtureSeed | Phase7FixtureState; + workingDirectory?: string; + scenario?: Phase7SemanticScenarioPolicy; + capabilities?: string[]; + explicitClaims?: string[]; + companyId?: string; + actorId?: string; + taskId?: string; + runId?: string; + sessionId?: string; + turnTimeoutMs?: number; +} + +export interface Phase7LiveTurnResult { + turnId: string; + status: "completed" | "failed" | "interrupted" | "cancelled"; + assistantText: string; + snapshot: Phase7LiveSessionSnapshot; +} + +export interface Phase7InteractionResolution { + interactionId: string; + outcome: "answered" | "accepted" | "rejected" | "expired"; + result: Phase7JsonValue; +} + +export interface Phase7LiveSessionStore { + load(sessionId: string): Promise; + save(snapshot: Phase7LiveSessionSnapshot): Promise; + delete(sessionId: string): Promise; +} + +export class InMemoryPhase7LiveSessionStore implements Phase7LiveSessionStore { + readonly #snapshots = new Map(); + + async load(sessionId: string): Promise { + const snapshot = this.#snapshots.get(sessionId); + return snapshot === undefined ? null : structuredClone(snapshot); + } + + async save(snapshot: Phase7LiveSessionSnapshot): Promise { + this.#snapshots.set(snapshot.sessionId, structuredClone(snapshot)); + } + + async delete(sessionId: string): Promise { + this.#snapshots.delete(sessionId); + } +} + +export type Phase7LiveTransportFactory = ( + options: Phase7RunnerdCodexTransportOptions, +) => Phase7RunnerdCodexTransport; + +export interface Phase7LiveSessionServiceOptions { + store?: Phase7LiveSessionStore; + transportFactory?: Phase7LiveTransportFactory; + transportOptions?: Phase7RunnerdCodexTransportOptions; + now?: () => Date; +} + +interface TurnWaiter { + resolve: (result: Omit) => void; + reject: (error: Error) => void; + timer: ReturnType; + assistantText: string; +} + +function record(value: unknown): Record { + return typeof value === "object" && value !== null && !Array.isArray(value) + ? value as Record + : {}; +} + +function text(value: unknown, fallback = ""): string { + return typeof value === "string" ? value : fallback; +} + +function jsonValue(value: unknown): Phase7JsonValue { + return JSON.parse(JSON.stringify(value ?? null)) as Phase7JsonValue; +} + +function dynamicToolSpec(definition: Phase7SemanticToolDefinition): Record { + return { + name: definition.name, + description: definition.description, + inputSchema: definition.inputSchema, + }; +} + +function userInput(message: string): Record { + return { type: "text", text: message, text_elements: [] }; +} + +function terminalStatus(value: unknown): Phase7LiveTurnResult["status"] { + const status = text(value, "completed"); + return status === "failed" || status === "interrupted" || status === "cancelled" + ? status + : "completed"; +} + +function pendingInteractions(port: Phase7MockControlPlaneAdapter): Phase7FixtureInteraction[] { + return port.snapshot().interactions.filter((interaction) => interaction.status === "pending"); +} + +export class Phase7LiveSessionService { + readonly #store: Phase7LiveSessionStore; + readonly #transportFactory: Phase7LiveTransportFactory; + readonly #transportOptions: Phase7RunnerdCodexTransportOptions; + readonly #now: () => Date; + readonly #sessions = new Map(); + + constructor(options: Phase7LiveSessionServiceOptions = {}) { + this.#store = options.store ?? new InMemoryPhase7LiveSessionStore(); + this.#transportFactory = options.transportFactory ?? createPhase7RunnerdCodexTransport; + this.#transportOptions = options.transportOptions ?? {}; + this.#now = options.now ?? (() => new Date()); + } + + async create(input: CreatePhase7LiveSessionInput = {}): Promise { + const port = new Phase7MockControlPlaneAdapter(input.seed); + const seedState = port.serialize(); + await port.start(); + const state = port.snapshot(); + const sessionId = input.sessionId ?? randomUUID(); + const runId = input.runId ?? randomUUID(); + const capabilities = [...(input.capabilities ?? [])]; + const scenario = input.scenario ?? { id: "phase7-live-default" }; + const authority: Phase7LiveAuthoritySnapshot = { + active: true, + runId, + companyId: input.companyId ?? state.company.id, + actorId: input.actorId ?? state.actors[0]!.id, + taskId: input.taskId ?? state.tasks[0]!.id, + sessionId, + scenarioId: scenario.id, + capabilities, + explicitClaims: [...(input.explicitClaims ?? capabilities)], + }; + await port.openFixtureRun({ + identity: { + runId, + sessionId, + companyId: authority.companyId, + issueId: authority.taskId, + agentId: authority.actorId, + }, + backendKind: "runner", + sourceInstanceId: "phase7-live-runnerd", + capabilities, + }); + const session = await Phase7LiveSession.open({ + port, + authority, + config: { + workingDirectory: input.workingDirectory ?? process.cwd(), + scenario, + capabilities, + explicitClaims: authority.explicitClaims, + seedState, + turnTimeoutMs: input.turnTimeoutMs ?? 120_000, + }, + store: this.#store, + transportFactory: this.#transportFactory, + transportOptions: this.#transportOptions, + now: this.#now, + }); + this.#sessions.set(session.id, session); + return session; + } + + async restore(sessionId: string): Promise { + const live = this.#sessions.get(sessionId); + if (live !== undefined) return live; + const snapshot = await this.#store.load(sessionId); + if (snapshot === null) throw new Error(`phase7 live session ${sessionId} was not found`); + const session = await Phase7LiveSession.restore({ + snapshot, + store: this.#store, + transportFactory: this.#transportFactory, + transportOptions: this.#transportOptions, + now: this.#now, + }); + this.#sessions.set(session.id, session); + return session; + } + + async reconnect(sessionId: string): Promise { + const session = await this.restore(sessionId); + await session.reconnect(); + return session.snapshot(); + } + + async reset(sessionId: string): Promise { + const session = await this.restore(sessionId); + const config = session.snapshot().config; + await session.shutdown("reset"); + this.#sessions.delete(sessionId); + await this.#store.delete(sessionId); + const seedPort = Phase7MockControlPlaneAdapter.restore(config.seedState); + return this.create({ + seed: seedPort.snapshot(), + workingDirectory: config.workingDirectory, + scenario: config.scenario, + capabilities: config.capabilities, + explicitClaims: config.explicitClaims, + turnTimeoutMs: config.turnTimeoutMs, + }); + } + + async stop(sessionId: string, reason = "operator stopped session"): Promise { + const session = await this.restore(sessionId); + await session.shutdown(reason); + this.#sessions.delete(sessionId); + return session.snapshot(); + } + + async shutdown(sessionId: string, reason = "service shutdown"): Promise { + const session = this.#sessions.get(sessionId); + if (session === undefined) return; + await session.shutdown(reason); + this.#sessions.delete(sessionId); + } +} + +interface OpenSessionOptions { + port: Phase7MockControlPlaneAdapter; + authority: Phase7LiveAuthoritySnapshot; + config: Phase7LiveSessionConfigSnapshot; + store: Phase7LiveSessionStore; + transportFactory: Phase7LiveTransportFactory; + transportOptions: Phase7RunnerdCodexTransportOptions; + now: () => Date; +} + +interface RestoreSessionOptions extends Omit { + snapshot: Phase7LiveSessionSnapshot; +} + +export class Phase7LiveSession { + readonly #port: Phase7MockControlPlaneAdapter; + readonly #dispatcher: Phase7SemanticDispatcher; + readonly #store: Phase7LiveSessionStore; + readonly #transportFactory: Phase7LiveTransportFactory; + readonly #transportOptions: Phase7RunnerdCodexTransportOptions; + readonly #now: () => Date; + readonly #authority: Phase7LiveAuthoritySnapshot; + readonly #config: Phase7LiveSessionConfigSnapshot; + readonly #createdAt: string; + readonly #transcript: Phase7LiveTranscriptEntry[]; + readonly #evidence: Phase7LiveEvidenceEntry[]; + readonly #priorAuthorizationRecords: Phase7SemanticAuthorizationRecord[]; + #transport: CodexAppServerTransport | null = null; + #processEvidence: Phase7RunnerdProcessEvidence | null = null; + #providerThreadId = ""; + #providerSessionId: string | null = null; + #status: Phase7LiveSessionStatus = "starting"; + #activeTurnId: string | null = null; + #revision = 0; + #updatedAt: string; + #entryCounter = 0; + #turnWaiter: TurnWaiter | null = null; + #pump: Promise | null = null; + #persistChain: Promise = Promise.resolve(); + + private constructor(options: OpenSessionOptions & { snapshot?: Phase7LiveSessionSnapshot }) { + this.#port = options.port; + this.#authority = structuredClone(options.authority); + this.#config = structuredClone(options.config); + this.#store = options.store; + this.#transportFactory = options.transportFactory; + this.#transportOptions = options.transportOptions; + this.#now = options.now; + this.#dispatcher = new Phase7SemanticDispatcher(this.#port, { + scenario: this.#config.scenario, + explicitClaims: this.#config.explicitClaims, + }); + this.#createdAt = options.snapshot?.createdAt ?? this.#now().toISOString(); + this.#updatedAt = options.snapshot?.updatedAt ?? this.#createdAt; + this.#transcript = structuredClone(options.snapshot?.transcript ?? []); + this.#evidence = structuredClone(options.snapshot?.evidence ?? []); + this.#priorAuthorizationRecords = structuredClone(options.snapshot?.authorizationRecords ?? []); + this.#providerThreadId = options.snapshot?.providerThreadId ?? ""; + this.#providerSessionId = options.snapshot?.providerSessionId ?? null; + this.#activeTurnId = options.snapshot?.activeTurnId ?? null; + this.#revision = options.snapshot?.revision ?? 0; + this.#entryCounter = this.#transcript.length + this.#evidence.length; + } + + static async open(options: OpenSessionOptions): Promise { + const session = new Phase7LiveSession(options); + await session.#connect(false); + return session; + } + + static async restore(options: RestoreSessionOptions): Promise { + if (options.snapshot.schema !== LIVE_SESSION_SCHEMA) { + throw new Error("unsupported Phase 7 live session snapshot"); + } + if (!options.snapshot.authority.active) { + throw new Error("phase7 session authority was cleared and cannot be resumed"); + } + const port = Phase7MockControlPlaneAdapter.restore(options.snapshot.mockState); + if (port.snapshot().lifecycle !== "running") await port.start(); + const session = new Phase7LiveSession({ + ...options, + port, + authority: options.snapshot.authority, + config: options.snapshot.config, + snapshot: options.snapshot, + }); + await session.#connect(true); + return session; + } + + get id(): string { + return this.#authority.sessionId; + } + + mockState() { + return this.#port.snapshot(); + } + + pendingInteractions(): Phase7FixtureInteraction[] { + return structuredClone(pendingInteractions(this.#port)); + } + + snapshot(): Phase7LiveSessionSnapshot { + const records = [ + ...this.#priorAuthorizationRecords, + ...this.#dispatcher.authorizationRecords(), + ]; + const childKeys = this.#processEvidence?.childEnvironmentKeys ?? []; + return { + schema: LIVE_SESSION_SCHEMA, + revision: this.#revision, + sessionId: this.id, + providerThreadId: this.#providerThreadId, + providerSessionId: this.#providerSessionId, + status: this.#status, + activeTurnId: this.#activeTurnId, + createdAt: this.#createdAt, + updatedAt: this.#updatedAt, + authority: structuredClone(this.#authority), + config: structuredClone(this.#config), + mockState: this.#port.serialize(), + transcript: structuredClone(this.#transcript), + evidence: structuredClone(this.#evidence), + authorizationRecords: structuredClone(records), + process: this.#processEvidence === null ? null : structuredClone(this.#processEvidence), + networkEvidence: { + realPaperclipRequests: 0, + childPaperclipEnvironmentKeys: childKeys.filter((key) => key.startsWith("PAPERCLIP_")), + }, + }; + } + + async sendMessage(message: string): Promise { + const value = message.trim(); + if (value.length === 0) throw new Error("Phase 7 live messages cannot be empty"); + if (this.#transport === null || this.#status === "closed" || this.#status === "failed") { + throw new Error("Phase 7 live session is not connected"); + } + if (this.#activeTurnId !== null || this.#turnWaiter !== null) { + throw new Error("Phase 7 live session already has an active turn"); + } + this.#status = "running"; + this.#appendTranscript("user", value, null); + let response: Record; + const terminal = new Promise>((resolve, reject) => { + const timer = setTimeout(() => { + this.#turnWaiter = null; + reject(new Error(`Phase 7 Codex turn timed out after ${this.#config.turnTimeoutMs}ms`)); + }, this.#config.turnTimeoutMs); + this.#turnWaiter = { resolve, reject, timer, assistantText: "" }; + }); + try { + response = await this.#transport.request("turn/start", { + threadId: this.#providerThreadId, + cwd: this.#config.workingDirectory, + permissions: CODEX_PERMISSION_PROFILE, + runtimeWorkspaceRoots: [this.#config.workingDirectory], + input: [userInput(value)], + }); + } catch (error) { + this.#rejectTurn(error); + await terminal.catch(() => undefined); + throw error; + } + const turnId = text(record(response.turn).id); + if (turnId.length === 0) { + this.#rejectTurn(new Error("Codex turn response omitted turn.id")); + throw new Error("Codex turn response omitted turn.id"); + } + if (this.#activeTurnId !== null && this.#activeTurnId !== turnId) { + this.#rejectTurn(new Error("Codex turn identity changed during start")); + throw new Error("Codex turn identity changed during start"); + } + if (this.#turnWaiter !== null) this.#activeTurnId = turnId; + const lastUser = this.#transcript.at(-1); + if (lastUser?.role === "user" && lastUser.turnId === null) lastUser.turnId = turnId; + await this.#persist(); + const result = await terminal; + await this.#persist(); + return { ...result, snapshot: this.snapshot() }; + } + + async interrupt(reason = "operator interrupted turn"): Promise { + if (this.#transport === null || this.#activeTurnId === null) return this.snapshot(); + this.#status = "stopping"; + const turnId = this.#activeTurnId; + this.#appendEvidence("session", turnId, { action: "stop_requested", reason }); + await this.#transport.request("turn/interrupt", { + threadId: this.#providerThreadId, + turnId, + }); + await this.#persist(); + return this.snapshot(); + } + + async resolveInteraction(input: Phase7InteractionResolution): Promise { + const interaction = pendingInteractions(this.#port).find( + (candidate) => candidate.id === input.interactionId, + ); + if (interaction === undefined) throw new Error("interaction is not pending in this session"); + const outcome = await this.#port.tryApplyCommand({ + runId: this.#authority.runId, + idempotencyKey: `phase7-live-interaction:${input.interactionId}:${input.outcome}`, + command: { + kind: "resolve_human_input", + interactionId: input.interactionId, + outcome: input.outcome, + result: input.result, + }, + }); + if (!outcome.ok) throw new Error(`${outcome.error.code}: ${outcome.error.message}`); + this.#appendEvidence("interaction", null, { + interactionId: input.interactionId, + interactionKind: interaction.kind, + outcome: input.outcome, + result: input.result, + stateRevision: outcome.result.stateRevision, + }); + await this.#persist(); + return this.sendMessage(JSON.stringify({ + schema: "paperclip.semantic-interaction-result.v1", + interactionId: input.interactionId, + kind: interaction.kind, + outcome: input.outcome, + result: input.result, + instruction: "Continue using this typed interaction result in the same mock issue session.", + })); + } + + async reconnect(): Promise { + if (this.#activeTurnId !== null) { + throw new Error("stop the active turn before reconnecting the Codex transport"); + } + await this.#disconnect("reconnect"); + await this.#connect(true); + } + + async shutdown(reason: string): Promise { + if (this.#activeTurnId !== null) { + try { + await this.interrupt(reason); + } catch (error) { + this.#appendEvidence("diagnostic", this.#activeTurnId, { + message: redactCodexDiagnostic(String(error)), + }); + } + } + await this.#disconnect(reason); + await this.#port.stop(); + this.#authority.active = false; + this.#status = "closed"; + this.#appendEvidence("cleanup", null, { + reason, + authorityCleared: true, + mockStopped: true, + processExited: this.#processEvidence?.runnerExited ?? true, + }); + await this.#persist(); + } + + async #connect(resume: boolean): Promise { + this.#status = "starting"; + const transportBundle = this.#transportFactory({ + ...this.#transportOptions, + onDiagnostic: (message) => { + this.#transportOptions.onDiagnostic?.(message); + this.#appendEvidence("diagnostic", this.#activeTurnId, { + message: redactCodexDiagnostic(message), + }); + }, + onEvidence: (evidence) => { + this.#processEvidence = structuredClone(evidence); + this.#transportOptions.onEvidence?.(evidence); + }, + }); + this.#transport = transportBundle.transport; + this.#processEvidence = structuredClone(transportBundle.evidence()); + this.#transport.setServerRequestHandler((request) => this.#handleServerRequest(request)); + const initialized = await this.#transport.request("initialize", { + clientInfo: { + name: "paperclip-runner", + title: "Paperclip Runner Phase 7", + version: "phase7-v1", + }, + capabilities: { experimentalApi: true, requestAttestation: false }, + }); + this.#transport.notify("initialized"); + const tools = this.#dispatcher.listTools(this.#authority.runId); + this.#appendEvidence("tool_exposure", null, { + operationIds: tools.map((tool) => tool.name), + scenarioId: this.#config.scenario.id, + }); + if (resume) { + const read = await this.#transport.request("thread/read", { + threadId: this.#providerThreadId, + includeTurns: true, + }); + const thread = record(read.thread); + if (text(thread.id) !== this.#providerThreadId) { + throw new Error("Codex reconnect read a different provider thread"); + } + const resumed = await this.#transport.request("thread/resume", { + threadId: this.#providerThreadId, + cwd: this.#config.workingDirectory, + config: createSkilllessCodexThreadConfig(this.#config.workingDirectory), + permissions: CODEX_PERMISSION_PROFILE, + runtimeWorkspaceRoots: [this.#config.workingDirectory], + baseInstructions: LIVE_BASE_INSTRUCTIONS, + persistExtendedHistory: true, + }); + const resumedThread = record(resumed.thread); + if (text(resumedThread.id) !== this.#providerThreadId) { + throw new Error("Codex reconnect resumed a different provider thread"); + } + this.#providerSessionId = text(resumedThread.sessionId) || this.#providerSessionId; + this.#activeTurnId = null; + } else { + const opened = await this.#transport.request("thread/start", { + cwd: this.#config.workingDirectory, + config: createSkilllessCodexThreadConfig(this.#config.workingDirectory), + permissions: CODEX_PERMISSION_PROFILE, + runtimeWorkspaceRoots: [this.#config.workingDirectory], + approvalPolicy: "never", + baseInstructions: LIVE_BASE_INSTRUCTIONS, + dynamicTools: tools.map(dynamicToolSpec), + experimentalRawEvents: false, + persistExtendedHistory: true, + }); + const thread = record(opened.thread); + this.#providerThreadId = text(thread.id); + if (this.#providerThreadId.length === 0) throw new Error("Codex thread response omitted thread.id"); + this.#providerSessionId = text(thread.sessionId) || text(record(initialized.user).sessionId) || null; + } + this.#status = "idle"; + this.#appendEvidence("session", null, { + action: resume ? "resumed" : "started", + sessionId: this.id, + providerThreadId: this.#providerThreadId, + providerSessionId: this.#providerSessionId, + mode: "live_codex", + runner: "paperclip-runnerd", + controlPlane: "mock", + }); + this.#pump = this.#pumpNotifications(); + await this.#persist(); + } + + async #disconnect(reason: string): Promise { + const transport = this.#transport; + this.#transport = null; + if (transport !== null) await transport.close(); + if (this.#pump !== null) await this.#pump.catch(() => undefined); + this.#pump = null; + this.#appendEvidence("process", null, { + action: "transport_closed", + reason, + runnerExited: this.#processEvidence?.runnerExited ?? true, + runnerPid: this.#processEvidence?.runnerPid ?? null, + codexPid: this.#processEvidence?.codexPid ?? null, + }); + } + + async #handleServerRequest(request: CodexRpcServerRequest): Promise> { + if (request.method !== "item/tool/call") { + return { + success: false, + contentItems: [{ type: "inputText", text: "Unsupported Codex client request." }], + }; + } + const operationId = text(request.params.tool); + const threadId = text(request.params.threadId); + const turnId = text(request.params.turnId); + const callId = text(request.params.callId, String(request.id)); + if ( + threadId !== this.#providerThreadId || + turnId.length === 0 || + turnId !== this.#activeTurnId || + callId.length === 0 + ) { + return { + success: false, + contentItems: [{ type: "inputText", text: "Tool call was outside the active Phase 7 thread and turn." }], + }; + } + const beforeRevision = this.#port.snapshot().revision; + this.#appendEvidence("tool_call", turnId, { + callId, + operationId, + input: jsonValue(request.params.arguments), + beforeRevision, + }); + const result = await this.#dispatcher.dispatch({ + runId: this.#authority.runId, + callId, + operationId, + input: request.params.arguments, + }); + this.#appendEvidence("tool_result", turnId, { + callId, + operationId, + result: jsonValue(result), + beforeRevision, + afterRevision: result.stateRevision, + }); + await this.#persist(); + return this.#codexToolResponse(result); + } + + #codexToolResponse(result: Phase7SemanticToolResult): Record { + return { + success: result.ok, + contentItems: [{ + type: "inputText", + text: JSON.stringify(result), + }], + }; + } + + async #pumpNotifications(): Promise { + const transport = this.#transport; + if (transport === null) return; + try { + for await (const notification of transport.notifications()) { + this.#handleNotification(notification); + } + } catch (error) { + if (this.#transport !== transport) return; + this.#status = "failed"; + this.#appendEvidence("diagnostic", this.#activeTurnId, { + message: redactCodexDiagnostic(String(error)), + }); + this.#rejectTurn(error); + await this.#persist(); + } + } + + #handleNotification(notification: CodexRpcNotification): void { + const params = notification.params; + const turn = record(params.turn); + const item = record(params.item); + const turnId = text(params.turnId, text(turn.id)); + this.#appendEvidence("provider_event", turnId || null, { + method: notification.method, + params: jsonValue(params), + }); + if (notification.method === "turn/started") { + if (turnId.length > 0) this.#activeTurnId = turnId; + } else if (notification.method === "item/agentMessage/delta") { + if (this.#turnWaiter !== null) this.#turnWaiter.assistantText += text(params.delta); + } else if (notification.method === "item/completed" && text(item.type) === "agentMessage") { + const assistantText = text(item.text); + if (assistantText.length > 0 && this.#turnWaiter !== null) { + this.#turnWaiter.assistantText = assistantText; + } + } else if (notification.method === "turn/completed") { + this.#completeTurn(turnId, terminalStatus(turn.status)); + } + void this.#persist(); + } + + #completeTurn(turnId: string, status: Phase7LiveTurnResult["status"]): void { + const waiter = this.#turnWaiter; + if (waiter === null) return; + clearTimeout(waiter.timer); + this.#turnWaiter = null; + const assistantText = waiter.assistantText.trim(); + if (assistantText.length > 0) this.#appendTranscript("assistant", assistantText, turnId); + this.#activeTurnId = null; + this.#status = "idle"; + waiter.resolve({ turnId, status, assistantText }); + } + + #rejectTurn(error: unknown): void { + const waiter = this.#turnWaiter; + if (waiter === null) return; + clearTimeout(waiter.timer); + this.#turnWaiter = null; + this.#activeTurnId = null; + this.#status = "failed"; + waiter.reject(error instanceof Error ? error : new Error(String(error))); + } + + #appendTranscript(role: Phase7LiveTranscriptEntry["role"], message: string, turnId: string | null): void { + this.#entryCounter += 1; + this.#transcript.push({ + id: `transcript-${String(this.#entryCounter).padStart(6, "0")}`, + role, + text: message, + turnId, + at: this.#now().toISOString(), + }); + } + + #appendEvidence( + kind: Phase7LiveEvidenceEntry["kind"], + turnId: string | null, + data: Record, + ): void { + this.#entryCounter += 1; + this.#evidence.push({ + id: `evidence-${String(this.#entryCounter).padStart(6, "0")}`, + kind, + at: this.#now().toISOString(), + turnId, + data: jsonValue(data) as Record, + }); + if (this.#evidence.length > 2_000) this.#evidence.splice(0, this.#evidence.length - 2_000); + } + + #persist(): Promise { + this.#revision += 1; + this.#updatedAt = this.#now().toISOString(); + const snapshot = this.snapshot(); + this.#persistChain = this.#persistChain.then(() => this.#store.save(snapshot)); + return this.#persistChain; + } +} + +export function phase7InteractionKindFromMock( + kind: Phase7InteractionKind, +): "ask_user_questions" | "request_confirmation" | "request_checkbox_confirmation" | "request_item_verdicts" | "suggest_tasks" { + if (kind === "questions") return "ask_user_questions"; + if (kind === "confirmation") return "request_confirmation"; + if (kind === "checkbox") return "request_checkbox_confirmation"; + if (kind === "item_verdicts") return "request_item_verdicts"; + return "suggest_tasks"; +} + +export type { Phase7OpenFixtureRunInput }; diff --git a/packages/paperclip-runner/src/phase7/runnerd-codex-transport.ts b/packages/paperclip-runner/src/phase7/runnerd-codex-transport.ts new file mode 100644 index 00000000000..96751189af3 --- /dev/null +++ b/packages/paperclip-runner/src/phase7/runnerd-codex-transport.ts @@ -0,0 +1,106 @@ +import { resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { + ProcessCodexAppServerTransport, + createSanitizedCodexEnvironment, + type CodexAppServerTransport, + type CodexTransportProcessInfo, +} from "../drivers/codex/app-server-transport.js"; +import { createIsolatedCodexAppServerArgs } from "../drivers/codex/codex-app-server-driver.js"; + +const packageRoot = fileURLToPath(new URL("../..", import.meta.url)); +const executableSuffix = process.platform === "win32" ? ".exe" : ""; + +export interface Phase7RunnerdProcessEvidence { + runnerPid: number | null; + runnerProcessGroupId: number | null; + codexPid: number | null; + runnerExited: boolean; + runnerExitCode: number | null; + runnerSignal: NodeJS.Signals | null; + childEnvironmentKeys: string[]; + diagnostics: string[]; +} + +export interface Phase7RunnerdCodexTransportOptions { + runnerBinary?: string; + codexCommand?: string; + codexArgs?: string[]; + environment?: NodeJS.ProcessEnv; + closeGraceMs?: number; + onDiagnostic?: (message: string) => void; + onEvidence?: (evidence: Readonly) => void; +} + +export interface Phase7RunnerdCodexTransport { + transport: CodexAppServerTransport; + evidence(): Readonly; +} + +export function defaultPhase7RunnerdBinary(): string { + return resolve( + packageRoot, + `runner/target/debug/paperclip-runnerd${executableSuffix}`, + ); +} + +/** + * Starts the real package-local runnerd as process owner. runnerd then starts + * the real Codex app-server and transparently proxies its stdio JSON-RPC. + */ +export function createPhase7RunnerdCodexTransport( + options: Phase7RunnerdCodexTransportOptions = {}, +): Phase7RunnerdCodexTransport { + const environment = createSanitizedCodexEnvironment(options.environment); + const evidence: Phase7RunnerdProcessEvidence = { + runnerPid: null, + runnerProcessGroupId: null, + codexPid: null, + runnerExited: false, + runnerExitCode: null, + runnerSignal: null, + childEnvironmentKeys: Object.keys(environment).sort(), + diagnostics: [], + }; + const publish = () => options.onEvidence?.(structuredClone(evidence)); + const diagnostic = (message: string) => { + evidence.diagnostics.push(message); + if (evidence.diagnostics.length > 64) evidence.diagnostics.shift(); + const match = /phase7 codex proxy started runner_pid=(\d+) codex_pid=(\d+)/.exec(message); + if (match) { + evidence.runnerPid = Number(match[1]); + evidence.codexPid = Number(match[2]); + } + options.onDiagnostic?.(message); + publish(); + }; + const processUpdate = (info: CodexTransportProcessInfo) => { + evidence.runnerPid = info.pid; + evidence.runnerProcessGroupId = info.processGroupId; + evidence.runnerExited = info.exited; + evidence.runnerExitCode = info.exitCode; + evidence.runnerSignal = info.signal; + publish(); + }; + const codexArgs = options.codexArgs ?? createIsolatedCodexAppServerArgs(options.environment); + const runnerArgs = [ + "--codex-app-server-proxy", + "--codex-command", + options.codexCommand ?? "codex", + ...codexArgs.flatMap((argument) => ["--codex-arg", argument]), + ]; + const transport = new ProcessCodexAppServerTransport({ + command: options.runnerBinary ?? defaultPhase7RunnerdBinary(), + args: runnerArgs, + environment, + processGroup: true, + closeGraceMs: options.closeGraceMs, + onDiagnostic: diagnostic, + onProcess: processUpdate, + }); + return { + transport, + evidence: () => structuredClone(evidence), + }; +}