From 3b27ea5c368759f0f324c50637f6a5c78cb2389a Mon Sep 17 00:00:00 2001 From: Dev Agent Date: Thu, 20 Aug 2026 02:21:05 +0000 Subject: [PATCH] fix(runner): resume governed live attempts durably Co-Authored-By: Paperclip --- .../docs/capability-live-runnerd-codex.md | 45 +- packages/paperclip-runner/package.json | 4 + .../runner-core/src/bin/paperclip-runnerd.rs | 46 +- .../src/live/durable-live-session-store.ts | 174 ++++++ packages/paperclip-runner/src/live/index.ts | 1 + .../src/live/live-session.test.ts | 267 ++++++++- .../paperclip-runner/src/live/live-session.ts | 521 +++++++++++++++++- .../src/live/runnerd-codex-transport.ts | 2 +- packages/paperclip-runner/src/testing.ts | 1 + .../fake-durable-codex-app-server.mjs | 163 ++++++ 10 files changed, 1189 insertions(+), 35 deletions(-) create mode 100644 packages/paperclip-runner/src/live/durable-live-session-store.ts create mode 100644 packages/paperclip-runner/test/fixtures/fake-durable-codex-app-server.mjs diff --git a/packages/paperclip-runner/docs/capability-live-runnerd-codex.md b/packages/paperclip-runner/docs/capability-live-runnerd-codex.md index 8a3df440873..3b93e8ec5de 100644 --- a/packages/paperclip-runner/docs/capability-live-runnerd-codex.md +++ b/packages/paperclip-runner/docs/capability-live-runnerd-codex.md @@ -26,18 +26,42 @@ the separate skillless, network-disabled workspace permission profile. ## Stable session API -Import the package root and construct `CapabilityLiveSessionService`: +Production workers import the live entrypoint and bind a durable store to the +attempt's immutable authority tuple: ```ts -import { CapabilityLiveSessionService } from "@paperclipai/paperclip-runner/testing"; +import { + CapabilityLiveSessionService, + DurableCapabilityLiveSessionStore, +} from "@paperclipai/paperclip-runner/live"; +const binding = { sessionId, runId, companyId, actorId, taskId }; +const store = new DurableCapabilityLiveSessionStore({ directory, binding }); const service = new CapabilityLiveSessionService({ store }); -const session = await service.create({ workingDirectory }); +const session = await service.create({ ...binding, attemptId, 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); ``` +After a worker terminates, a new worker must construct a new service and use the +production resume entrypoint. It must not call `create` again: + +```ts +const resumed = await new CapabilityLiveSessionService({ store }).resume({ + sessionId, + attemptId: resumeAttemptId, + resumeOf: killedAttemptId, +}); +await resumed.reconcileActiveTurn(); +``` + +`resume` first commits the killed attempt as `terminated` and the successor as +`running`. Only then does it start runnerd, read the checkpointed provider +thread, and resume that exact thread. A missing or corrupt checkpoint, authority +binding mismatch, attempt-lineage mismatch, or provider thread/session drift +fails closed. The killed attempt and its usage remain immutable. + The handoff surface for later tracks is: - `create(input)` starts runnerd, Codex, one mock run, and one dynamic-tool thread. @@ -48,14 +72,23 @@ The handoff surface for later tracks is: 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. +- `resume({ sessionId, attemptId, resumeOf })` is the cross-worker production + path and records distinct linked attempts before provider recovery. +- `recordUsage(receipt)` durably commits an attempt-bound, exactly-once provider + response receipt before the caller acknowledges that response. Reusing a + receipt with different contents fails closed. +- `reconcileActiveTurn()` interrupts and records the terminal fact for a turn + that was active in the checkpoint when its worker terminated. - `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 `CapabilityLiveSessionStore` for refresh and -server-restart survival. The included in-memory store is intended for tests and -single-process consumers. +`DurableCapabilityLiveSessionStore` writes a checksummed, revisioned checkpoint +with atomic rename plus file and directory fsync. It persists provider identity, +mock state and semantic idempotency receipts, attempt lineage, active and +terminal turn facts, and the usage ledger. The included in-memory store remains +limited to 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 diff --git a/packages/paperclip-runner/package.json b/packages/paperclip-runner/package.json index d48495d6d71..c8a2041339c 100644 --- a/packages/paperclip-runner/package.json +++ b/packages/paperclip-runner/package.json @@ -17,6 +17,10 @@ "types": "./dist/evals/index.d.ts", "import": "./dist/evals/index.js" }, + "./live": { + "types": "./dist/live/index.d.ts", + "import": "./dist/live/index.js" + }, "./browser": { "types": "./dist/browser/index.d.ts", "import": "./dist/browser/index.js" 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 f9fab1a101a..85e77f20968 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,7 @@ +use std::io::{self, BufRead, BufReader, Write}; use std::path::PathBuf; use std::process::{Command, ExitCode, Stdio}; +use std::thread; use std::time::Duration; use paperclip_runner_core::durable::{ @@ -90,8 +92,8 @@ fn run_codex_app_server_proxy(args: &[String]) -> Result<(), LocalRunnerError> { command .args(&codex_args) .env_clear() - .stdin(Stdio::inherit()) - .stdout(Stdio::inherit()) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) .stderr(Stdio::inherit()); for key in [ "ALL_PROXY", @@ -119,6 +121,40 @@ fn run_codex_app_server_proxy(args: &[String]) -> Result<(), LocalRunnerError> { let mut child = command.spawn().map_err(|error| { LocalRunnerError::invalid(format!("failed to start Codex app-server: {error}")) })?; + let mut child_stdin = child + .stdin + .take() + .ok_or_else(|| LocalRunnerError::invalid("failed to capture Codex app-server stdin"))?; + let mut child_stdout = child + .stdout + .take() + .ok_or_else(|| LocalRunnerError::invalid("failed to capture Codex app-server stdout"))?; + thread::spawn(move || { + let mut stdin = io::stdin().lock(); + let mut line = Vec::new(); + while stdin.read_until(b'\n', &mut line).unwrap_or(0) > 0 { + if child_stdin.write_all(&line).is_err() || child_stdin.flush().is_err() { + break; + } + line.clear(); + } + }); + let stdout_relay = thread::spawn(move || { + let mut stdout = io::stdout().lock(); + let mut child_stdout = BufReader::new(&mut child_stdout); + let mut line = Vec::new(); + let mut relayed = 0; + loop { + let bytes = child_stdout.read_until(b'\n', &mut line)?; + if bytes == 0 { + return Ok::(relayed); + } + stdout.write_all(&line)?; + stdout.flush()?; + relayed += bytes as u64; + line.clear(); + } + }); eprintln!( "paperclip-runnerd: native codex proxy started runner_pid={} codex_pid={}", std::process::id(), @@ -127,6 +163,12 @@ fn run_codex_app_server_proxy(args: &[String]) -> Result<(), LocalRunnerError> { let status = child.wait().map_err(|error| { LocalRunnerError::invalid(format!("failed to wait for Codex app-server: {error}")) })?; + stdout_relay + .join() + .map_err(|_| LocalRunnerError::invalid("Codex app-server stdout relay panicked"))? + .map_err(|error| { + LocalRunnerError::invalid(format!("Codex app-server stdout relay failed: {error}")) + })?; eprintln!( "paperclip-runnerd: native codex proxy exited success={} code={}", status.success(), diff --git a/packages/paperclip-runner/src/live/durable-live-session-store.ts b/packages/paperclip-runner/src/live/durable-live-session-store.ts new file mode 100644 index 00000000000..b2537d97ccd --- /dev/null +++ b/packages/paperclip-runner/src/live/durable-live-session-store.ts @@ -0,0 +1,174 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdir, open, readFile, rename, unlink } from "node:fs/promises"; +import { join, resolve } from "node:path"; + +import { + assertCapabilityLiveSessionSnapshot, + type CapabilityLiveSessionSnapshot, + type CapabilityLiveSessionStore, +} from "./live-session.js"; + +const DURABLE_STORE_SCHEMA = "paperclip.capability.live-session-checkpoint.v1" as const; + +export interface CapabilityLiveSessionStoreBinding { + sessionId: string; + runId: string; + companyId: string; + actorId: string; + taskId: string; +} + +interface DurableCheckpointEnvelope { + schema: typeof DURABLE_STORE_SCHEMA; + binding: CapabilityLiveSessionStoreBinding; + snapshotSha256: string; + snapshot: CapabilityLiveSessionSnapshot; +} + +export interface DurableCapabilityLiveSessionStoreOptions { + directory: string; + binding: CapabilityLiveSessionStoreBinding; +} + +function sha256(value: string): string { + return createHash("sha256").update(value).digest("hex"); +} + +function requireIdentifier(value: string, name: string): void { + if (value.trim().length === 0) throw new Error(`capability_live_store_${name}_required`); +} + +function sameBinding( + left: CapabilityLiveSessionStoreBinding, + right: CapabilityLiveSessionStoreBinding, +): boolean { + return left.sessionId === right.sessionId + && left.runId === right.runId + && left.companyId === right.companyId + && left.actorId === right.actorId + && left.taskId === right.taskId; +} + +/** + * Attempt-bound, crash-safe checkpoint storage for the governed live path. + * + * Each replacement is written, fsynced, renamed, and followed by a directory + * fsync. A checksum, schema check, revision check, and immutable run binding + * make partial writes, stale writers, corrupt JSON, and cross-attempt reuse + * fail closed rather than silently opening another provider thread. + */ +export class DurableCapabilityLiveSessionStore implements CapabilityLiveSessionStore { + readonly #directory: string; + readonly #binding: CapabilityLiveSessionStoreBinding; + readonly #checkpointPath: string; + + constructor(options: DurableCapabilityLiveSessionStoreOptions) { + this.#directory = resolve(options.directory); + this.#binding = structuredClone(options.binding); + for (const [name, value] of Object.entries(this.#binding)) requireIdentifier(value, name); + const key = sha256(this.#binding.sessionId); + this.#checkpointPath = join(this.#directory, `${key}.json`); + } + + async load(sessionId: string): Promise { + this.#assertSessionId(sessionId); + let raw: string; + try { + raw = await readFile(this.#checkpointPath, "utf8"); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } + let envelope: DurableCheckpointEnvelope; + try { + envelope = JSON.parse(raw) as DurableCheckpointEnvelope; + } catch { + throw new Error("capability_live_checkpoint_corrupt: invalid JSON"); + } + if (envelope.schema !== DURABLE_STORE_SCHEMA) { + throw new Error("capability_live_checkpoint_corrupt: unsupported envelope schema"); + } + if (!sameBinding(envelope.binding, this.#binding)) { + throw new Error("capability_live_checkpoint_binding_mismatch"); + } + assertCapabilityLiveSessionSnapshot(envelope.snapshot); + const serialized = JSON.stringify(envelope.snapshot); + if (sha256(serialized) !== envelope.snapshotSha256) { + throw new Error("capability_live_checkpoint_corrupt: checksum mismatch"); + } + this.#assertSnapshotBinding(envelope.snapshot); + return structuredClone(envelope.snapshot); + } + + async save(snapshot: CapabilityLiveSessionSnapshot): Promise { + assertCapabilityLiveSessionSnapshot(snapshot); + this.#assertSnapshotBinding(snapshot); + await mkdir(this.#directory, { recursive: true, mode: 0o700 }); + + const current = await this.load(snapshot.sessionId); + if (current !== null) { + if (current.revision > snapshot.revision) { + throw new Error("capability_live_checkpoint_stale_revision"); + } + if (current.revision === snapshot.revision) { + if (JSON.stringify(current) === JSON.stringify(snapshot)) return; + throw new Error("capability_live_checkpoint_revision_conflict"); + } + } + + const serialized = JSON.stringify(snapshot); + const envelope: DurableCheckpointEnvelope = { + schema: DURABLE_STORE_SCHEMA, + binding: structuredClone(this.#binding), + snapshotSha256: sha256(serialized), + snapshot: structuredClone(snapshot), + }; + const temporaryPath = `${this.#checkpointPath}.${process.pid}.${randomUUID()}.tmp`; + const file = await open(temporaryPath, "wx", 0o600); + try { + await file.writeFile(`${JSON.stringify(envelope)}\n`, "utf8"); + await file.sync(); + } finally { + await file.close(); + } + try { + await rename(temporaryPath, this.#checkpointPath); + const directory = await open(this.#directory, "r"); + try { + await directory.sync(); + } finally { + await directory.close(); + } + } catch (error) { + await unlink(temporaryPath).catch(() => undefined); + throw error; + } + } + + async delete(sessionId: string): Promise { + this.#assertSessionId(sessionId); + await unlink(this.#checkpointPath).catch((error: NodeJS.ErrnoException) => { + if (error.code !== "ENOENT") throw error; + }); + } + + #assertSessionId(sessionId: string): void { + if (sessionId !== this.#binding.sessionId) { + throw new Error("capability_live_checkpoint_binding_mismatch"); + } + } + + #assertSnapshotBinding(snapshot: CapabilityLiveSessionSnapshot): void { + const authority = snapshot.authority; + const actual: CapabilityLiveSessionStoreBinding = { + sessionId: snapshot.sessionId, + runId: authority.runId, + companyId: authority.companyId, + actorId: authority.actorId, + taskId: authority.taskId, + }; + if (!sameBinding(actual, this.#binding)) { + throw new Error("capability_live_checkpoint_binding_mismatch"); + } + } +} diff --git a/packages/paperclip-runner/src/live/index.ts b/packages/paperclip-runner/src/live/index.ts index 6d359d462e9..2ae16f01d4a 100644 --- a/packages/paperclip-runner/src/live/index.ts +++ b/packages/paperclip-runner/src/live/index.ts @@ -1,5 +1,6 @@ export * from "../scenarios/index.js"; export * from "./clean-room.js"; export * from "./live-session.js"; +export * from "./durable-live-session-store.js"; export * from "./runnerd-codex-transport.js"; export * from "./turn-stream.js"; diff --git a/packages/paperclip-runner/src/live/live-session.test.ts b/packages/paperclip-runner/src/live/live-session.test.ts index 185900756e6..706d77561e0 100644 --- a/packages/paperclip-runner/src/live/live-session.test.ts +++ b/packages/paperclip-runner/src/live/live-session.test.ts @@ -1,4 +1,10 @@ import { describe, expect, it, vi } from "vitest"; +import { createHash } from "node:crypto"; +import { existsSync } from "node:fs"; +import { mkdtemp, readFile, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; import type { CodexAppServerTransport, @@ -9,8 +15,11 @@ import type { import { InMemoryCapabilityLiveSessionStore, CapabilityLiveSessionService, + reconcileCapabilityLiveUsage, type CapabilityLiveTransportFactory, } from "./live-session.js"; +import { DurableCapabilityLiveSessionStore } from "./durable-live-session-store.js"; +import { defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js"; class AsyncNotifications implements AsyncIterable { #values: CodexRpcNotification[] = []; @@ -46,6 +55,8 @@ interface FakeProviderState { nextTurn: number; lastToolResult: string; transports: FakeCapabilityCodexTransport[]; + turns: Map; + holdAfterTool: boolean; } class FakeCapabilityCodexTransport implements CodexAppServerTransport { @@ -74,7 +85,13 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { }; } if (method === "thread/read") { - return { thread: { id: this.state.threadId, sessionId: this.state.providerSessionId, turns: [] } }; + return { + thread: { + id: this.state.threadId, + sessionId: this.state.providerSessionId, + turns: [...this.state.turns].map(([id, status]) => ({ id, status })), + }, + }; } if (method === "thread/resume") { return { @@ -86,6 +103,7 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { if (method === "turn/start") { const turnId = `turn-${++this.state.nextTurn}`; this.#activeTurnId = turnId; + this.state.turns.set(turnId, "inProgress"); const input = Array.isArray(params.input) ? params.input[0] as Record : {}; const message = String(input.text ?? ""); void this.#runTurn(turnId, message); @@ -93,6 +111,7 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { } if (method === "turn/interrupt") { const turnId = String(params.turnId); + this.state.turns.set(turnId, "interrupted"); this.notificationsQueue.push({ method: "turn/completed", params: { threadId: this.state.threadId, turn: { id: turnId, status: "interrupted" } }, @@ -139,6 +158,9 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { await new Promise((resolve) => setTimeout(resolve, 0)); let assistantText: string; if (message.includes("progress")) { + const idempotencyKey = message.includes("idempotent") + ? "progress-governed-once" + : `progress-${turnId}`; const result = await this.#toolCall({ id: `request-${turnId}`, method: "item/tool/call", @@ -148,12 +170,13 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { callId: `call-${turnId}`, tool: "report_progress", arguments: { - idempotencyKey: `progress-${turnId}`, + idempotencyKey, body: "Progress persisted through the live Codex tool loop.", }, }, }); this.state.lastToolResult = String((result.contentItems as Array>)[0]?.text); + if (this.state.holdAfterTool) await new Promise(() => undefined); 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}`; @@ -191,6 +214,7 @@ class FakeCapabilityCodexTransport implements CodexAppServerTransport { item: { id: `message-${turnId}`, type: "agentMessage", text: assistantText }, }, }); + this.state.turns.set(turnId, "completed"); this.notificationsQueue.push({ method: "turn/completed", params: { threadId: this.state.threadId, turn: { id: turnId, status: "completed" } }, @@ -231,6 +255,8 @@ function providerState(): FakeProviderState { nextTurn: 0, lastToolResult: "nothing yet", transports: [], + turns: new Map(), + holdAfterTool: false, }; } @@ -332,4 +358,241 @@ describe("Capability live runnerd and Codex session", () => { expect(stopped.process?.runnerExited).toBe(true); expect((await store.load(replacement.id))?.status).toBe("closed"); }); + + it("durably resumes a distinct attempt, preserves partial usage, and deduplicates the semantic effect", async () => { + const state = providerState(); + const directory = await mkdtemp(join(tmpdir(), "capability-live-durable-")); + const binding = { + sessionId: "session-governed-resume", + runId: "run-governed-resume", + companyId: "company-1", + actorId: "actor-1", + taskId: "task-1", + }; + const firstStore = new DurableCapabilityLiveSessionStore({ directory, binding }); + const firstService = new CapabilityLiveSessionService({ + store: firstStore, + transportFactory: fakeTransportFactory(state), + }); + const first = await firstService.create({ + ...binding, + attemptId: "attempt-killed", + turnTimeoutMs: 500, + }); + state.holdAfterTool = true; + const killedTurn = first.sendMessage("Apply idempotent progress once."); + await vi.waitFor(async () => { + expect((await firstStore.load(binding.sessionId))?.mockState).toContain( + "Progress persisted through the live Codex tool loop.", + ); + }); + await expect(first.recordUsage({ + receiptId: "provider-response-1", + providerResponseId: "response-1", + turnId: "turn-1", + providerCalls: 1, + inputTokens: 100, + outputTokens: 20, + cachedInputTokens: 40, + reasoningTokens: 5, + costNanodollars: 1_500, + })).resolves.toBe("committed"); + await state.transports[0]?.close(); + await expect(killedTurn).rejects.toThrow("timed out"); + + // A new service models worker termination: it owns no in-memory session. + const resumedService = new CapabilityLiveSessionService({ + store: new DurableCapabilityLiveSessionStore({ directory, binding }), + transportFactory: fakeTransportFactory(state), + }); + const resumed = await resumedService.resume({ + sessionId: binding.sessionId, + attemptId: "attempt-resumed", + resumeOf: "attempt-killed", + }); + expect(resumed.snapshot().providerThreadId).toBe(state.threadId); + expect(resumed.snapshot().attempts).toMatchObject([ + { attemptId: "attempt-killed", status: "terminated", failureCode: "worker_terminated" }, + { attemptId: "attempt-resumed", resumeOf: "attempt-killed", status: "running" }, + ]); + + await expect(resumed.reconcileActiveTurn()).resolves.toMatchObject({ + turnId: "turn-1", + status: "interrupted", + }); + state.holdAfterTool = false; + + const duplicate = await resumed.sendMessage("Apply idempotent progress once."); + expect(duplicate.assistantText).toContain('"disposition":"duplicate"'); + expect(resumed.mockState().comments).toHaveLength(1); + await expect(resumed.recordUsage({ + receiptId: "provider-response-1", + providerResponseId: "response-1", + turnId: "turn-1", + providerCalls: 1, + inputTokens: 100, + outputTokens: 20, + cachedInputTokens: 40, + reasoningTokens: 5, + costNanodollars: 1_500, + })).resolves.toBe("duplicate"); + await resumed.recordUsage({ + receiptId: "provider-response-2", + providerResponseId: "response-2", + providerCalls: 1, + inputTokens: 80, + outputTokens: 10, + costNanodollars: 1_000, + }); + await resumed.completeAttempt("succeeded"); + + const final = await firstStore.load(binding.sessionId); + expect(final?.attempts).toMatchObject([ + { attemptId: "attempt-killed", status: "terminated" }, + { attemptId: "attempt-resumed", status: "succeeded" }, + ]); + expect(final?.usageLedger).toHaveLength(2); + expect(final && reconcileCapabilityLiveUsage(final)).toEqual({ + providerCalls: 2, + inputTokens: 180, + outputTokens: 30, + cachedInputTokens: 40, + reasoningTokens: 5, + costNanodollars: 2_500, + }); + }); + + it("fails closed for missing, corrupt, mismatched, and provider-drifted checkpoints", async () => { + const directory = await mkdtemp(join(tmpdir(), "capability-live-fail-closed-")); + const binding = { + sessionId: "session-fail-closed", + runId: "run-fail-closed", + companyId: "company-1", + actorId: "actor-1", + taskId: "task-1", + }; + const missingService = new CapabilityLiveSessionService({ + store: new DurableCapabilityLiveSessionStore({ directory, binding }), + transportFactory: fakeTransportFactory(providerState()), + }); + await expect(missingService.resume({ + sessionId: binding.sessionId, + attemptId: "attempt-missing-resume", + resumeOf: "attempt-missing", + })).rejects.toThrow("capability_live_resume_checkpoint_missing"); + + const state = providerState(); + const store = new DurableCapabilityLiveSessionStore({ directory, binding }); + const firstService = new CapabilityLiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + await firstService.create({ ...binding, attemptId: "attempt-original" }); + const mismatched = new DurableCapabilityLiveSessionStore({ + directory, + binding: { ...binding, taskId: "task-other" }, + }); + await expect(mismatched.load(binding.sessionId)).rejects.toThrow( + "capability_live_checkpoint_binding_mismatch", + ); + + state.providerSessionId = "codex-provider-drifted"; + const driftedService = new CapabilityLiveSessionService({ + store, + transportFactory: fakeTransportFactory(state), + }); + await expect(driftedService.resume({ + sessionId: binding.sessionId, + attemptId: "attempt-drifted", + resumeOf: "attempt-original", + })).rejects.toThrow("capability_live_provider_session_drift"); + expect((await store.load(binding.sessionId))?.attempts?.at(-1)).toMatchObject({ + attemptId: "attempt-drifted", + status: "failed", + failureCode: "provider_recovery_failed", + }); + + const checkpointFile = join(directory, `${createHash("sha256").update(binding.sessionId).digest("hex")}.json`); + const checkpoint = await readFile(checkpointFile, "utf8"); + await writeFile(checkpointFile, checkpoint.replace('"snapshotSha256":"', '"snapshotSha256":"corrupt-')); + await expect(store.load(binding.sessionId)).rejects.toThrow("capability_live_checkpoint_corrupt"); + }); + + it.skipIf(process.platform === "win32" || !existsSync(defaultCapabilityRunnerdBinary()))( + "terminates real runnerd after a durable receipt and resumes its exact provider thread", + async () => { + const directory = await mkdtemp(join(tmpdir(), "capability-live-real-runnerd-")); + const providerStatePath = join(directory, "provider-state.json"); + const fixture = fileURLToPath(new URL("../../test/fixtures/fake-durable-codex-app-server.mjs", import.meta.url)); + const binding = { + sessionId: "session-real-runnerd-resume", + runId: "run-real-runnerd-resume", + companyId: "company-1", + actorId: "actor-1", + taskId: "task-1", + }; + const store = new DurableCapabilityLiveSessionStore({ directory, binding }); + const transportOptions = { + codexCommand: process.execPath, + codexArgs: [fixture, providerStatePath], + closeGraceMs: 100, + }; + const firstService = new CapabilityLiveSessionService({ store, transportOptions }); + const first = await firstService.create({ + ...binding, + attemptId: "attempt-real-killed", + turnTimeoutMs: 2_000, + }); + const killedTurn = first.sendMessage("Apply the governed idempotent effect."); + await vi.waitFor(async () => { + const checkpoint = await store.load(binding.sessionId); + expect(checkpoint?.mockState).toContain("One durable governed effect."); + expect(checkpoint?.activeTurnId).toBe("turn-1"); + expect(checkpoint?.process?.runnerProcessGroupId).not.toBeNull(); + expect(checkpoint?.process?.codexPid).not.toBeNull(); + }); + await first.recordUsage({ + receiptId: "real-response-1", + providerResponseId: "fixture-response-1", + turnId: "turn-1", + providerCalls: 1, + inputTokens: 10, + outputTokens: 2, + costNanodollars: 100, + }); + const killedCheckpoint = await store.load(binding.sessionId); + const processGroupId = killedCheckpoint?.process?.runnerProcessGroupId; + expect(processGroupId).toBeTypeOf("number"); + process.kill(-processGroupId!, "SIGKILL"); + await expect(killedTurn).rejects.toThrow(); + + const resumedService = new CapabilityLiveSessionService({ + store: new DurableCapabilityLiveSessionStore({ directory, binding }), + transportOptions, + }); + const resumed = await resumedService.resume({ + sessionId: binding.sessionId, + attemptId: "attempt-real-resumed", + resumeOf: "attempt-real-killed", + }); + expect(resumed.snapshot().providerThreadId).toBe("thread-durable-runnerd"); + await expect(resumed.reconcileActiveTurn()).resolves.toMatchObject({ status: "interrupted" }); + const duplicate = await resumed.sendMessage("Apply the governed idempotent effect again."); + expect(duplicate.assistantText).toContain("duplicate"); + expect(resumed.mockState().comments).toHaveLength(1); + await resumed.completeAttempt("succeeded"); + const final = await store.load(binding.sessionId); + expect(final?.attempts).toMatchObject([ + { attemptId: "attempt-real-killed", status: "terminated" }, + { attemptId: "attempt-real-resumed", status: "succeeded", resumeOf: "attempt-real-killed" }, + ]); + expect(final?.usageLedger).toHaveLength(1); + expect(final?.terminalTurns).toEqual(expect.arrayContaining([ + expect.objectContaining({ turnId: "turn-1", status: "interrupted" }), + expect.objectContaining({ turnId: "turn-2", status: "completed" }), + ])); + await resumedService.shutdown(resumed.id, "test complete"); + }, + 15_000, + ); }); diff --git a/packages/paperclip-runner/src/live/live-session.ts b/packages/paperclip-runner/src/live/live-session.ts index bcf1f9b7323..a19bfe3963f 100644 --- a/packages/paperclip-runner/src/live/live-session.ts +++ b/packages/paperclip-runner/src/live/live-session.ts @@ -95,6 +95,50 @@ export interface CapabilityLiveSessionConfigSnapshot { turnTimeoutMs: number; } +export type CapabilityLiveAttemptStatus = "running" | "succeeded" | "failed" | "terminated"; + +export interface CapabilityLiveAttemptSnapshot { + attemptId: string; + resumeOf: string | null; + status: CapabilityLiveAttemptStatus; + startedAt: string; + finishedAt: string | null; + failureCode: string | null; +} + +export interface CapabilityLiveTurnTerminalFact { + turnId: string; + status: CapabilityLiveTurnResult["status"]; + reconciledByAttemptId: string; + observedAt: string; +} + +export interface CapabilityLiveUsageReceipt { + receiptId: string; + attemptId: string; + providerResponseId: string | null; + turnId: string | null; + providerCalls: number; + inputTokens: number; + outputTokens: number; + cachedInputTokens: number; + reasoningTokens: number; + costNanodollars: number; + observedAt: string; +} + +export interface RecordCapabilityLiveUsageInput { + receiptId: string; + providerResponseId?: string | null; + turnId?: string | null; + providerCalls?: number; + inputTokens?: number; + outputTokens?: number; + cachedInputTokens?: number; + reasoningTokens?: number; + costNanodollars?: number; +} + export interface CapabilityLiveSessionSnapshot { schema: typeof LIVE_SESSION_SCHEMA; revision: number; @@ -118,6 +162,13 @@ export interface CapabilityLiveSessionSnapshot { realPaperclipRequests: 0; childPaperclipEnvironmentKeys: string[]; }; + /** Governed worker attempts. A resumed worker is a new attempt, never a rewrite. */ + attempts?: CapabilityLiveAttemptSnapshot[]; + currentAttemptId?: string; + /** Provider terminal facts reconciled by the attempt that observed them. */ + terminalTurns?: CapabilityLiveTurnTerminalFact[]; + /** Exact-once provider usage/cost receipts retained across every attempt. */ + usageLedger?: CapabilityLiveUsageReceipt[]; } export interface CreateCapabilityLiveSessionInput { @@ -131,9 +182,16 @@ export interface CreateCapabilityLiveSessionInput { taskId?: string; runId?: string; sessionId?: string; + attemptId?: string; turnTimeoutMs?: number; } +export interface ResumeCapabilityLiveSessionInput { + sessionId: string; + attemptId: string; + resumeOf: string; +} + export interface CapabilityLiveTurnResult { turnId: string; status: "completed" | "failed" | "interrupted" | "cancelled"; @@ -268,6 +326,158 @@ function jsonValue(value: unknown): CapabilityJsonValue { return JSON.parse(JSON.stringify(value ?? null)) as CapabilityJsonValue; } +function nonNegativeInteger(value: number | undefined, name: string): number { + const normalized = value ?? 0; + if (!Number.isSafeInteger(normalized) || normalized < 0) { + throw new Error(`capability_live_usage_${name}_invalid`); + } + return normalized; +} + +function requireNonEmpty(value: string, name: string): string { + if (value.trim().length === 0) throw new Error(`capability_live_${name}_required`); + return value; +} + +function providerTurnStatus(value: unknown): CapabilityLiveTurnResult["status"] | null { + const status = text(value).toLowerCase(); + if (status === "completed") return "completed"; + if (status === "failed") return "failed"; + if (status === "interrupted") return "interrupted"; + if (status === "cancelled" || status === "canceled") return "cancelled"; + return null; +} + +/** Reject malformed or partially written checkpoints before any provider process starts. */ +export function assertCapabilityLiveSessionSnapshot( + value: unknown, +): asserts value is CapabilityLiveSessionSnapshot { + const snapshot = record(value); + if (snapshot.schema !== LIVE_SESSION_SCHEMA) { + throw new Error("capability_live_checkpoint_corrupt: unsupported snapshot schema"); + } + for (const name of ["sessionId", "providerThreadId", "mockState", "createdAt", "updatedAt"] as const) { + if (text(snapshot[name]).length === 0) { + throw new Error(`capability_live_checkpoint_corrupt: missing ${name}`); + } + } + if (!Number.isSafeInteger(snapshot.revision) || Number(snapshot.revision) < 1) { + throw new Error("capability_live_checkpoint_corrupt: invalid revision"); + } + const authority = record(snapshot.authority); + for (const name of ["runId", "companyId", "actorId", "taskId", "sessionId"] as const) { + if (text(authority[name]).length === 0) { + throw new Error(`capability_live_checkpoint_corrupt: missing authority.${name}`); + } + } + if (authority.sessionId !== snapshot.sessionId) { + throw new Error("capability_live_checkpoint_binding_mismatch"); + } + if (typeof snapshot.mockState !== "string") { + throw new Error("capability_live_checkpoint_corrupt: invalid mockState"); + } + try { + JSON.parse(snapshot.mockState); + } catch { + throw new Error("capability_live_checkpoint_corrupt: invalid mockState JSON"); + } + if (snapshot.attempts !== undefined && !Array.isArray(snapshot.attempts)) { + throw new Error("capability_live_checkpoint_corrupt: invalid attempts"); + } + const attemptIds = new Set(); + for (const value of snapshot.attempts ?? []) { + const attempt = record(value); + const attemptId = text(attempt.attemptId); + if ( + attemptId.length === 0 || + attemptIds.has(attemptId) || + !["running", "succeeded", "failed", "terminated"].includes(text(attempt.status)) || + text(attempt.startedAt).length === 0 + ) { + throw new Error("capability_live_checkpoint_corrupt: invalid attempt"); + } + const resumeOf = attempt.resumeOf === null ? null : text(attempt.resumeOf); + if (resumeOf !== null && !attemptIds.has(resumeOf)) { + throw new Error("capability_live_checkpoint_corrupt: invalid attempt lineage"); + } + attemptIds.add(attemptId); + } + if ( + snapshot.currentAttemptId !== undefined && + !attemptIds.has(text(snapshot.currentAttemptId)) + ) { + throw new Error("capability_live_checkpoint_corrupt: invalid current attempt"); + } + if (snapshot.terminalTurns !== undefined && !Array.isArray(snapshot.terminalTurns)) { + throw new Error("capability_live_checkpoint_corrupt: invalid terminalTurns"); + } + const terminalTurnIds = new Set(); + for (const value of snapshot.terminalTurns ?? []) { + const terminal = record(value); + const turnId = text(terminal.turnId); + if ( + turnId.length === 0 || + terminalTurnIds.has(turnId) || + !["completed", "failed", "interrupted", "cancelled"].includes(text(terminal.status)) || + text(terminal.reconciledByAttemptId).length === 0 || + text(terminal.observedAt).length === 0 + ) { + throw new Error("capability_live_checkpoint_corrupt: invalid terminal fact"); + } + terminalTurnIds.add(turnId); + } + if (snapshot.usageLedger !== undefined && !Array.isArray(snapshot.usageLedger)) { + throw new Error("capability_live_checkpoint_corrupt: invalid usageLedger"); + } + const usageReceiptIds = new Set(); + for (const value of snapshot.usageLedger ?? []) { + const receipt = record(value); + const receiptId = text(receipt.receiptId); + if ( + receiptId.length === 0 || + usageReceiptIds.has(receiptId) || + text(receipt.attemptId).length === 0 || + text(receipt.observedAt).length === 0 + ) { + throw new Error("capability_live_checkpoint_corrupt: invalid usage receipt"); + } + for (const name of [ + "providerCalls", + "inputTokens", + "outputTokens", + "cachedInputTokens", + "reasoningTokens", + "costNanodollars", + ] as const) { + if (!Number.isSafeInteger(receipt[name]) || Number(receipt[name]) < 0) { + throw new Error("capability_live_checkpoint_corrupt: invalid usage receipt"); + } + } + usageReceiptIds.add(receiptId); + } +} + +export function reconcileCapabilityLiveUsage( + snapshot: CapabilityLiveSessionSnapshot, +): Omit { + const unique = new Map((snapshot.usageLedger ?? []).map((receipt) => [receipt.receiptId, receipt])); + return [...unique.values()].reduce((total, receipt) => ({ + providerCalls: total.providerCalls + receipt.providerCalls, + inputTokens: total.inputTokens + receipt.inputTokens, + outputTokens: total.outputTokens + receipt.outputTokens, + cachedInputTokens: total.cachedInputTokens + receipt.cachedInputTokens, + reasoningTokens: total.reasoningTokens + receipt.reasoningTokens, + costNanodollars: total.costNanodollars + receipt.costNanodollars, + }), { + providerCalls: 0, + inputTokens: 0, + outputTokens: 0, + cachedInputTokens: 0, + reasoningTokens: 0, + costNanodollars: 0, + }); +} + function dynamicToolSpec(definition: CapabilitySemanticToolDefinition): Record { return { name: definition.name, @@ -352,6 +562,7 @@ export class CapabilityLiveSessionService { transportFactory: this.#transportFactory, transportOptions: this.#transportOptions, now: this.#now, + attemptId: input.attemptId ?? runId, }); this.#sessions.set(session.id, session); return session; @@ -362,6 +573,7 @@ export class CapabilityLiveSessionService { if (live !== undefined) return live; const snapshot = await this.#store.load(sessionId); if (snapshot === null) throw new Error(`capability live session ${sessionId} was not found`); + assertCapabilityLiveSessionSnapshot(snapshot); const session = await CapabilityLiveSession.restore({ snapshot, store: this.#store, @@ -373,6 +585,92 @@ export class CapabilityLiveSessionService { return session; } + /** + * Production worker-restart entrypoint. It records the killed attempt and its + * successor durably before runnerd/Codex is started, then restores only the + * provider thread named by that checkpoint. + */ + async resume(input: ResumeCapabilityLiveSessionInput): Promise { + requireNonEmpty(input.attemptId, "attempt_id"); + requireNonEmpty(input.resumeOf, "resume_of"); + if (input.attemptId === input.resumeOf) throw new Error("capability_live_resume_cycle"); + if (this.#sessions.has(input.sessionId)) { + throw new Error("capability_live_resume_requires_new_worker"); + } + const snapshot = await this.#store.load(input.sessionId); + if (snapshot === null) { + throw new Error("capability_live_resume_checkpoint_missing: refusing to open a second provider session"); + } + assertCapabilityLiveSessionSnapshot(snapshot); + const attempts = structuredClone(snapshot.attempts ?? [{ + attemptId: snapshot.authority.runId, + resumeOf: null, + status: "running" as const, + startedAt: snapshot.createdAt, + finishedAt: null, + failureCode: null, + }]); + const currentAttemptId = snapshot.currentAttemptId ?? attempts.at(-1)?.attemptId; + if (currentAttemptId !== input.resumeOf) throw new Error("capability_live_resume_binding_mismatch"); + if (attempts.some((attempt) => attempt.attemptId === input.attemptId)) { + throw new Error("capability_live_resume_attempt_exists"); + } + const previous = attempts.find((attempt) => attempt.attemptId === input.resumeOf); + if (previous === undefined || previous.status !== "running") { + throw new Error("capability_live_resume_source_not_running"); + } + const resumedAt = this.#now().toISOString(); + previous.status = "terminated"; + previous.finishedAt = resumedAt; + previous.failureCode = "worker_terminated"; + attempts.push({ + attemptId: input.attemptId, + resumeOf: input.resumeOf, + status: "running", + startedAt: resumedAt, + finishedAt: null, + failureCode: null, + }); + const prepared: CapabilityLiveSessionSnapshot = { + ...snapshot, + revision: snapshot.revision + 1, + updatedAt: resumedAt, + status: "starting", + attempts, + currentAttemptId: input.attemptId, + }; + await this.#store.save(prepared); + try { + const session = await CapabilityLiveSession.restore({ + snapshot: prepared, + store: this.#store, + transportFactory: this.#transportFactory, + transportOptions: this.#transportOptions, + now: this.#now, + }); + this.#sessions.set(session.id, session); + return session; + } catch (error) { + const failedAt = this.#now().toISOString(); + const latest = await this.#store.load(input.sessionId) ?? prepared; + const failedAttempts = structuredClone(latest.attempts ?? []); + const failed = failedAttempts.find((attempt) => attempt.attemptId === input.attemptId); + if (failed !== undefined) { + failed.status = "failed"; + failed.finishedAt = failedAt; + failed.failureCode = "provider_recovery_failed"; + } + await this.#store.save({ + ...latest, + revision: latest.revision + 1, + updatedAt: failedAt, + status: "failed", + attempts: failedAttempts, + }); + throw error; + } + } + async reconnect(sessionId: string): Promise { const session = await this.restore(sessionId); await session.reconnect(); @@ -419,9 +717,10 @@ interface OpenSessionOptions { transportFactory: CapabilityLiveTransportFactory; transportOptions: CapabilityRunnerdCodexTransportOptions; now: () => Date; + attemptId: string; } -interface RestoreSessionOptions extends Omit { +interface RestoreSessionOptions extends Omit { snapshot: CapabilityLiveSessionSnapshot; } @@ -438,6 +737,10 @@ export class CapabilityLiveSession { readonly #transcript: CapabilityLiveTranscriptEntry[]; readonly #evidence: CapabilityLiveEvidenceEntry[]; readonly #priorAuthorizationRecords: CapabilitySemanticAuthorizationRecord[]; + readonly #attempts: CapabilityLiveAttemptSnapshot[]; + readonly #terminalTurns: CapabilityLiveTurnTerminalFact[]; + readonly #usageLedger: CapabilityLiveUsageReceipt[]; + #currentAttemptId: string; #transport: CodexAppServerTransport | null = null; #processEvidence: CapabilityRunnerdProcessEvidence | null = null; #providerThreadId = ""; @@ -472,6 +775,18 @@ export class CapabilityLiveSession { this.#transcript = structuredClone(options.snapshot?.transcript ?? []); this.#evidence = structuredClone(options.snapshot?.evidence ?? []); this.#priorAuthorizationRecords = structuredClone(options.snapshot?.authorizationRecords ?? []); + const initialAttemptId = options.snapshot?.currentAttemptId ?? options.attemptId; + this.#attempts = structuredClone(options.snapshot?.attempts ?? [{ + attemptId: initialAttemptId, + resumeOf: null, + status: "running", + startedAt: this.#createdAt, + finishedAt: null, + failureCode: null, + }]); + this.#currentAttemptId = initialAttemptId; + this.#terminalTurns = structuredClone(options.snapshot?.terminalTurns ?? []); + this.#usageLedger = structuredClone(options.snapshot?.usageLedger ?? []); this.#providerThreadId = options.snapshot?.providerThreadId ?? ""; this.#providerSessionId = options.snapshot?.providerSessionId ?? null; this.#providerModel = options.snapshot?.providerModel === undefined @@ -484,7 +799,12 @@ export class CapabilityLiveSession { static async open(options: OpenSessionOptions): Promise { const session = new CapabilityLiveSession(options); - await session.#connect(false); + try { + await session.#connect(false); + } catch (error) { + await session.#abortConnect("initial provider connection failed"); + throw error; + } return session; } @@ -503,8 +823,14 @@ export class CapabilityLiveSession { authority: options.snapshot.authority, config: options.snapshot.config, snapshot: options.snapshot, + attemptId: options.snapshot.currentAttemptId ?? options.snapshot.authority.runId, }); - await session.#connect(true); + try { + await session.#connect(true); + } catch (error) { + await session.#abortConnect("provider recovery failed"); + throw error; + } return session; } @@ -624,9 +950,65 @@ export class CapabilityLiveSession { realPaperclipRequests: 0, childPaperclipEnvironmentKeys: childKeys.filter((key) => key.startsWith("PAPERCLIP_")), }, + attempts: structuredClone(this.#attempts), + currentAttemptId: this.#currentAttemptId, + terminalTurns: structuredClone(this.#terminalTurns), + usageLedger: structuredClone(this.#usageLedger), }; } + async recordUsage(input: RecordCapabilityLiveUsageInput): Promise<"committed" | "duplicate"> { + const receiptId = requireNonEmpty(input.receiptId, "usage_receipt_id"); + const receipt: CapabilityLiveUsageReceipt = { + receiptId, + attemptId: this.#currentAttemptId, + providerResponseId: input.providerResponseId ?? null, + turnId: input.turnId ?? this.#activeTurnId, + providerCalls: nonNegativeInteger(input.providerCalls ?? 1, "provider_calls"), + inputTokens: nonNegativeInteger(input.inputTokens, "input_tokens"), + outputTokens: nonNegativeInteger(input.outputTokens, "output_tokens"), + cachedInputTokens: nonNegativeInteger(input.cachedInputTokens, "cached_input_tokens"), + reasoningTokens: nonNegativeInteger(input.reasoningTokens, "reasoning_tokens"), + costNanodollars: nonNegativeInteger(input.costNanodollars, "cost_nanodollars"), + observedAt: this.#now().toISOString(), + }; + const existing = this.#usageLedger.find((candidate) => candidate.receiptId === receiptId); + if (existing !== undefined) { + const comparable = (value: CapabilityLiveUsageReceipt) => JSON.stringify({ + ...value, + attemptId: undefined, + observedAt: undefined, + }); + if (comparable(existing) !== comparable(receipt)) { + throw new Error("capability_live_usage_receipt_conflict"); + } + return "duplicate"; + } + this.#usageLedger.push(receipt); + await this.#persist(); + return "committed"; + } + + async completeAttempt( + status: Exclude, + failureCode: string | null = null, + ): Promise { + const attempt = this.#attempts.find((candidate) => candidate.attemptId === this.#currentAttemptId); + if (attempt === undefined || attempt.status !== "running") { + throw new Error("capability_live_attempt_not_running"); + } + if (status === "succeeded" && this.#activeTurnId !== null) { + throw new Error("capability_live_attempt_active_turn"); + } + attempt.status = status; + attempt.finishedAt = this.#now().toISOString(); + attempt.failureCode = status === "failed" + ? requireNonEmpty(failureCode ?? "attempt_failed", "attempt_failure_code") + : null; + await this.#persist(); + return this.snapshot(); + } + async sendMessage(message: string): Promise { const value = message.trim(); if (value.length === 0) throw new Error("Capability live messages cannot be empty"); @@ -643,21 +1025,7 @@ export class CapabilityLiveSession { // last transcript entry and the user message would never learn its turn. const userEntryId = this.#appendTranscript("user", value, null); let response: Record; - const terminal = new Promise>((resolve, reject) => { - const timer = setTimeout(() => { - const waiter = this.#turnWaiter; - this.#turnWaiter = null; - if (waiter !== null) { - this.#emit({ - turnId: this.#activeTurnId, - kind: "error", - message: `Capability Codex turn timed out after ${this.#config.turnTimeoutMs}ms`, - }); - } - reject(new Error(`Capability Codex turn timed out after ${this.#config.turnTimeoutMs}ms`)); - }, this.#config.turnTimeoutMs); - this.#turnWaiter = { resolve, reject, timer, assistantText: "", draftId: null }; - }); + const terminal = this.#armTurnWaiter(); try { response = await this.#transport.request("turn/start", { threadId: this.#providerThreadId, @@ -693,6 +1061,45 @@ export class CapabilityLiveSession { return { ...result, snapshot: this.snapshot() }; } + #armTurnWaiter(): Promise> { + if (this.#turnWaiter !== null) throw new Error("Capability live session already has a turn waiter"); + return new Promise>((resolve, reject) => { + const timer = setTimeout(() => { + const waiter = this.#turnWaiter; + this.#turnWaiter = null; + if (waiter !== null) { + this.#emit({ + turnId: this.#activeTurnId, + kind: "error", + message: `Capability Codex turn timed out after ${this.#config.turnTimeoutMs}ms`, + }); + } + reject(new Error(`Capability Codex turn timed out after ${this.#config.turnTimeoutMs}ms`)); + }, this.#config.turnTimeoutMs); + this.#turnWaiter = { resolve, reject, timer, assistantText: "", draftId: null }; + }); + } + + /** Interrupts and durably reconciles a checkpointed active turn after restart. */ + async reconcileActiveTurn( + reason = "worker restarted while provider turn was active", + ): Promise { + if (this.#activeTurnId === null) return null; + const turnId = this.#activeTurnId; + const terminal = this.#armTurnWaiter(); + try { + await this.interrupt(reason); + } catch (error) { + this.#rejectTurn(error); + await terminal.catch(() => undefined); + throw error; + } + const result = await terminal; + if (result.turnId !== turnId) throw new Error("capability_live_provider_session_drift: terminal turn mismatch"); + 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"; @@ -811,7 +1218,27 @@ export class CapabilityLiveSession { }); const thread = record(read.thread); if (text(thread.id) !== this.#providerThreadId) { - throw new Error("Codex reconnect read a different provider thread"); + throw new Error("capability_live_provider_session_drift: thread/read identity mismatch"); + } + const readSessionId = text(thread.sessionId); + if ( + this.#providerSessionId !== null + && readSessionId.length > 0 + && readSessionId !== this.#providerSessionId + ) { + throw new Error("capability_live_provider_session_drift: thread/read session mismatch"); + } + if (this.#activeTurnId !== null) { + const turns = Array.isArray(thread.turns) ? thread.turns.map(record) : []; + const recoveredTurn = turns.find((turn) => text(turn.id) === this.#activeTurnId); + if (recoveredTurn === undefined) { + throw new Error("capability_live_provider_session_drift: checkpointed active turn missing"); + } + const recoveredTerminal = providerTurnStatus(recoveredTurn.status); + if (recoveredTerminal !== null) { + this.#recordTerminalFact(this.#activeTurnId, recoveredTerminal); + this.#activeTurnId = null; + } } const resumed = await this.#transport.request("thread/resume", { threadId: this.#providerThreadId, @@ -824,11 +1251,18 @@ export class CapabilityLiveSession { }); const resumedThread = record(resumed.thread); if (text(resumedThread.id) !== this.#providerThreadId) { - throw new Error("Codex reconnect resumed a different provider thread"); + throw new Error("capability_live_provider_session_drift: thread/resume identity mismatch"); } - this.#providerSessionId = text(resumedThread.sessionId) || this.#providerSessionId; + const resumedSessionId = text(resumedThread.sessionId); + if ( + this.#providerSessionId !== null + && resumedSessionId.length > 0 + && resumedSessionId !== this.#providerSessionId + ) { + throw new Error("capability_live_provider_session_drift: thread/resume session mismatch"); + } + this.#providerSessionId = resumedSessionId || this.#providerSessionId; this.#captureProviderModel(resumed); - this.#activeTurnId = null; } else { const opened = await this.#transport.request("thread/start", { cwd: this.#config.workingDirectory, @@ -847,7 +1281,7 @@ export class CapabilityLiveSession { this.#providerSessionId = text(thread.sessionId) || text(record(initialized.user).sessionId) || null; this.#captureProviderModel(opened); } - this.#status = "idle"; + this.#status = this.#activeTurnId === null ? "idle" : "running"; this.#appendEvidence("session", null, { action: resume ? "resumed" : "started", sessionId: this.id, @@ -883,6 +1317,11 @@ export class CapabilityLiveSession { }); } + async #abortConnect(reason: string): Promise { + await this.#disconnect(reason).catch(() => undefined); + await this.#port.stop().catch(() => undefined); + } + async #handleServerRequest(request: CodexRpcServerRequest): Promise> { if (request.method !== "item/tool/call") { return { @@ -894,6 +1333,19 @@ export class CapabilityLiveSession { const threadId = text(request.params.threadId); const turnId = text(request.params.turnId); const callId = text(request.params.callId, String(request.id)); + // Codex may issue its first tool request immediately after writing the + // turn/start response, before the response and turn/started notification + // have crossed the two async readers. Bind that request only while this + // session is already waiting for the single turn it just started. + if ( + this.#activeTurnId === null && + this.#turnWaiter !== null && + this.#status === "running" && + threadId === this.#providerThreadId && + turnId.length > 0 + ) { + this.#activeTurnId = turnId; + } if ( threadId !== this.#providerThreadId || turnId.length === 0 || @@ -1011,17 +1463,38 @@ export class CapabilityLiveSession { #completeTurn(turnId: string, status: CapabilityLiveTurnResult["status"]): void { const waiter = this.#turnWaiter; - if (waiter === null) return; + if (waiter === null) { + this.#recordTerminalFact(turnId, status); + this.#activeTurnId = null; + this.#status = "idle"; + this.#emit({ turnId, kind: "terminal", status, assistantText: "" }); + return; + } clearTimeout(waiter.timer); this.#turnWaiter = null; const assistantText = waiter.assistantText.trim(); this.#settleAssistantDraft(waiter, assistantText, turnId); + this.#recordTerminalFact(turnId, status); this.#activeTurnId = null; this.#status = "idle"; this.#emit({ turnId, kind: "terminal", status, assistantText }); waiter.resolve({ turnId, status, assistantText }); } + #recordTerminalFact(turnId: string, status: CapabilityLiveTurnResult["status"]): void { + const existing = this.#terminalTurns.find((terminal) => terminal.turnId === turnId); + if (existing !== undefined) { + if (existing.status !== status) throw new Error("capability_live_terminal_fact_conflict"); + return; + } + this.#terminalTurns.push({ + turnId, + status, + reconciledByAttemptId: this.#currentAttemptId, + observedAt: this.#now().toISOString(), + }); + } + #rejectTurn(error: unknown): void { const waiter = this.#turnWaiter; if (waiter === null) return; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 5af3d6bde6e..b794807935e 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -67,7 +67,7 @@ export function createCapabilityRunnerdCodexTransport( const diagnostic = (message: string) => { evidence.diagnostics.push(message); if (evidence.diagnostics.length > 64) evidence.diagnostics.shift(); - const match = /capability codex proxy started runner_pid=(\d+) codex_pid=(\d+)/.exec(message); + const match = /(?:capability|native) codex proxy started runner_pid=(\d+) codex_pid=(\d+)/.exec(message); if (match) { evidence.runnerPid = Number(match[1]); evidence.codexPid = Number(match[2]); diff --git a/packages/paperclip-runner/src/testing.ts b/packages/paperclip-runner/src/testing.ts index 71198fbe9a2..7ac52f9a48d 100644 --- a/packages/paperclip-runner/src/testing.ts +++ b/packages/paperclip-runner/src/testing.ts @@ -16,3 +16,4 @@ export * from "./mock-core/capability-control-plane-types.js"; export * from "./mock-core/capability-mock-control-plane-adapter.js"; export * from "./protocol/conformance-fixture.js"; export * from "./live/live-session.js"; +export * from "./live/durable-live-session-store.js"; diff --git a/packages/paperclip-runner/test/fixtures/fake-durable-codex-app-server.mjs b/packages/paperclip-runner/test/fixtures/fake-durable-codex-app-server.mjs new file mode 100644 index 00000000000..f343e40ad16 --- /dev/null +++ b/packages/paperclip-runner/test/fixtures/fake-durable-codex-app-server.mjs @@ -0,0 +1,163 @@ +#!/usr/bin/env node +import { readFileSync, writeFileSync } from "node:fs"; + +const statePath = process.argv[2]; +if (!statePath) throw new Error("state path is required"); + +function load() { + try { + return JSON.parse(readFileSync(statePath, "utf8")); + } catch (error) { + if (error?.code !== "ENOENT") throw error; + return { + threadId: "thread-durable-runnerd", + sessionId: "session-durable-runnerd", + nextTurn: 0, + heldOnce: false, + turns: {}, + }; + } +} + +let state = load(); +const pendingTools = new Map(); +let buffer = ""; + +function save() { + writeFileSync(statePath, `${JSON.stringify(state)}\n`, { mode: 0o600 }); +} + +function send(value) { + process.stdout.write(`${JSON.stringify(value)}\n`); +} + +function finishTurn(turnId, toolResult) { + send({ + method: "item/completed", + params: { + threadId: state.threadId, + turnId, + item: { id: `message-${turnId}`, type: "agentMessage", text: `Tool result: ${JSON.stringify(toolResult)}` }, + }, + }); + state.turns[turnId] = "completed"; + save(); + send({ + method: "turn/completed", + params: { threadId: state.threadId, turn: { id: turnId, status: "completed" } }, + }); +} + +function handleRequest(message) { + const { id, method, params = {} } = message; + if (method === "initialize") { + send({ id, result: { user: { sessionId: state.sessionId } } }); + return; + } + if (method === "thread/start") { + save(); + send({ + id, + result: { + model: "gpt-fixture", + modelProvider: "openai-fixture", + thread: { id: state.threadId, sessionId: state.sessionId }, + }, + }); + return; + } + if (method === "thread/read") { + send({ + id, + result: { + thread: { + id: state.threadId, + sessionId: state.sessionId, + turns: Object.entries(state.turns).map(([turnId, status]) => ({ id: turnId, status })), + }, + }, + }); + return; + } + if (method === "thread/resume") { + send({ + id, + result: { + model: "gpt-fixture", + modelProvider: "openai-fixture", + thread: { id: state.threadId, sessionId: state.sessionId }, + }, + }); + return; + } + if (method === "turn/start") { + const turnId = `turn-${++state.nextTurn}`; + state.turns[turnId] = "inProgress"; + save(); + send({ id, result: { turn: { id: turnId, status: "inProgress" } } }); + send({ + method: "turn/started", + params: { threadId: state.threadId, turn: { id: turnId, status: "inProgress" } }, + }); + const toolRequestId = `tool-${turnId}`; + pendingTools.set(toolRequestId, turnId); + send({ + id: toolRequestId, + method: "item/tool/call", + params: { + threadId: state.threadId, + turnId, + callId: `governed-idempotent-call-${turnId}`, + tool: "report_progress", + arguments: { + idempotencyKey: "governed-idempotency-key", + body: "One durable governed effect.", + }, + }, + }); + return; + } + if (method === "turn/interrupt") { + const turnId = String(params.turnId); + state.turns[turnId] = "interrupted"; + save(); + send({ id, result: {} }); + send({ + method: "turn/completed", + params: { threadId: state.threadId, turn: { id: turnId, status: "interrupted" } }, + }); + return; + } + send({ id, error: { code: -32601, message: `unsupported method ${method}` } }); +} + +function handle(message) { + if (message.method) { + if (message.id !== undefined) handleRequest(message); + return; + } + const turnId = pendingTools.get(String(message.id)); + if (!turnId) return; + pendingTools.delete(String(message.id)); + if (!state.heldOnce) { + state.heldOnce = true; + save(); + return; + } + finishTurn(turnId, message.result); +} + +process.stdin.setEncoding("utf8"); +process.stdin.resume(); +process.stdin.on("data", (chunk) => { + buffer += chunk; + for (;;) { + const newline = buffer.indexOf("\n"); + if (newline < 0) break; + const line = buffer.slice(0, newline).trim(); + buffer = buffer.slice(newline + 1); + if (line) { + handle(JSON.parse(line)); + } + } +});