prototype: prove Pi runsc transport continuity

2026-08-22 23:04:23 -04:00
parent a69dca2907
commit 5c79a7ec9b
9 changed files with 1095 additions and 0 deletions
@@ -0,0 +1 @@
+.prototype-state/
@@ -0,0 +1,83 @@
+# Active-turn continuity under runsc — throwaway prototype
+
+> **PROTOTYPE: do not ship or import this code.** It answers whether a supervisor-owned bridge can keep one Pi RPC process and its active work alive while the Paseo-side daemon client crashes and reconnects.
+
+## Question
+
+Can a Pi process running inside gVisor continue a long-running RPC operation across a Paseo daemon crash, replay missed output without duplicate timeline delivery, accept cancellation from the replacement daemon, and clean up its runsc container afterward? This establishes the transport precondition for active-turn continuity; it does not execute an authenticated model turn.
+
+## Shape
+
+```text
+daemon client generation 1 ─┐
+                             ├─ Unix JSONL bridge ─ retained pipes ─ runsc ─ Pi RPC
+replacement generation 2 ───┘         │
+                                      └─ sequenced bounded replay buffer
+```
+
+The bridge is the throwaway stand-in for the proposed root-owned supervisor. It owns `runsc` and Pi's stdin/stdout for the complete runtime lifetime. Daemon clients receive a fenced writer lease and acknowledge bridge sequence numbers.
+
+The demo:
+
+1. builds a disposable Node/Pi rootfs with Docker;
+2. launches Pi 0.84.2 inside rootless runsc with an OCI bundle;
+3. proves a second live writer is rejected;
+4. starts a long-running real Pi RPC `bash` command;
+5. kills the first daemon client with `SIGKILL` after receiving—but before acknowledging—a streamed frame;
+6. reconnects a replacement client using the last durable cursor;
+7. proves the bridge and runsc state identities are unchanged;
+8. replays the missed frame and suppresses its duplicate timeline delivery by bridge sequence;
+9. sends `abort_bash` through the replacement connection and observes the original command cancel;
+10. confirms the Pi session identity is unchanged; and
+11. explicitly closes Pi and verifies the runsc container state is removed.
+
+This is not the production supervisor. It uses rootless runsc, an in-memory replay buffer, one process, one host user, and no network. It does not prove supervisor-crash recovery, security policy, durable spool design, resource limits, or the official Docker topology.
+
+## Prerequisites
+
+- Linux with unprivileged user namespaces enabled
+- Docker access for building/exporting the disposable rootfs
+- Node 22+
+- an official `runsc` binary
+
+Download and verify the current official x86-64 binary outside the repository:
+
+```bash
+mkdir -p /tmp/paseo-runsc-bin
+cd /tmp/paseo-runsc-bin
+curl -fLO https://storage.googleapis.com/gvisor/releases/release/latest/x86_64/runsc
+curl -fLO https://storage.googleapis.com/gvisor/releases/release/latest/x86_64/runsc.sha512
+sha512sum -c runsc.sha512
+chmod +x runsc
+```
+
+## Run
+
+```bash
+packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/run.sh
+```
+
+Override locations when needed:
+
+```bash
+RUNSC=/path/to/runsc \
+PASEO_RUNSC_PROTOTYPE_STATE=/tmp/paseo-active-turn-state \
+packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/run.sh
+```
+
+Generated rootfs, bundle, logs, cursors, and results live in `.prototype-state/` beside the scripts and are ignored by the prototype branch.
+
+## Success evidence
+
+The command exits zero and prints a final JSON verdict containing:
+
+- `verdict: "pi_rpc_transport_continuity_proved"`
+- `sameRuntime: true`
+- `writerFencePassed: true`
+- `sawReplayedTick: true`
+- `duplicateReplaySuppressed` greater than zero
+- `bashCancelled: true`
+- `piSessionStable: true`
+- `runscContainerRemoved: true`
+
+The result also records bridge PID, runsc process PID, runsc state PID, container ID, generation, cursor, and buffer state so continuity is not inferred from timeline completion alone.
@@ -0,0 +1,46 @@
+# Prototype result
+
+## Verdict
+
+A supervisor-owned reconnectable bridge can preserve Pi RPC transport across a Paseo daemon crash while Pi runs inside gVisor.
+
+The prototype used a real Pi 0.84.2 process inside `runsc release-20260817.0`. It did not execute an authenticated model turn. The human accepted this as sufficient to settle bridge feasibility; authenticated model-turn and extension-UI behavior remain required in the later proof-of-concept verification matrix.
+
+## Observed demonstration
+
+The first daemon client launched a long-running Pi RPC `bash` command and was killed with `SIGKILL` after receiving bridge sequence 5 without acknowledging it. The replacement client attached at its last durable cursor.
+
+The run reported:
+
+```json
+{
+  "verdict": "pi_rpc_transport_continuity_proved",
+  "sameRuntime": true,
+  "writerFencePassed": true,
+  "sawReplayedTick": true,
+  "duplicateReplaySuppressed": 1,
+  "bashCancelled": true,
+  "piSessionStable": true,
+  "runscContainerRemoved": true
+}
+```
+
+The bridge PID, foreground `runsc` PID, runsc state PID, container ID, and Pi session ID were unchanged across client generations. The replacement client replayed the unacknowledged frame, suppressed its duplicate delivery by bridge sequence, sent `abort_bash` to the original Pi process, observed cancellation, and explicitly closed the runtime. A final `runsc state` lookup confirmed removal.
+
+## What this settles
+
+- Pi's pipes can live below the daemon in a longer-lived supervisor bridge.
+- A bridge-owned sequence/cursor protocol can provide replay and downstream duplicate suppression.
+- Exclusive writer leases can fence concurrent or stale daemon generations.
+- A replacement daemon can send control commands to the original Pi process after reattachment.
+- Explicit close can cleanly terminate Pi and remove its runsc container.
+
+## What remains open
+
+- authenticated model-turn and tool-call continuity;
+- pending extension UI request replay and response fencing;
+- durable on-disk spooling and overflow behavior;
+- supervisor crash and restart;
+- root-owned supervisor authorization and path validation;
+- non-rootless runsc, network policy, credentials, resource limits, and both supported daemon topologies;
+- integration with Paseo's real `AgentManager`, timeline reducer, cancellation, and persistence contracts.
@@ -0,0 +1,290 @@
+#!/usr/bin/env node
+
+// PROTOTYPE ONLY. This process stands in for the root-owned sandbox supervisor.
+// It intentionally owns runsc and Pi's pipes instead of the daemon client.
+
+import { execFileSync, spawn } from "node:child_process";
+import { randomUUID } from "node:crypto";
+import { existsSync, unlinkSync } from "node:fs";
+import { createServer } from "node:net";
+import { resolve } from "node:path";
+import { createInterface } from "node:readline";
+
+const options = parseArgs(process.argv.slice(2));
+const runsc = resolve(options.runsc);
+const bundle = resolve(options.bundle);
+const runtimeRoot = resolve(options.runtimeRoot);
+const socketPath = resolve(options.socket);
+const containerId = options.containerId;
+const maxFrames = Number(options.maxFrames ?? 10_000);
+
+if (existsSync(socketPath)) unlinkSync(socketPath);
+
+let nextSequence = 1;
+let lostBefore = 0;
+let generation = 0;
+let activeWriter = null;
+let closing = false;
+let runtimeExit = null;
+const frames = [];
+const clients = new Set();
+
+const server = createServer((socket) => {
+  clients.add(socket);
+  socket.setEncoding("utf8");
+  socket.bridge = { lease: null, generation: null, acknowledged: 0 };
+
+  let pending = "";
+  socket.on("data", (chunk) => {
+    pending += chunk;
+    while (true) {
+      const newline = pending.indexOf("\n");
+      if (newline < 0) break;
+      const line = pending.slice(0, newline);
+      pending = pending.slice(newline + 1);
+      if (!line.trim()) continue;
+      try {
+        handleMessage(socket, JSON.parse(line));
+      } catch (error) {
+        send(socket, { type: "error", code: "invalid_message", message: String(error) });
+      }
+    }
+  });
+
+  socket.on("close", () => {
+    clients.delete(socket);
+    if (activeWriter === socket) activeWriter = null;
+  });
+  socket.on("error", () => {});
+});
+
+server.listen(socketPath, () => {
+  process.stderr.write(
+    `${JSON.stringify({ event: "bridge_listening", socketPath, bridgePid: process.pid })}\n`,
+  );
+});
+
+const runtime = spawn(
+  runsc,
+  [`--root=${runtimeRoot}`, "--rootless=true", "--network=none", "run", containerId],
+  {
+    cwd: bundle,
+    stdio: ["pipe", "pipe", "pipe"],
+  },
+);
+
+publish("bridge", {
+  type: "runtime_started",
+  bridgePid: process.pid,
+  runscPid: runtime.pid,
+  containerId,
+});
+
+createInterface({ input: runtime.stdout }).on("line", (line) => {
+  let payload;
+  try {
+    payload = JSON.parse(line);
+  } catch {
+    payload = { type: "unparsed_stdout", line };
+  }
+  publish("stdout", payload);
+});
+
+createInterface({ input: runtime.stderr }).on("line", (line) => {
+  publish("stderr", { type: "runtime_stderr", line });
+});
+
+runtime.on("exit", (code, signal) => {
+  runtimeExit = { code, signal };
+  publish("bridge", { type: "runtime_exit", code, signal, containerId });
+  setTimeout(() => {
+    for (const client of clients) client.end();
+    server.close(() => {
+      if (existsSync(socketPath)) unlinkSync(socketPath);
+      process.exit(code === 0 || closing ? 0 : 1);
+    });
+  }, 250).unref();
+});
+
+runtime.on("error", (error) => {
+  publish("bridge", { type: "runtime_spawn_error", message: String(error) });
+});
+
+process.on("SIGTERM", () => closeRuntime("SIGTERM"));
+process.on("SIGINT", () => closeRuntime("SIGINT"));
+
+const messageHandlers = {
+  attach: handleAttach,
+  command: handleCommand,
+  ack: handleAck,
+  inspect: handleInspect,
+  close: handleClose,
+};
+
+function handleMessage(socket, message) {
+  const handler = messageHandlers[message.type];
+  if (!handler) {
+    send(socket, { type: "error", code: "unknown_message", messageType: message.type });
+    return;
+  }
+  handler(socket, message);
+}
+
+function handleAttach(socket, message) {
+  const cursor = Number(message.cursor ?? 0);
+  if (!Number.isSafeInteger(cursor) || cursor < 0) {
+    send(socket, { type: "error", code: "invalid_cursor" });
+    return;
+  }
+  if (cursor < lostBefore) {
+    send(socket, { type: "error", code: "replay_gap", lostBefore });
+    return;
+  }
+  if (activeWriter && activeWriter !== socket) {
+    send(socket, { type: "error", code: "writer_already_attached" });
+    return;
+  }
+  generation += 1;
+  const lease = `generation-${generation}-${randomUUID()}`;
+  socket.bridge = { lease, generation, acknowledged: cursor };
+  activeWriter = socket;
+  send(socket, {
+    type: "attached",
+    lease,
+    generation,
+    cursor,
+    nextSequence,
+    runtimeExit,
+  });
+  for (const frame of frames) {
+    if (frame.sequence > cursor) send(socket, frame);
+  }
+}
+
+function handleCommand(socket, message) {
+  if (!ownsWriterLease(socket, message.lease)) return;
+  if (runtimeExit || closing || !runtime.stdin.writable) {
+    send(socket, { type: "error", code: "runtime_not_writable" });
+    return;
+  }
+  const payload = structuredClone(message.payload);
+  let effectiveId = null;
+  if (payload && typeof payload.id === "string") {
+    effectiveId = `g${socket.bridge.generation}:${payload.id}`;
+    payload.id = effectiveId;
+  }
+  runtime.stdin.write(`${JSON.stringify(payload)}\n`);
+  send(socket, {
+    type: "command_accepted",
+    requestId: message.requestId ?? null,
+    effectiveId,
+    generation: socket.bridge.generation,
+  });
+}
+
+function handleAck(socket, message) {
+  if (!ownsWriterLease(socket, message.lease)) return;
+  const sequence = Number(message.sequence);
+  if (!Number.isSafeInteger(sequence) || sequence < socket.bridge.acknowledged) {
+    send(socket, { type: "error", code: "invalid_ack" });
+    return;
+  }
+  socket.bridge.acknowledged = sequence;
+  send(socket, { type: "acknowledged", sequence });
+}
+
+function handleInspect(socket, message) {
+  if (!ownsWriterLease(socket, message.lease)) return;
+  send(socket, {
+    type: "inspection",
+    bridgePid: process.pid,
+    runscPid: runtime.pid,
+    containerId,
+    generation: socket.bridge.generation,
+    nextSequence,
+    bufferedFrames: frames.length,
+    lostBefore,
+    runtimeExit,
+    runscState: inspectRunsc(),
+  });
+}
+
+function handleClose(socket, message) {
+  if (!ownsWriterLease(socket, message.lease)) return;
+  send(socket, { type: "closing" });
+  closeRuntime("client_close");
+}
+
+function ownsWriterLease(socket, lease) {
+  if (activeWriter !== socket || socket.bridge.lease !== lease) {
+    send(socket, { type: "error", code: "stale_writer" });
+    return false;
+  }
+  return true;
+}
+
+function publish(channel, payload) {
+  const frame = {
+    type: "frame",
+    sequence: nextSequence++,
+    channel,
+    payload,
+  };
+  frames.push(frame);
+  if (frames.length > maxFrames) {
+    const removed = frames.shift();
+    lostBefore = removed.sequence;
+  }
+  for (const client of clients) {
+    if (client.bridge?.lease) send(client, frame);
+  }
+}
+
+function send(socket, message) {
+  if (!socket.destroyed) socket.write(`${JSON.stringify(message)}\n`);
+}
+
+function inspectRunsc() {
+  try {
+    return JSON.parse(
+      execFileSync(runsc, [`--root=${runtimeRoot}`, "state", containerId], {
+        encoding: "utf8",
+        stdio: ["ignore", "pipe", "pipe"],
+      }),
+    );
+  } catch (error) {
+    return { error: error.stderr?.toString().trim() || String(error) };
+  }
+}
+
+function closeRuntime(reason) {
+  if (closing) return;
+  closing = true;
+  publish("bridge", { type: "runtime_closing", reason });
+  runtime.stdin.end();
+  const escalation = setTimeout(() => {
+    try {
+      execFileSync(runsc, [`--root=${runtimeRoot}`, "kill", containerId, "KILL"], {
+        stdio: "ignore",
+      });
+    } catch {}
+    runtime.kill("SIGKILL");
+  }, 5_000);
+  escalation.unref();
+}
+
+function parseArgs(args) {
+  const parsed = {};
+  for (let index = 0; index < args.length; index += 2) {
+    const key = args[index];
+    const value = args[index + 1];
+    if (!key?.startsWith("--") || value === undefined) {
+      throw new Error(`invalid argument near ${key ?? "<end>"}`);
+    }
+    parsed[key.slice(2)] = value;
+  }
+  for (const required of ["runsc", "bundle", "runtimeRoot", "socket", "containerId"]) {
+    if (!parsed[required]) throw new Error(`missing --${required}`);
+  }
+  return parsed;
+}
@@ -0,0 +1,363 @@
+#!/usr/bin/env node
+
+// PROTOTYPE ONLY. Simulates the daemon-side owner before and after a crash.
+
+import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs";
+import { connect } from "node:net";
+import { resolve } from "node:path";
+
+const [mode, socketInput, stateInput] = process.argv.slice(2);
+if (!mode || !socketInput || !stateInput) {
+  process.stderr.write("usage: client.mjs <first|second> <socket> <state-dir>\n");
+  process.exit(2);
+}
+
+const socketPath = resolve(socketInput);
+const stateDir = resolve(stateInput);
+const cursorPath = resolve(stateDir, "cursor.json");
+const observedPath = resolve(stateDir, "observed.jsonl");
+const baselinePath = resolve(stateDir, "baseline.json");
+mkdirSync(stateDir, { recursive: true });
+
+async function runFirstClient() {
+  writeFileSync(cursorPath, `${JSON.stringify({ sequence: 0 })}\n`);
+  writeFileSync(observedPath, "");
+
+  const connection = await BridgeConnection.open(socketPath);
+  const attached = await connection.attach(0);
+  const fencePassed = await proveWriterFence();
+
+  connection.send({ type: "inspect", lease: attached.lease });
+  const inspection = await connection.waitFor((message) => message.type === "inspection");
+
+  connection.send({
+    type: "command",
+    lease: attached.lease,
+    requestId: "state-before",
+    payload: { id: "state-before", type: "get_state" },
+  });
+  const stateFrame = await consumeUntil(
+    connection,
+    attached.lease,
+    (frame) => frame.channel === "stdout" && frame.payload?.command === "get_state",
+  );
+
+  const baseline = {
+    bridgePid: inspection.bridgePid,
+    runscPid: inspection.runscPid,
+    runscStatePid: inspection.runscState?.pid,
+    containerId: inspection.containerId,
+    piSessionId:
+      stateFrame.payload?.data?.sessionId ?? stateFrame.payload?.data?.session_id ?? null,
+    writerFencePassed: fencePassed,
+  };
+  writeFileSync(baselinePath, `${JSON.stringify(baseline, null, 2)}\n`);
+
+  connection.send({
+    type: "command",
+    lease: attached.lease,
+    requestId: "long-bash",
+    payload: {
+      id: "long-bash",
+      type: "bash",
+      command:
+        'echo "shell_pid=$$"; i=1; while [ "$i" -le 30 ]; do echo "tick-$i"; i=$((i+1)); sleep 0.4; done',
+    },
+  });
+
+  while (true) {
+    const message = await connection.next();
+    if (message.type !== "frame") continue;
+    recordObserved(message);
+    if (JSON.stringify(message.payload).includes("tick-3")) {
+      process.stdout.write(
+        `${JSON.stringify({
+          event: "simulated_daemon_crash",
+          sequenceDeliveredButNotAcknowledged: message.sequence,
+          baseline,
+        })}\n`,
+      );
+      process.kill(process.pid, "SIGKILL");
+    }
+    acknowledge(connection, attached.lease, message.sequence);
+  }
+}
+
+async function runSecondClient() {
+  const baseline = parseJson(readFileSync(baselinePath, "utf8"), baselinePath);
+  const cursor = parseJson(readFileSync(cursorPath, "utf8"), cursorPath).sequence;
+  const seen = loadObservedSequences();
+  const connection = await BridgeConnection.open(socketPath);
+  const attached = await connection.attach(cursor);
+  const inspection = await inspectRuntime(connection, attached.lease);
+  assertSameRuntime(baseline, inspection);
+
+  const progress = await continueInterruptedCommand(connection, attached.lease, seen);
+  const piSessionIdAfter = readSessionId(progress.stateAfter);
+  validateContinuity(baseline, piSessionIdAfter, progress);
+
+  const result = {
+    sameRuntime: true,
+    writerFencePassed: baseline.writerFencePassed,
+    sawReplayedTick: progress.sawReplayedTick,
+    duplicateReplaySuppressed: progress.duplicateReplaySuppressed,
+    bashCancelled: progress.bashCancelled,
+    piSessionStable: !baseline.piSessionId || piSessionIdAfter === baseline.piSessionId,
+    baseline,
+    inspection,
+  };
+  writeFileSync(resolve(stateDir, "result.json"), `${JSON.stringify(result, null, 2)}\n`);
+  process.stdout.write(`${JSON.stringify({ event: "continuity_proved", ...result })}\n`);
+
+  connection.send({ type: "close", lease: attached.lease });
+  await connection.waitFor(isRuntimeExitFrame, 15_000);
+  connection.socket.end();
+}
+
+async function inspectRuntime(connection, lease) {
+  connection.send({ type: "inspect", lease });
+  return connection.waitFor((message) => message.type === "inspection");
+}
+
+function assertSameRuntime(baseline, inspection) {
+  const sameRuntime =
+    inspection.bridgePid === baseline.bridgePid &&
+    inspection.runscPid === baseline.runscPid &&
+    inspection.runscState?.pid === baseline.runscStatePid &&
+    inspection.containerId === baseline.containerId;
+  if (!sameRuntime) {
+    throw new Error(
+      `runtime changed across reconnect: ${JSON.stringify({ baseline, inspection })}`,
+    );
+  }
+}
+
+async function continueInterruptedCommand(connection, lease, seen) {
+  const progress = {
+    duplicateReplaySuppressed: 0,
+    sawReplayedTick: false,
+    bashCancelled: false,
+    abortSent: false,
+    stateRequested: false,
+    stateAfter: null,
+  };
+  while (!progress.stateAfter) {
+    const message = await connection.next();
+    if (message.type !== "frame") continue;
+    observeFrame(connection, lease, seen, message, progress);
+    advanceRecovery(connection, lease, message, progress);
+  }
+  return progress;
+}
+
+function observeFrame(connection, lease, seen, message, progress) {
+  if (seen.has(message.sequence)) progress.duplicateReplaySuppressed += 1;
+  else {
+    seen.add(message.sequence);
+    recordObserved(message);
+  }
+  acknowledge(connection, lease, message.sequence);
+  if (JSON.stringify(message.payload).includes("tick-3")) progress.sawReplayedTick = true;
+}
+
+function advanceRecovery(connection, lease, message, progress) {
+  if (progress.sawReplayedTick && !progress.abortSent) {
+    progress.abortSent = true;
+    connection.send({
+      type: "command",
+      lease,
+      requestId: "abort-after-reconnect",
+      payload: { type: "abort_bash" },
+    });
+  }
+  if (message.payload?.command === "bash" && message.payload?.data?.cancelled === true) {
+    progress.bashCancelled = true;
+  }
+  if (progress.bashCancelled && !progress.stateRequested) {
+    progress.stateRequested = true;
+    connection.send({
+      type: "command",
+      lease,
+      requestId: "state-after",
+      payload: { id: "state-after", type: "get_state" },
+    });
+  }
+  if (progress.stateRequested && message.payload?.command === "get_state") {
+    progress.stateAfter = message.payload;
+  }
+}
+
+function readSessionId(state) {
+  return state?.data?.sessionId ?? state?.data?.session_id ?? null;
+}
+
+function validateContinuity(baseline, piSessionIdAfter, progress) {
+  if (baseline.piSessionId && piSessionIdAfter !== baseline.piSessionId) {
+    throw new Error(`Pi session changed: ${baseline.piSessionId} -> ${piSessionIdAfter}`);
+  }
+  if (!baseline.writerFencePassed) throw new Error("duplicate writer was not fenced");
+  if (!progress.sawReplayedTick) throw new Error("missed frame was not replayed");
+  if (!progress.duplicateReplaySuppressed) {
+    throw new Error("duplicate replay suppression was not exercised");
+  }
+  if (!progress.bashCancelled) throw new Error("bash command was not cancelled after reconnect");
+}
+
+function isRuntimeExitFrame(message) {
+  return (
+    message.type === "frame" &&
+    message.channel === "bridge" &&
+    message.payload?.type === "runtime_exit"
+  );
+}
+
+async function proveWriterFence() {
+  const second = await BridgeConnection.open(socketPath);
+  second.send({ type: "attach", cursor: 0 });
+  const response = await second.waitFor(
+    (message) => message.type === "error" || message.type === "attached",
+  );
+  second.socket.end();
+  return response.type === "error" && response.code === "writer_already_attached";
+}
+
+async function consumeUntil(connection, lease, predicate) {
+  while (true) {
+    const message = await connection.next();
+    if (message.type !== "frame") continue;
+    recordObserved(message);
+    acknowledge(connection, lease, message.sequence);
+    if (predicate(message)) return message;
+  }
+}
+
+function acknowledge(connection, lease, sequence) {
+  writeFileSync(cursorPath, `${JSON.stringify({ sequence })}\n`);
+  connection.send({ type: "ack", lease, sequence });
+}
+
+function recordObserved(frame) {
+  appendFileSync(observedPath, `${JSON.stringify(frame)}\n`);
+}
+
+function loadObservedSequences() {
+  if (!existsSync(observedPath)) return new Set();
+  return new Set(
+    readFileSync(observedPath, "utf8")
+      .split("\n")
+      .filter(Boolean)
+      .map((line) => parseJson(line, observedPath).sequence),
+  );
+}
+
+function parseJson(text, source) {
+  try {
+    return JSON.parse(text);
+  } catch (error) {
+    throw new Error(`invalid JSON from ${source}`, { cause: error });
+  }
+}
+
+class BridgeConnection {
+  static async open(path) {
+    const socket = connect(path);
+    await new Promise((resolvePromise, reject) => {
+      socket.once("connect", resolvePromise);
+      socket.once("error", reject);
+    });
+    return new BridgeConnection(socket);
+  }
+
+  constructor(socket) {
+    this.socket = socket;
+    this.queue = [];
+    this.waiters = [];
+    this.pending = "";
+    socket.setEncoding("utf8");
+    socket.on("data", (chunk) => this.#onData(chunk));
+    socket.on("error", (error) => this.#fail(error));
+    socket.on("close", () => this.#fail(new Error("bridge connection closed")));
+  }
+
+  send(message) {
+    this.socket.write(`${JSON.stringify(message)}\n`);
+  }
+
+  async attach(cursor) {
+    this.send({ type: "attach", cursor });
+    const response = await this.waitFor(
+      (message) => message.type === "attached" || message.type === "error",
+    );
+    if (response.type === "error") throw new Error(`attach failed: ${JSON.stringify(response)}`);
+    return response;
+  }
+
+  next(timeoutMs = 10_000) {
+    if (this.queue.length) return Promise.resolve(this.queue.shift());
+    return new Promise((resolvePromise, reject) => {
+      const timer = setTimeout(
+        () => reject(new Error("timed out waiting for bridge message")),
+        timeoutMs,
+      );
+      this.waiters.push({
+        resolve: (value) => {
+          clearTimeout(timer);
+          resolvePromise(value);
+        },
+        reject: (error) => {
+          clearTimeout(timer);
+          reject(error);
+        },
+      });
+    });
+  }
+
+  async waitFor(predicate, timeoutMs = 10_000) {
+    const deadline = Date.now() + timeoutMs;
+    const skipped = [];
+    try {
+      while (Date.now() < deadline) {
+        const message = await this.next(Math.max(1, deadline - Date.now()));
+        if (predicate(message)) {
+          this.queue.unshift(...skipped);
+          return message;
+        }
+        skipped.push(message);
+      }
+      throw new Error("timed out waiting for matching bridge message");
+    } catch (error) {
+      this.queue.unshift(...skipped);
+      throw error;
+    }
+  }
+
+  #onData(chunk) {
+    this.pending += chunk;
+    while (true) {
+      const newline = this.pending.indexOf("\n");
+      if (newline < 0) return;
+      const line = this.pending.slice(0, newline);
+      this.pending = this.pending.slice(newline + 1);
+      if (!line.trim()) continue;
+      let message;
+      try {
+        message = JSON.parse(line);
+      } catch (error) {
+        this.#fail(new Error("bridge returned invalid JSON", { cause: error }));
+        this.socket.destroy();
+        return;
+      }
+      const waiter = this.waiters.shift();
+      if (waiter) waiter.resolve(message);
+      else this.queue.push(message);
+    }
+  }
+
+  #fail(error) {
+    for (const waiter of this.waiters.splice(0)) waiter.reject(error);
+  }
+}
+
+if (mode === "first") await runFirstClient();
+else if (mode === "second") await runSecondClient();
+else throw new Error(`unknown client mode: ${mode}`);
@@ -0,0 +1,163 @@
+#!/usr/bin/env node
+
+// PROTOTYPE ONLY. Runs an end-to-end crash/reconnect demonstration with Pi in runsc.
+
+import { execFileSync, spawn } from "node:child_process";
+import { closeSync, existsSync, mkdirSync, openSync, readFileSync, rmSync } from "node:fs";
+import { dirname, resolve } from "node:path";
+import { fileURLToPath } from "node:url";
+
+const scriptDir = dirname(fileURLToPath(import.meta.url));
+const stateDir = resolve(
+  process.env.PASEO_RUNSC_PROTOTYPE_STATE ?? resolve(scriptDir, ".prototype-state"),
+);
+const runsc = resolve(process.env.RUNSC ?? "/tmp/paseo-runsc-bin/runsc");
+const bundle = resolve(stateDir, "bundle");
+const rootfs = resolve(bundle, "rootfs");
+const workspace = resolve(stateDir, "workspace");
+const piState = resolve(stateDir, "pi-state");
+const clientState = resolve(stateDir, "client-state");
+const runtimeRoot = resolve(stateDir, "runsc-root");
+// Linux pathname Unix sockets are limited to roughly 108 bytes. Keep this outside the deep source path.
+const socketPath = `/tmp/paseo-pi-bridge-${process.pid}.sock`;
+const containerId = `paseo-pi-prototype-${process.pid}`;
+const bridgeLog = resolve(stateDir, "bridge.log");
+
+if (!existsSync(runsc)) throw new Error(`runsc not found: ${runsc}`);
+if (!existsSync(resolve(rootfs, "usr/local/bin/pi"))) {
+  throw new Error(`prototype rootfs not prepared: run ${resolve(scriptDir, "prepare-rootfs.sh")}`);
+}
+
+for (const path of [clientState, runtimeRoot, workspace, piState])
+  rmSync(path, { recursive: true, force: true });
+rmSync(socketPath, { force: true });
+rmSync(resolve(bundle, "config.json"), { force: true });
+for (const path of [clientState, runtimeRoot, workspace, piState])
+  mkdirSync(path, { recursive: true });
+
+execFileSync(
+  process.execPath,
+  [resolve(scriptDir, "generate-bundle.mjs"), runsc, bundle, workspace, piState],
+  { stdio: "inherit" },
+);
+
+const logFd = openSync(bridgeLog, "w");
+const bridge = spawn(
+  process.execPath,
+  [
+    resolve(scriptDir, "bridge.mjs"),
+    "--runsc",
+    runsc,
+    "--bundle",
+    bundle,
+    "--runtimeRoot",
+    runtimeRoot,
+    "--socket",
+    socketPath,
+    "--containerId",
+    containerId,
+  ],
+  { stdio: ["ignore", logFd, logFd], detached: false },
+);
+closeSync(logFd);
+
+try {
+  await waitForPath(socketPath, 10_000);
+
+  process.stdout.write("\n=== daemon generation one: start long Pi RPC command ===\n");
+  const first = spawn(
+    process.execPath,
+    [resolve(scriptDir, "client.mjs"), "first", socketPath, clientState],
+    { stdio: "inherit" },
+  );
+  const firstExit = await waitForExit(first, 30_000);
+  if (firstExit.signal !== "SIGKILL") {
+    throw new Error(`first daemon did not crash as expected: ${JSON.stringify(firstExit)}`);
+  }
+  if (bridge.exitCode !== null) throw new Error("bridge exited with the first daemon client");
+
+  process.stdout.write("\n=== daemon generation two: attach, replay, cancel, close ===\n");
+  const second = spawn(
+    process.execPath,
+    [resolve(scriptDir, "client.mjs"), "second", socketPath, clientState],
+    { stdio: "inherit" },
+  );
+  const secondExit = await waitForExit(second, 30_000);
+  if (secondExit.code !== 0) {
+    throw new Error(`replacement daemon failed: ${JSON.stringify(secondExit)}`);
+  }
+
+  const bridgeExit = await waitForExit(bridge, 15_000);
+  if (bridgeExit.code !== 0) {
+    throw new Error(`bridge did not cleanly exit: ${JSON.stringify(bridgeExit)}`);
+  }
+
+  let stateRemoved = false;
+  try {
+    execFileSync(runsc, [`--root=${runtimeRoot}`, "state", containerId], { stdio: "ignore" });
+  } catch {
+    stateRemoved = true;
+  }
+  if (!stateRemoved) throw new Error("runsc container state remained after explicit close");
+
+  let result;
+  try {
+    result = JSON.parse(readFileSync(resolve(clientState, "result.json"), "utf8"));
+  } catch (error) {
+    throw new Error("replacement daemon wrote an invalid result", { cause: error });
+  }
+  const summary = {
+    verdict: "pi_rpc_transport_continuity_proved",
+    runscContainerRemoved: stateRemoved,
+    bridgeLog,
+    ...result,
+  };
+  process.stdout.write("\n=== verdict ===\n");
+  process.stdout.write(`${JSON.stringify(summary, null, 2)}\n`);
+} catch (error) {
+  process.stderr.write("\n=== prototype failed ===\n");
+  process.stderr.write(`${error.stack ?? String(error)}\n`);
+  if (existsSync(bridgeLog)) process.stderr.write(readFileSync(bridgeLog, "utf8"));
+  throw error;
+} finally {
+  if (bridge.exitCode === null) bridge.kill("SIGTERM");
+  try {
+    execFileSync(runsc, [`--root=${runtimeRoot}`, "kill", containerId, "KILL"], {
+      stdio: "ignore",
+    });
+  } catch {}
+  try {
+    execFileSync(runsc, [`--root=${runtimeRoot}`, "delete", "--force", containerId], {
+      stdio: "ignore",
+    });
+  } catch {}
+}
+
+async function waitForPath(path, timeoutMs) {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (existsSync(path)) return;
+    await new Promise((resolvePromise) => setTimeout(resolvePromise, 50));
+  }
+  throw new Error(`timed out waiting for ${path}`);
+}
+
+function waitForExit(child, timeoutMs) {
+  if (child.exitCode !== null || child.signalCode !== null) {
+    return Promise.resolve({ code: child.exitCode, signal: child.signalCode });
+  }
+  return new Promise((resolvePromise, reject) => {
+    const timer = setTimeout(() => {
+      child.kill("SIGKILL");
+      reject(new Error(`timed out waiting for pid ${child.pid}`));
+    }, timeoutMs);
+    child.once("exit", (code, signal) => {
+      clearTimeout(timer);
+      resolvePromise({ code, signal });
+    });
+    child.once("error", (error) => {
+      clearTimeout(timer);
+      reject(error);
+    });
+  });
+}
@@ -0,0 +1,84 @@
+#!/usr/bin/env node
+
+// PROTOTYPE ONLY. Generates a rootless OCI bundle for one Pi RPC process.
+
+import { execFileSync } from "node:child_process";
+import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
+import { resolve } from "node:path";
+
+const [runscInput, bundleInput, workspaceInput, stateInput] = process.argv.slice(2);
+if (!runscInput || !bundleInput || !workspaceInput || !stateInput) {
+  process.stderr.write("usage: generate-bundle.mjs <runsc> <bundle> <workspace> <state>\n");
+  process.exit(2);
+}
+
+const runsc = resolve(runscInput);
+const bundle = resolve(bundleInput);
+const workspace = resolve(workspaceInput);
+const state = resolve(stateInput);
+
+mkdirSync(bundle, { recursive: true });
+mkdirSync(workspace, { recursive: true });
+mkdirSync(state, { recursive: true });
+mkdirSync(resolve(state, "home"), { recursive: true });
+mkdirSync(resolve(state, "pi-agent"), { recursive: true });
+mkdirSync(resolve(state, "pi-sessions"), { recursive: true });
+
+execFileSync(runsc, ["--rootless=true", "spec"], { cwd: bundle, stdio: "inherit" });
+
+const configPath = resolve(bundle, "config.json");
+let config;
+try {
+  config = JSON.parse(readFileSync(configPath, "utf8"));
+} catch (error) {
+  throw new Error(`runsc generated invalid OCI config at ${configPath}`, { cause: error });
+}
+config.root = { path: "rootfs", readonly: true };
+config.hostname = "paseo-pi-prototype";
+config.process.terminal = false;
+config.process.cwd = "/workspace";
+config.process.args = ["/usr/local/bin/pi", "--mode", "rpc", "--no-session"];
+config.process.env = [
+  "PATH=/usr/local/bin:/usr/bin:/bin",
+  "HOME=/state/home",
+  "PI_CODING_AGENT_DIR=/state/pi-agent",
+  "PI_CODING_AGENT_SESSION_DIR=/state/pi-sessions",
+  "PI_OFFLINE=1",
+  "PASEO_AGENT_ID=prototype-active-turn",
+  "PASEO_AGENT_CWD=/workspace",
+];
+config.process.noNewPrivileges = true;
+config.process.rlimits = [
+  { type: "RLIMIT_NOFILE", hard: 1024, soft: 1024 },
+  { type: "RLIMIT_NPROC", hard: 256, soft: 256 },
+];
+config.mounts = [
+  { destination: "/proc", type: "proc", source: "proc" },
+  {
+    destination: "/dev",
+    type: "tmpfs",
+    source: "tmpfs",
+    options: ["nosuid", "strictatime", "mode=755", "size=65536k"],
+  },
+  {
+    destination: "/tmp",
+    type: "tmpfs",
+    source: "tmpfs",
+    options: ["nosuid", "nodev", "mode=1777", "size=64m"],
+  },
+  {
+    destination: "/workspace",
+    type: "bind",
+    source: workspace,
+    options: ["rbind", "rw", "nosuid", "nodev"],
+  },
+  {
+    destination: "/state",
+    type: "bind",
+    source: state,
+    options: ["rbind", "rw", "nosuid", "nodev", "noexec"],
+  },
+];
+
+writeFileSync(configPath, `${JSON.stringify(config, null, 2)}\n`, { mode: 0o600 });
+process.stdout.write(`${configPath}\n`);
@@ -0,0 +1,46 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# PROTOTYPE ONLY. Builds an expendable rootfs for the active-turn bridge demo.
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+STATE_DIR="${PASEO_RUNSC_PROTOTYPE_STATE:-$SCRIPT_DIR/.prototype-state}"
+ROOTFS="$STATE_DIR/bundle/rootfs"
+IMAGE="paseo-pi-runsc-prototype:0.84.2"
+
+mkdir -p "$STATE_DIR/bundle"
+
+if [[ -x "$ROOTFS/usr/local/bin/pi" ]]; then
+  printf 'prototype rootfs already prepared: %s\n' "$ROOTFS"
+  exit 0
+fi
+
+printf 'building disposable Pi image %s\n' "$IMAGE"
+docker build --quiet --tag "$IMAGE" - <<'DOCKERFILE'
+FROM node:24-bookworm-slim
+RUN apt-get update \
+  && apt-get install -y --no-install-recommends bash ca-certificates git ripgrep fd-find \
+  && rm -rf /var/lib/apt/lists/* \
+  && npm install --global @earendil-works/pi-coding-agent@0.84.2 \
+  && mkdir -p /workspace /state/home /state/pi-agent /state/pi-sessions /tmp \
+  && chmod 1777 /tmp
+ENV PATH=/usr/local/bin:/usr/bin:/bin
+ENV HOME=/state/home
+ENV PI_CODING_AGENT_DIR=/state/pi-agent
+ENV PI_CODING_AGENT_SESSION_DIR=/state/pi-sessions
+ENV PI_OFFLINE=1
+WORKDIR /workspace
+DOCKERFILE
+
+container_id="$(docker create "$IMAGE")"
+cleanup() {
+  docker rm -f "$container_id" >/dev/null 2>&1 || true
+}
+trap cleanup EXIT
+
+rm -rf "$ROOTFS"
+mkdir -p "$ROOTFS"
+printf 'exporting rootfs to %s\n' "$ROOTFS"
+docker export "$container_id" | tar -C "$ROOTFS" -xf -
+
+printf 'prepared prototype rootfs: %s\n' "$ROOTFS"
@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# PROTOTYPE ONLY. Generated state stays ignored beside this script.
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+RUNSC="${RUNSC:-/tmp/paseo-runsc-bin/runsc}"
+
+if [[ ! -x "$RUNSC" ]]; then
+  cat >&2 <<EOF
+runsc was not found at $RUNSC.
+Download and verify the official binary first, then set RUNSC=/path/to/runsc.
+See README.md.
+EOF
+  exit 1
+fi
+
+"$SCRIPT_DIR/prepare-rootfs.sh"
+RUNSC="$RUNSC" node "$SCRIPT_DIR/demo.mjs"