docs: research Pi active-turn reattachment

goals/research/pi-active-turn-reattachment.md

@@ -0,0 +1,109 @@
+# Pi active-turn reattachment
+
+Research for **Prepare agent session sandboxing with gVisor**.
+
+## Question
+
+Can a Pi turn started through `pi --mode rpc` keep running while the Paseo daemon crashes, then reattach to the replacement daemon with no lost or duplicated stream events?
+
+This research uses Pi `@earendil-works/pi-coding-agent` **0.84.2**, matching the installed `pi` binary. The primary source was cached with OpenSrc at `/srv/data/opensrc/repos/github.com/earendil-works/pi/0.84.2/packages/coding-agent`.
+
+## Verdict
+
+**Not with direct Pi RPC or native session resume. Feasible with a separate supervisor-owned reconnectable bridge.**
+
+Pi RPC is one process with one stdin command stream and one stdout event stream. It has no attach endpoint, client identity, event acknowledgment, or event replay protocol. Current Paseo owns those pipes in the daemon process and closes Pi during normal daemon shutdown. After an abrupt daemon death, Pi also observes stdin EOF and exits; even if it survived temporarily, a replacement daemon has no handle with which to adopt the pipes.
+
+A supervisor that outlives the daemon can preserve the Pi process. It must continue owning and draining Pi's pipes, add its own reconnectable IPC protocol, sequence and retain every Pi frame, fence command writers, and replay from a daemon acknowledgment cursor. This is bridge reattachment, not a Pi feature.
+
+## Three different operations
+
+| Operation                                                                                         | Result                                                                                                                                                   |
+| ------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| **Direct process attachment**: a second RPC client connects to the already-running Pi process     | **Unsupported.** Pi exposes only the process's stdin/stdout. There is no socket, attach command, or transferable runtime identifier in the RPC protocol. |
+| **Bridge reattachment**: a new daemon reconnects to a supervisor that still owns Pi and its pipes | **Feasible.** The bridge must supply discovery, writer fencing, buffering, replay, and lifecycle semantics that Pi does not supply.                      |
+| **Native session resume after a failed turn**: start a new Pi process with `--session <jsonl>`    | **Conversation recovery only.** It reconstructs persisted messages but does not continue the old provider request, tool execution, or event stream.      |
+
+## Pi RPC findings
+
+### One process owns one duplex JSONL stream
+
+Pi documents RPC mode as commands on stdin and responses/events on stdout. Its typed client implements that contract by spawning one child with `stdio: ["pipe", "pipe", "pipe"]`; it stores request promises and event listeners in the client process ([RPC documentation](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/docs/rpc.md#protocol-overview), [`rpc-client.ts:56-179`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-client.ts#L56-L179)).
+
+The RPC server reads only `process.stdin`, subscribes once to the active `AgentSession`, and writes each resulting frame immediately to `process.stdout`. It does not accept connections or model multiple clients ([`rpc-mode.ts:54-63`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L54-L63), [`rpc-mode.ts:351-362`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L351-L362), [`rpc-mode.ts:800-813`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L800-L813)). The public command union has session switching and history reads, but no attach, detach, claim, or reconnect operation ([`rpc-types.ts:19-76`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-types.ts#L19-L76)).
+
+Launching a second `pi --mode rpc --session <same-file>` creates a second runtime. It does not attach to the first. Pi's session writer appends directly to the JSONL file and has no cross-process lock in `SessionManager`, so concurrent interactive use of those runtimes is not a reattachment mechanism and is unsafe ([`session-manager.ts:1012-1068`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/core/session-manager.ts#L1012-L1068)).
+
+### Live RPC events are not replayed or acknowledged
+
+Most RPC events have no request ID. They are emitted as they happen; only direct RPC bash output carries the originating command ID. `message_update` is delta-based, and `message_end.message` is authoritative ([RPC event documentation](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/docs/rpc.md#events)). The server keeps no event sequence, subscriber cursor, acknowledgment table, or replay buffer; its output path serializes each subscribed event straight to stdout ([`rpc-mode.ts:60-62`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L60-L62), [`rpc-mode.ts:351-362`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L351-L362)). The bundled client similarly forwards non-response frames only to listeners currently in memory ([`rpc-client.ts:508-525`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-client.ts#L508-L525)).
+
+`get_entries({ since })` is useful catch-up for **persisted session entries**, not replay of the live RPC stream. It returns completed JSONL entries after a durable entry ID. It cannot reproduce `agent_start`, token deltas, partial tool output, queue changes, extension UI requests, or other transient frames ([RPC `get_entries` documentation](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/docs/rpc.md#get_entries), [`rpc-mode.ts:634-646`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L634-L646)).
+
+### Stdin EOF and stdout loss terminate the runtime
+
+Pi treats stdin end as shutdown. Shutdown disposes the runtime, pauses stdin, flushes stdout when applicable, and exits the process ([`rpc-mode.ts:722-813`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-mode.ts#L722-L813)). A non-retryable stdout write failure also exits the process ([`output-guard.ts:21-31`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/core/output-guard.ts#L21-L31), [`output-guard.ts:84-91`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/core/output-guard.ts#L84-L91)).
+
+This makes stdin/stdout ownership part of Pi's lifetime. A replacement daemon cannot reopen anonymous pipes created by the old daemon. Keeping Pi alive requires another process to retain both pipe ends needed by the host, continuously drain stdout, and avoid closing stdin while the daemon is absent.
+
+### Native session resume does not resume an active turn
+
+Pi persists messages only when a `message_end` event arrives. A user prompt is emitted as `message_end` before the model call, while the assistant message is persisted only after the provider stream produces its final result ([`agent-loop.ts:102-116`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/agent/src/agent-loop.ts#L102-L116), [`agent-loop.ts:281-371`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/agent/src/agent-loop.ts#L281-L371), [`agent-session.ts:640-667`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/core/agent-session.ts#L640-L667)). A hard process loss can therefore leave a durable user message with no completed assistant response; partial deltas are not session entries.
+
+Starting a new process with `--session` rebuilds completed session context. RPC exposes no command to continue the unfinished provider request or resume an interrupted tool execution; the available recovery action is a new prompt or an explicit retry instruction ([`rpc-types.ts:19-76`](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/src/modes/rpc/rpc-types.ts#L19-L76), [Pi sessions documentation](https://github.com/earendil-works/pi/blob/v0.84.2/packages/coding-agent/docs/sessions.md)). This is native conversation resume after a failed turn, not active-turn continuation.
+
+## Current Paseo behavior
+
+Paseo builds `pi --mode rpc` and adds `--session <nativeHandle>` when resuming a persisted Pi session ([`runtime.ts:75-150`](../../packages/server/src/server/agent/providers/pi/runtime.ts), [`docs/providers.md`](../../docs/providers.md)). Each `PiCliRuntime.startSession()` creates one daemon-owned `JsonlRpcProcess`; that transport spawns Pi with dedicated pipes, owns request IDs and pending promises, parses stdout, and fans non-response frames to in-memory subscribers ([`cli-runtime.ts:49-109`](../../packages/server/src/server/agent/providers/pi/cli-runtime.ts), [`jsonl-rpc-process.ts:101-229`](../../packages/server/src/server/agent/providers/jsonl-rpc-process.ts)). There is no persisted transport endpoint, pipe handle, frame cursor, or replay buffer.
+
+Paseo converts a Pi process exit into `turn_failed` only while its in-memory `PiRpcAgentSession` still has an active turn ([`agent.ts:2013-2071`](../../packages/server/src/server/agent/providers/pi/agent.ts)). On graceful daemon shutdown, `bootstrap.stop()` closes every managed agent, and Pi session close ends stdin then terminates the child process tree ([`bootstrap.ts:1714-1778`](../../packages/server/src/server/bootstrap.ts), [`agent-manager.ts:1470-1520`](../../packages/server/src/server/agent/agent-manager.ts), [`agent.ts:1532-1544`](../../packages/server/src/server/agent/providers/pi/agent.ts), [`jsonl-rpc-process.ts:205-229`](../../packages/server/src/server/agent/providers/jsonl-rpc-process.ts)).
+
+Paseo's persisted agent record and Pi's JSONL session handle are enough to rebuild completed history after restart. They do not preserve the live Pi process, active RPC request table, unanswered extension UI requests, or events emitted while the daemon is absent. Paseo's client WebSocket reconnection and authoritative timeline catch-up operate above `AgentManager`; they cannot recover frames that never reached the manager ([`docs/architecture.md`](../../docs/architecture.md), [`docs/agent-lifecycle.md`](../../docs/agent-lifecycle.md)).
+
+## Bounded local experiments
+
+The experiments used Pi 0.84.2 and files under `/tmp`; they did not modify Paseo production files.
+
+1. **Two processes, one session file.** Two simultaneous `pi --mode rpc --session /tmp/pi-rpc-reattach-session.jsonl` processes both answered `get_state` with the same session ID but different PIDs and separate extension UI event streams. This confirms that `--session` opens independent runtimes; it does not attach the second client to the first process.
+2. **Parent disappearance.** A Node parent spawned `pi --mode rpc --no-session`, started an RPC `bash` command containing `sleep 30`, then exited without killing the child. The Pi PID was alive immediately after the parent exited and gone within one second, long before the command could complete. This experiment used the direct RPC bash command rather than an LLM turn, so it establishes pipe/EOF lifetime behavior; the source paths above establish that the same RPC shutdown disposes an active agent runtime.
+
+## Requirements for a throwaway prototype
+
+Build the prototype outside the production Pi provider first. It should prove the transport and lifecycle contract, not gVisor policy or final API design.
+
+### Minimum design
+
+1. Run a small supervisor independently of the Paseo daemon. The supervisor, not the daemon, spawns Pi and owns stdin/stdout/stderr for the full runtime lifetime.
+2. Expose a local reconnectable endpoint, preferably a Unix-domain socket, keyed by a durable random bridge runtime ID. Authenticate the caller and reject a second active writer.
+3. Frame every Pi stdout JSONL record with a bridge-owned monotonic sequence before forwarding it. Retain frames until the daemon acknowledges a sequence. Fail loudly on buffer or spool exhaustion; never silently skip ahead.
+4. Continuously drain Pi stdout while no daemon is attached. Keep Pi stdin open. Preserve unanswered `extension_ui_request` frames and process-exit state for replay.
+5. Namespace or translate daemon RPC request IDs. A restarted daemon currently begins again at `req_1`; stale responses from the pre-crash generation must not satisfy new requests.
+6. Persist enough discovery state for a replacement daemon to find the bridge: Paseo agent ID, bridge runtime ID, Pi PID/start identity, Pi session file, daemon acknowledgment cursor, process state, and active Paseo turn ID.
+7. Define commands with explicit semantics: attach with last acknowledged bridge sequence, write command under an exclusive lease, acknowledge events, inspect state, close Pi, and abandon an orphan.
+8. Separate **daemon disconnect** from **agent close**. Disconnect keeps Pi alive for a bounded grace period; Stop, archive, workspace teardown, or explicit provider close must still terminate it.
+
+### Required demonstration
+
+Use a controllable fake Pi stream first, then a real Pi turn that executes a long-running tool:
+
+- kill the Paseo-side prototype client with `SIGKILL` after `turn_start`;
+- show that the Pi PID and provider/tool operation remain unchanged;
+- reconnect a new client using only persisted bridge identity and its last acknowledgment;
+- replay every missed frame in order and deliver the terminal turn event exactly once;
+- issue `abort` or steering after reattachment and show that it reaches the original Pi process;
+- restart during an unanswered extension UI request and successfully answer the replayed request;
+- verify duplicate writers, stale generations, replay-buffer overflow, Pi exit, and supervisor exit produce explicit terminal outcomes.
+
+The prototype succeeds only if the reconstructed Paseo timeline matches an uninterrupted control run and the same Pi process completes the turn. Reopening the JSONL session in a new Pi process is a failed result for this ticket.
+
+## Implications for backend and lifecycle tickets
+
+- **Backend transport:** introduce a reconnectable runtime transport below `PiRpcAgentSession`; do not teach `JsonlRpcProcess` to pretend anonymous pipes are reconnectable. Persist bridge identity separately from Pi's session-file `nativeHandle`.
+- **Startup reconciliation:** before normal native resume, ask the supervisor for live runtimes and rebind matching Paseo agents. Restore the same active Paseo turn identity and replay through the normal provider event reducer before accepting new prompts.
+- **Lifecycle ownership:** graceful daemon restart must detach from bridge-owned runtimes rather than call `session.close()`. Explicit agent close/archive must still cross the bridge and terminate Pi. Add orphan grace, adoption fencing, and supervisor-loss behavior.
+- **Timeline correctness:** Pi's session entries can supplement completed-history reconciliation, but bridge sequence/acknowledgment must be the authority for transient active-turn events. Deduplicate replay before broadcasting to clients.
+- **Permissions:** pending extension UI requests are part of the active turn. Their IDs and resolution state must survive daemon replacement, and only one daemon generation may answer them.
+- **Failure reporting:** if Pi or the supervisor dies, mark the original turn failed once, then allow native session resume. Do not report native resume as continuation of that turn.
+- **Sandbox boundary:** the gVisor-owned supervisor is the natural owner for this bridge if it already outlives daemon processes. Its control socket and spool become security-sensitive host interfaces and need per-runtime authorization, bounded storage, and cleanup.
+
+Backend and lifecycle implementation tickets should not promise active-turn continuity until the bridge prototype passes the crash/replay demonstration above.

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/.gitignore

@@ -1 +0,0 @@
-.prototype-state/

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/README.md

@@ -1,83 +0,0 @@
-# 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.

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/RESULT.md

@@ -1,46 +0,0 @@
-# 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.

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/bridge.mjs

@@ -1,290 +0,0 @@
-#!/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;
-}

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/client.mjs

@@ -1,363 +0,0 @@
-#!/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}`);

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/demo.mjs

@@ -1,163 +0,0 @@
-#!/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);
-    });
-  });
-}

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/generate-bundle.mjs

@@ -1,84 +0,0 @@
-#!/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`);

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/prepare-rootfs.sh

@@ -1,46 +0,0 @@
-#!/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"

packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/run.sh

@@ -1,19 +0,0 @@
-#!/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"