Gabriel Vidal
e429c9cde3
Branch a new conversation off any user message: POST /api/fork locates the fork point by message ordinal (find_cut_line drives the real ParserState so the cut can't drift from the viewer), the sidecar copies the CC session file truncated at that line under a fresh session id (sessionId rewritten per record, trailing queue-operation/last-prompt bookkeeping trimmed) and resumes it; the pi runner branches pi's native session tree (--fork-from / --cut-user-ord) and mirrors the truncated CC-schema prefix. forkedFrom is stamped into the conversation meta; cutText canaries refuse a drifted slice. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
437 lines
17 KiB
JavaScript
Executable File
437 lines
17 KiB
JavaScript
Executable File
#!/usr/bin/env node
|
|
/**
|
|
* pi-runner — run one headless pi (pi.dev) session and mirror it as a
|
|
* Claude-Code-schema transcript.
|
|
*
|
|
* The ai-agent sidecar launches this exactly like it launches `claude`:
|
|
*
|
|
* pi-runner -p <prompt> --session-id <uuid> [--model <id>]
|
|
* pi-runner -p <prompt> --resume <uuid> [--model <id>]
|
|
*
|
|
* It spawns `pi --mode json` (session persisted under SESSIONS_DIR keyed by the
|
|
* SAME uuid, so resume is just the same --session-id again) and translates pi's
|
|
* AgentSessionEvent stream into the JSONL schema the viewer's parser
|
|
* (backend/conversations.py) already understands, appended live to
|
|
*
|
|
* ~/.pi-runner/transcripts/<cwd with / → ->/<uuid>.jsonl
|
|
*
|
|
* which the ai-agent container watches read-only (PI_TRANSCRIPTS_DIR mount).
|
|
*
|
|
* Model → provider routing: ids with a vendor prefix ("qwen/qwen3.6-35b-a3b")
|
|
* run on OpenRouter; bare ids ("qwen3.6-35b-a3b") run on the EVOX2 LM Studio
|
|
* provider from ~/.pi/agent/models.json (the box is woken over WoL first).
|
|
*/
|
|
import { spawn, execSync } from "node:child_process";
|
|
import { appendFileSync, existsSync, mkdirSync, readFileSync, readdirSync,
|
|
writeFileSync } from "node:fs";
|
|
import { createSocket } from "node:dgram";
|
|
import os from "node:os";
|
|
import path from "node:path";
|
|
import readline from "node:readline";
|
|
import { fileURLToPath } from "node:url";
|
|
|
|
const HERE = path.dirname(fileURLToPath(import.meta.url));
|
|
const HOME = os.homedir();
|
|
const PI_BIN = process.env.PI_BIN || path.join(HERE, "node_modules", ".bin", "pi");
|
|
const RUNNER_ROOT = process.env.PI_RUNNER_ROOT || path.join(HOME, ".pi-runner");
|
|
const TRANSCRIPTS_DIR = path.join(RUNNER_ROOT, "transcripts");
|
|
const SESSIONS_DIR = path.join(RUNNER_ROOT, "sessions");
|
|
const ENV_FILE = process.env.RUNNER_ENV_FILE || path.join(HOME, "homelab", ".env.claude");
|
|
const DEFAULT_MODEL = process.env.RUNNER_DEFAULT_MODEL || "qwen/qwen3.6-35b-a3b";
|
|
const PERM_EXT = process.env.RUNNER_PERM_EXT || path.join(HERE, "extensions", "folder-permissions.ts");
|
|
// EVOX2 (LM Studio) wake config — used only for the local provider.
|
|
const EVOX2_URL = process.env.EVOX2_URL || "http://100.93.171.39:1234";
|
|
const EVOX2_MAC = process.env.EVOX2_MAC || "84:47:09:77:48:b4";
|
|
const EVOX2_BCAST = process.env.EVOX2_BCAST || "192.168.1.255";
|
|
const WAKE_TIMEOUT_S = Number(process.env.WAKE_TIMEOUT || 150);
|
|
|
|
// pi's thinking scale. The sidecar speaks each harness's own vocabulary, so it
|
|
// sends us `off` — but it drives both CLIs with one flag shape, so accept the
|
|
// claude spelling too rather than choke on it. Anything else is dropped: an
|
|
// unrecognised level would make `pi` exit 2 before the session even starts.
|
|
const PI_THINKING = ["off", "minimal", "low", "medium", "high", "xhigh", "max"];
|
|
const THINKING_ALIASES = { disabled: "off" };
|
|
|
|
function thinkingLevel(raw) {
|
|
const level = THINKING_ALIASES[raw] ?? raw;
|
|
return PI_THINKING.includes(level) ? level : null;
|
|
}
|
|
|
|
// ── args (claude-compatible flag shape, see sidecar.py) ─────────────────────
|
|
function parseArgs(argv) {
|
|
const a = { prompt: "", sessionId: null, resume: false, model: null, thinking: null,
|
|
forkFrom: null, cutLine: 0, cutUserOrd: 0, cutText: null };
|
|
for (let i = 0; i < argv.length; i++) {
|
|
const v = argv[i];
|
|
if (v === "-p" || v === "--prompt") a.prompt = argv[++i] ?? "";
|
|
else if (v === "--session-id") a.sessionId = argv[++i];
|
|
else if (v === "--resume") { a.sessionId = argv[++i]; a.resume = true; }
|
|
else if (v === "--model") a.model = argv[++i];
|
|
else if (v === "--thinking") a.thinking = thinkingLevel(argv[++i]);
|
|
// fork: branch --fork-from's session at its --cut-user-ord'th user message
|
|
// (mirror sliced at raw line --cut-line), run under --session-id.
|
|
else if (v === "--fork-from") a.forkFrom = argv[++i];
|
|
else if (v === "--cut-line") a.cutLine = Number(argv[++i]) || 0;
|
|
else if (v === "--cut-user-ord") a.cutUserOrd = Number(argv[++i]) || 0;
|
|
else if (v === "--cut-text") a.cutText = argv[++i];
|
|
// claude flags the sidecar may still pass — accepted and ignored:
|
|
else if (v === "--output-format" || v === "--permission-mode") i++;
|
|
else if (v === "--remote-control") continue;
|
|
else if (!v.startsWith("-") && !a.prompt) a.prompt = v;
|
|
}
|
|
return a;
|
|
}
|
|
|
|
function loadEnvFile(file) {
|
|
try {
|
|
for (const line of readFileSync(file, "utf8").split("\n")) {
|
|
const m = line.match(/^([A-Z0-9_]+)=(.*)$/);
|
|
if (m && !process.env[m[1]]) process.env[m[1]] = m[2];
|
|
}
|
|
} catch { /* no env file — rely on the process env */ }
|
|
}
|
|
|
|
// ── EVOX2 wake (local provider only) ────────────────────────────────────────
|
|
async function lmStudioUp() {
|
|
try {
|
|
const r = await fetch(`${EVOX2_URL}/v1/models`, {
|
|
signal: AbortSignal.timeout(4000),
|
|
headers: { Authorization: "Bearer x" }, // any reply (even 401) = awake
|
|
});
|
|
return r.status < 500;
|
|
} catch { return false; }
|
|
}
|
|
|
|
function sendWol() {
|
|
const mac = EVOX2_MAC.split(":").map((h) => parseInt(h, 16));
|
|
const packet = Buffer.alloc(102, 0xff);
|
|
for (let i = 6; i < 102; i += 6) Buffer.from(mac).copy(packet, i);
|
|
const sock = createSocket("udp4");
|
|
sock.bind(() => {
|
|
sock.setBroadcast(true);
|
|
sock.send(packet, 9, EVOX2_BCAST, () => sock.close());
|
|
});
|
|
}
|
|
|
|
async function ensureEvox2Awake() {
|
|
if (await lmStudioUp()) return;
|
|
process.stderr.write("[pi-runner] EVOX2 asleep — sending WoL…\n");
|
|
const deadline = Date.now() + WAKE_TIMEOUT_S * 1000;
|
|
while (Date.now() < deadline) {
|
|
sendWol();
|
|
await new Promise((r) => setTimeout(r, 5000));
|
|
if (await lmStudioUp()) return;
|
|
}
|
|
throw new Error(`EVOX2 LM Studio unreachable after ${WAKE_TIMEOUT_S}s (${EVOX2_URL})`);
|
|
}
|
|
|
|
// ── Claude-Code-schema transcript writer ────────────────────────────────────
|
|
const nowIso = () => new Date().toISOString();
|
|
|
|
/** pi built-in tool call → Claude-Code tool name + input shape.
|
|
*
|
|
* The whole viewer (cost-by-tool buckets, auto project/service tags, bash
|
|
* cards, Read image previews, edit diffs) keys on Claude's PascalCase tool
|
|
* names and `file_path`-style argument keys, so the mirror translates pi's
|
|
* lowercase built-ins (`read`/`bash`/`edit`/… with `path`/`edits[]`) at write
|
|
* time. Unknown (extension) tools pass through untouched — they render as
|
|
* generic tool cards, which is right. */
|
|
function mapToolCall(name, args) {
|
|
const a = args ?? {};
|
|
switch (name) {
|
|
case "read": {
|
|
const { path: file_path, ...rest } = a;
|
|
return { name: "Read", input: { file_path, ...rest } };
|
|
}
|
|
case "write": {
|
|
const { path: file_path, ...rest } = a;
|
|
return { name: "Write", input: { file_path, ...rest } };
|
|
}
|
|
case "edit": {
|
|
// pi edit is multi-edit natively: {path, edits:[{oldText,newText}]}.
|
|
const edits = (a.edits ?? []).map((e) => ({
|
|
old_string: e?.oldText ?? "",
|
|
new_string: e?.newText ?? "",
|
|
}));
|
|
if (edits.length === 1)
|
|
return { name: "Edit", input: { file_path: a.path, ...edits[0] } };
|
|
return { name: "MultiEdit", input: { file_path: a.path, edits } };
|
|
}
|
|
case "bash": {
|
|
const { timeout, ...rest } = a;
|
|
// pi timeout is seconds; Claude's Bash carries milliseconds.
|
|
return { name: "Bash", input: {
|
|
...rest, ...(typeof timeout === "number" ? { timeout: timeout * 1000 } : {}),
|
|
} };
|
|
}
|
|
case "grep":
|
|
return { name: "Grep", input: { ...a, ...(a.caseInsensitive ? { "-i": true } : {}) } };
|
|
case "find":
|
|
return { name: "Glob", input: a };
|
|
case "ls":
|
|
return { name: "LS", input: a };
|
|
default:
|
|
return { name, input: a };
|
|
}
|
|
}
|
|
|
|
function gitBranch(cwd) {
|
|
try {
|
|
return execSync("git rev-parse --abbrev-ref HEAD", {
|
|
cwd, stdio: ["ignore", "pipe", "ignore"],
|
|
}).toString().trim() || null;
|
|
} catch { return null; }
|
|
}
|
|
|
|
class TranscriptWriter {
|
|
constructor(sessionId, cwd) {
|
|
this.sessionId = sessionId;
|
|
this.cwd = cwd;
|
|
this.branch = gitBranch(cwd);
|
|
const encoded = cwd.replaceAll("/", "-");
|
|
const dir = path.join(TRANSCRIPTS_DIR, encoded);
|
|
mkdirSync(dir, { recursive: true });
|
|
this.file = path.join(dir, `${sessionId}.jsonl`);
|
|
}
|
|
|
|
append(record) {
|
|
appendFileSync(this.file, JSON.stringify({
|
|
sessionId: this.sessionId,
|
|
cwd: this.cwd,
|
|
gitBranch: this.branch,
|
|
timestamp: nowIso(),
|
|
...record,
|
|
}) + "\n");
|
|
}
|
|
|
|
user(content) {
|
|
this.append({ type: "user", message: { role: "user", content } });
|
|
}
|
|
|
|
/** pi assistant message → CC assistant record. Splits any inline
|
|
* `<think>…</think>` (the LM Studio path) into proper thinking blocks. */
|
|
assistant(msg) {
|
|
const blocks = [];
|
|
for (const b of msg.content ?? []) {
|
|
if (!b || typeof b !== "object") continue;
|
|
if (b.type === "text") {
|
|
let rest = b.text ?? "";
|
|
const think = /<think>([\s\S]*?)<\/think>/g;
|
|
let m; let cursor = 0; const parts = [];
|
|
while ((m = think.exec(rest))) {
|
|
if (rest.slice(cursor, m.index).trim())
|
|
parts.push({ type: "text", text: rest.slice(cursor, m.index) });
|
|
if (m[1].trim()) parts.push({ type: "thinking", thinking: m[1] });
|
|
cursor = m.index + m[0].length;
|
|
}
|
|
if (rest.slice(cursor).trim()) parts.push({ type: "text", text: rest.slice(cursor) });
|
|
blocks.push(...parts);
|
|
} else if (b.type === "thinking") {
|
|
blocks.push({ type: "thinking", thinking: b.thinking ?? "" });
|
|
} else if (b.type === "toolCall") {
|
|
const mapped = mapToolCall(b.name, b.arguments);
|
|
blocks.push({ type: "tool_use", id: String(b.id), ...mapped });
|
|
}
|
|
}
|
|
const u = msg.usage ?? {};
|
|
this.append({
|
|
type: "assistant",
|
|
// pi computes the real provider cost per message (0 for the free local
|
|
// EVOX2 provider) — the parser prefers it over its own price table.
|
|
// Omitted when pi reports none, so the parser's estimate kicks in.
|
|
...(typeof u.cost?.total === "number" ? { costUSD: u.cost.total } : {}),
|
|
message: {
|
|
role: "assistant",
|
|
model: msg.model,
|
|
stop_reason: msg.stopReason,
|
|
usage: {
|
|
input_tokens: u.input ?? 0,
|
|
output_tokens: u.output ?? 0,
|
|
cache_read_input_tokens: u.cacheRead ?? 0,
|
|
cache_creation_input_tokens: u.cacheWrite ?? 0,
|
|
},
|
|
content: blocks,
|
|
},
|
|
});
|
|
}
|
|
|
|
toolResult(msg) {
|
|
// Text survives; image blocks (vision `read`) become a placeholder — the
|
|
// viewer previews images from the Read card's file_path, not the result.
|
|
const text = (msg.content ?? [])
|
|
.map((b) => (b && b.type === "text" ? b.text : b?.type === "image" ? "[image]" : ""))
|
|
.filter(Boolean).join("\n");
|
|
this.append({
|
|
type: "user",
|
|
message: {
|
|
role: "user",
|
|
content: [{
|
|
type: "tool_result",
|
|
tool_use_id: String(msg.toolCallId),
|
|
content: [{ type: "text", text }],
|
|
is_error: !!msg.isError,
|
|
}],
|
|
},
|
|
});
|
|
}
|
|
}
|
|
|
|
// ── fork: branch a session at a user message ────────────────────────────────
|
|
const collapse = (s) => (s || "").split(/\s+/).filter(Boolean).join(" ");
|
|
|
|
function findSessionFile(sid) {
|
|
// pi names session files <timestamp>_<sessionId>.jsonl and resolves
|
|
// --session-id by scanning the session dir, so we match the same way.
|
|
try {
|
|
for (const f of readdirSync(SESSIONS_DIR))
|
|
if (f.endsWith(`_${sid}.jsonl`)) return path.join(SESSIONS_DIR, f);
|
|
} catch { /* no sessions dir yet */ }
|
|
return null;
|
|
}
|
|
|
|
function findMirrorFile(sid) {
|
|
try {
|
|
for (const dir of readdirSync(TRANSCRIPTS_DIR)) {
|
|
const p = path.join(TRANSCRIPTS_DIR, dir, `${sid}.jsonl`);
|
|
if (existsSync(p)) return p;
|
|
}
|
|
} catch { /* no transcripts dir yet */ }
|
|
return null;
|
|
}
|
|
|
|
function entryTexts(msg) {
|
|
const c = msg?.content;
|
|
if (typeof c === "string") return [c];
|
|
if (Array.isArray(c))
|
|
return c.filter((b) => b?.type === "text").map((b) => b.text ?? "");
|
|
return [];
|
|
}
|
|
|
|
/** Branch a pi session at a user message. pi sessions are trees of
|
|
* id/parentId entries, but a headless runner session is a linear chain — so
|
|
* the branch is simply the entries strictly before user message
|
|
* #cutUserOrd, written under a fresh session id (header rewritten, entry ids
|
|
* kept). Resuming that id then continues from the fork point. Also writes
|
|
* the truncated CC-schema mirror (source mirror's first cutLine lines under
|
|
* the new id) so the viewer shows the inherited history. */
|
|
function forkSession(args) {
|
|
const src = findSessionFile(args.forkFrom);
|
|
if (!src) throw new Error(`source pi session not found: ${args.forkFrom}`);
|
|
const kept = [];
|
|
let userSeen = 0, found = false;
|
|
for (const line of readFileSync(src, "utf8").split("\n")) {
|
|
if (!line.trim()) continue;
|
|
let o; try { o = JSON.parse(line); } catch { continue; }
|
|
if (o.type === "session") {
|
|
kept.push({ ...o, id: args.sessionId, timestamp: nowIso() });
|
|
continue;
|
|
}
|
|
if (o.type === "message" && o.message?.role === "user") {
|
|
if (userSeen === args.cutUserOrd) {
|
|
if (args.cutText) {
|
|
const want = collapse(args.cutText);
|
|
if (!entryTexts(o.message)
|
|
.some((t) => collapse(t).slice(0, want.length) === want))
|
|
throw new Error("fork point mismatch: the pi session diverged "
|
|
+ "from the transcript the viewer showed");
|
|
}
|
|
found = true;
|
|
break;
|
|
}
|
|
userSeen++;
|
|
}
|
|
kept.push(o);
|
|
}
|
|
if (!found)
|
|
throw new Error(`fork point not found: user message #${args.cutUserOrd} `
|
|
+ `(session has ${userSeen})`);
|
|
const stamp = nowIso().replaceAll(":", "-").replace(".", "-");
|
|
writeFileSync(path.join(SESSIONS_DIR, `${stamp}_${args.sessionId}.jsonl`),
|
|
kept.map((o) => JSON.stringify(o)).join("\n") + "\n");
|
|
|
|
// The viewer-facing mirror: same prefix, CC schema, new session id. Written
|
|
// to the path the live TranscriptWriter will append to. A purged mirror
|
|
// only loses the *displayed* history — pi's context is intact — so warn.
|
|
const mirror = findMirrorFile(args.forkFrom);
|
|
if (mirror && args.cutLine > 0) {
|
|
const out = readFileSync(mirror, "utf8").split("\n")
|
|
.slice(0, args.cutLine).map((l) => {
|
|
try {
|
|
const o = JSON.parse(l);
|
|
if (o.sessionId) o.sessionId = args.sessionId;
|
|
return JSON.stringify(o);
|
|
} catch { return l; }
|
|
});
|
|
const w = new TranscriptWriter(args.sessionId, process.cwd());
|
|
writeFileSync(w.file, out.join("\n") + "\n");
|
|
} else {
|
|
process.stderr.write("[pi-runner] fork: source mirror transcript missing "
|
|
+ "— forked conversation starts without visible history\n");
|
|
}
|
|
}
|
|
|
|
// ── main ────────────────────────────────────────────────────────────────────
|
|
async function main() {
|
|
const args = parseArgs(process.argv.slice(2));
|
|
if (!args.prompt.trim()) { console.error("pi-runner: empty prompt"); process.exit(2); }
|
|
if (!args.sessionId) { console.error("pi-runner: --session-id/--resume required"); process.exit(2); }
|
|
|
|
loadEnvFile(ENV_FILE);
|
|
const model = args.model || DEFAULT_MODEL;
|
|
const provider = model.includes("/") ? "openrouter" : "evox2";
|
|
if (provider === "openrouter" && !process.env.OPENROUTER_API_KEY)
|
|
{ console.error("pi-runner: OPENROUTER_API_KEY not set (env or " + ENV_FILE + ")"); process.exit(2); }
|
|
if (provider === "evox2") await ensureEvox2Awake();
|
|
|
|
const cwd = process.cwd();
|
|
mkdirSync(SESSIONS_DIR, { recursive: true });
|
|
if (args.forkFrom) forkSession(args); // before the writer appends the prompt
|
|
const writer = new TranscriptWriter(args.sessionId, cwd);
|
|
writer.user(args.prompt);
|
|
|
|
const piArgs = [
|
|
"--mode", "json", "-p",
|
|
"--session-dir", SESSIONS_DIR,
|
|
"--session-id", args.sessionId,
|
|
"--provider", provider,
|
|
"--model", model,
|
|
];
|
|
// Unset ⇒ no flag ⇒ pi keeps the model's own default level.
|
|
if (args.thinking) piArgs.push("--thinking", args.thinking);
|
|
if (existsSync(PERM_EXT)) piArgs.push("-e", PERM_EXT);
|
|
piArgs.push(args.prompt);
|
|
|
|
const child = spawn(PI_BIN, piArgs, {
|
|
cwd,
|
|
stdio: ["ignore", "pipe", "inherit"],
|
|
env: { ...process.env, PI_PERM_SCOPE: process.env.PI_PERM_SCOPE || cwd },
|
|
});
|
|
|
|
let interrupted = false;
|
|
const onSignal = () => { interrupted = true; }; // pi shares our process group
|
|
process.on("SIGINT", onSignal);
|
|
process.on("SIGTERM", onSignal);
|
|
|
|
const rl = readline.createInterface({ input: child.stdout, crlfDelay: Infinity });
|
|
rl.on("line", (line) => {
|
|
let e; try { e = JSON.parse(line); } catch { return; }
|
|
if (e.type !== "message_end" || !e.message) return;
|
|
const role = e.message.role;
|
|
try {
|
|
if (role === "assistant") writer.assistant(e.message);
|
|
else if (role === "toolResult") writer.toolResult(e.message);
|
|
// role === "user" is the prompt we already recorded ourselves.
|
|
} catch (err) {
|
|
process.stderr.write(`[pi-runner] transcript write failed: ${err}\n`);
|
|
}
|
|
});
|
|
|
|
child.on("exit", (code, signal) => {
|
|
if (interrupted || signal === "SIGINT" || signal === "SIGTERM")
|
|
writer.user("[Request interrupted by user]");
|
|
process.exit(code ?? (signal ? 130 : 0));
|
|
});
|
|
}
|
|
|
|
main().catch((e) => { console.error(`pi-runner: ${e.message ?? e}`); process.exit(1); });
|