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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-10 07:25:53 +00:00
1 parent 175a8b6f8c
commit aa87487a2c
13 files changed
+1677 -10

No files matched your search

+1
View File
@@ -3,6 +3,7 @@ node_modules/
**/node_modules
**/node_modules/
dist/
**/target/
ui/storybook-static/
.env
*.tsbuildinfo
+7 -1
View File
@@ -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)
+1
View File
@@ -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)
@@ -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.
+2
View File
@@ -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",
@@ -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<Duration,
))
}
fn repeated_values(args: &[String], name: &str) -> Result<Vec<String>, 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<BootstrapTicket>,
@@ -82,6 +158,9 @@ fn run_phase3(
fn run(bootstrap_ticket: Option<BootstrapTicket>) -> Result<(), Phase2Error> {
let args = std::env::args().skip(1).collect::<Vec<_>>();
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);
}
@@ -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<void> {
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;
});
@@ -19,6 +19,15 @@ export interface CodexAppServerTransport {
notifications(): AsyncIterable<CodexRpcNotification>;
setServerRequestHandler(handler: CodexServerRequestHandler): void;
close(): Promise<void>;
processInfo?(): CodexTransportProcessInfo;
}
export interface CodexTransportProcessInfo {
pid: number | null;
processGroupId: number | null;
exited: boolean;
exitCode: number | null;
signal: NodeJS.Signals | null;
}
class BoundedAsyncQueue<T> implements AsyncIterable<T> {
@@ -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<void>;
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<void>((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<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"));
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<false>((resolve) => setTimeout(() => resolve(false), this.#closeGraceMs)),
]);
if (!exited && !this.#exited) {
this.#signal("SIGKILL");
await Promise.race([
this.#exitPromise,
new Promise<void>((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));
+1
View File
@@ -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";
@@ -0,0 +1,2 @@
export * from "./live-session.js";
export * from "./runnerd-codex-transport.js";
@@ -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<CodexRpcNotification> {
#values: CodexRpcNotification[] = [];
#waiters: Array<(value: IteratorResult<CodexRpcNotification>) => 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<CodexRpcNotification> {
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<string, unknown> }> = [];
#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<string, unknown>): Promise<Record<string, unknown>> {
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<string, unknown> : {};
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<CodexRpcNotification> {
return this.notificationsQueue;
}
setServerRequestHandler(handler: CodexServerRequestHandler): void {
this.#handler = handler;
}
async close(): Promise<void> {
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<void> {
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<Record<string, unknown>>)[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<Record<string, unknown>>)[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<Record<string, unknown>> {
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");
});
});
@@ -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<string, Phase7JsonValue>;
}
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<Phase7LiveSessionSnapshot | null>;
save(snapshot: Phase7LiveSessionSnapshot): Promise<void>;
delete(sessionId: string): Promise<void>;
}
export class InMemoryPhase7LiveSessionStore implements Phase7LiveSessionStore {
readonly #snapshots = new Map<string, Phase7LiveSessionSnapshot>();
async load(sessionId: string): Promise<Phase7LiveSessionSnapshot | null> {
const snapshot = this.#snapshots.get(sessionId);
return snapshot === undefined ? null : structuredClone(snapshot);
}
async save(snapshot: Phase7LiveSessionSnapshot): Promise<void> {
this.#snapshots.set(snapshot.sessionId, structuredClone(snapshot));
}
async delete(sessionId: string): Promise<void> {
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<Phase7LiveTurnResult, "snapshot">) => void;
reject: (error: Error) => void;
timer: ReturnType<typeof setTimeout>;
assistantText: string;
}
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
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<string, unknown> {
return {
name: definition.name,
description: definition.description,
inputSchema: definition.inputSchema,
};
}
function userInput(message: string): Record<string, unknown> {
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<string, Phase7LiveSession>();
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<Phase7LiveSession> {
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<Phase7LiveSession> {
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<Phase7LiveSessionSnapshot> {
const session = await this.restore(sessionId);
await session.reconnect();
return session.snapshot();
}
async reset(sessionId: string): Promise<Phase7LiveSession> {
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<Phase7LiveSessionSnapshot> {
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<void> {
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<OpenSessionOptions, "port" | "authority" | "config"> {
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<void> | null = null;
#persistChain: Promise<void> = 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<Phase7LiveSession> {
const session = new Phase7LiveSession(options);
await session.#connect(false);
return session;
}
static async restore(options: RestoreSessionOptions): Promise<Phase7LiveSession> {
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<Phase7LiveTurnResult> {
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<string, unknown>;
const terminal = new Promise<Omit<Phase7LiveTurnResult, "snapshot">>((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<Phase7LiveSessionSnapshot> {
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<Phase7LiveTurnResult> {
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<void> {
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<void> {
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<void> {
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<void> {
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<Record<string, unknown>> {
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<string, unknown> {
return {
success: result.ok,
contentItems: [{
type: "inputText",
text: JSON.stringify(result),
}],
};
}
async #pumpNotifications(): Promise<void> {
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<string, unknown>,
): 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<string, Phase7JsonValue>,
});
if (this.#evidence.length > 2_000) this.#evidence.splice(0, this.#evidence.length - 2_000);
}
#persist(): Promise<void> {
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 };
@@ -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<Phase7RunnerdProcessEvidence>) => void;
}
export interface Phase7RunnerdCodexTransport {
transport: CodexAppServerTransport;
evidence(): Readonly<Phase7RunnerdProcessEvidence>;
}
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),
};
}