fix(runner): resume governed live attempts durably

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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

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"

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::<u64, io::Error>(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(),

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<CapabilityLiveSessionSnapshot | null> {
+    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<void> {
+    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<void> {
+    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");
+    }
+  }
+}

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";

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<CodexRpcNotification> {
   #values: CodexRpcNotification[] = [];
@@ -46,6 +55,8 @@ interface FakeProviderState {
   nextTurn: number;
   lastToolResult: string;
   transports: FakeCapabilityCodexTransport[];
+  turns: Map<string, string>;
+  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<string, unknown> : {};
       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<Record<string, unknown>>)[0]?.text);
+      if (this.state.holdAfterTool) await new Promise<never>(() => 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,
+  );
 });

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<string>();
+  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<string>();
+  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<string>();
+  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<CapabilityLiveUsageReceipt, "receiptId" | "attemptId" | "providerResponseId" | "turnId" | "observedAt"> {
+  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<string, unknown> {
   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<CapabilityLiveSession> {
+    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<CapabilityLiveSessionSnapshot> {
     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<OpenSessionOptions, "port" | "authority" | "config"> {
+interface RestoreSessionOptions extends Omit<OpenSessionOptions, "port" | "authority" | "config" | "attemptId"> {
   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<CapabilityLiveSession> {
     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<CapabilityLiveAttemptStatus, "running" | "terminated">,
+    failureCode: string | null = null,
+  ): Promise<CapabilityLiveSessionSnapshot> {
+    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<CapabilityLiveTurnResult> {
     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<string, unknown>;
-    const terminal = new Promise<Omit<CapabilityLiveTurnResult, "snapshot">>((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<Omit<CapabilityLiveTurnResult, "snapshot">> {
+    if (this.#turnWaiter !== null) throw new Error("Capability live session already has a turn waiter");
+    return new Promise<Omit<CapabilityLiveTurnResult, "snapshot">>((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<CapabilityLiveTurnResult | null> {
+    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<CapabilityLiveSessionSnapshot> {
     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<void> {
+    await this.#disconnect(reason).catch(() => undefined);
+    await this.#port.stop().catch(() => undefined);
+  }
+
   async #handleServerRequest(request: CodexRpcServerRequest): Promise<Record<string, unknown>> {
     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;

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]);

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";

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));
+    }
+  }
+});