diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/.gitignore b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/.gitignore new file mode 100644 index 000000000..d8b15fd3a --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/.gitignore @@ -0,0 +1 @@ +.prototype-state/ diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/README.md b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/README.md new file mode 100644 index 000000000..3dce26f67 --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/README.md @@ -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. diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/RESULT.md b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/RESULT.md new file mode 100644 index 000000000..eb2c69f3e --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/RESULT.md @@ -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. diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/bridge.mjs b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/bridge.mjs new file mode 100755 index 000000000..3e85f5aa7 --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/bridge.mjs @@ -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 ?? ""}`); + } + parsed[key.slice(2)] = value; + } + for (const required of ["runsc", "bundle", "runtimeRoot", "socket", "containerId"]) { + if (!parsed[required]) throw new Error(`missing --${required}`); + } + return parsed; +} diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/client.mjs b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/client.mjs new file mode 100755 index 000000000..b771ca94c --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/client.mjs @@ -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 \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}`); diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/demo.mjs b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/demo.mjs new file mode 100755 index 000000000..6dde4f70c --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/demo.mjs @@ -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); + }); + }); +} diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/generate-bundle.mjs b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/generate-bundle.mjs new file mode 100755 index 000000000..a8a47dc4f --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/generate-bundle.mjs @@ -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 \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`); diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/prepare-rootfs.sh b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/prepare-rootfs.sh new file mode 100755 index 000000000..11436e904 --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/prepare-rootfs.sh @@ -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" diff --git a/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/run.sh b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/run.sh new file mode 100755 index 000000000..57b209ee4 --- /dev/null +++ b/packages/server/src/server/agent/providers/pi/prototypes/active-turn-runsc/run.sh @@ -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 <