Something went wrong. Try again.
Parametric learning in your code harness.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468// `machine <name>`: use a machine directly, in Pi's interactive terminal UI.//// Pi shows the conversation; the machine supplies everything else. The model// sees exactly the machine's instructions and tools (as under the pi harness),// tools run in this process against a sandbox over the current directory, and// every exchange (a prompt and everything the agent did for it) is stored as a// run with its full trajectory, so interactive use produces the same records// that suites, grading, and training read. Slash commands reach back into the// machine://// /score grade the last exchange against the machine's rubric// /good, /bad mark the last exchange (stored as feedback on its run)// /correct <note> record what should change ("too terse"), for later rubric// and case proposals// /machine show the machine, its model, and this session's runs and spend//// Pi's own built-in tools, extensions, skills, context files, and ~/.pi setup// stay out, as in the harness. Pi's state for these sessions lives under// .machine/pi/, so `--continue` resumes the last conversation with a machine.
import { spawn } from "node:child_process";import { randomBytes } from "node:crypto";import { mkdir, writeFile } from "node:fs/promises";import { createServer, type IncomingMessage, type Server } from "node:http";import path from "node:path";import { piEnvironment, piResult, resolvePiModel, TINKER_PROVIDER, type PiMessage } from "./harnesses/pi.ts";import { TINKER_OAI_BASE_URL } from "./harnesses/chat-completions.ts";import type { Machine } from "./machine.ts";import type { Tool, ToolContext } from "./harness.ts";import type { Price } from "./pricing.ts";import { costOf } from "./pricing.ts";import { directorySandbox, type Isolation } from "./sandbox.ts";import { preview } from "./events.ts";import { nameSegments, newRunId } from "./store.ts";import type { Feedback, Message, Run } from "./types.ts";import type { Workspace } from "./workspace.ts";
export type InteractiveOptions = { /** Where the session works: tools act here. Defaults to the current directory. */ cwd?: string; /** Model override; defaults to the machine's. */ model?: string; /** The machine version recorded on each exchange's run, e.g. "v2". */ version?: string; /** Resume the last Pi session with this machine. */ continue?: boolean; /** Grade every exchange as it finishes (judge graders cost money). */ evaluate?: boolean; /** Price of a model, for the spend shown by /machine. */ price?: (model: string) => Price | undefined; isolation?: Isolation; /** The Pi executable. */ command?: string; /** Directory for Pi's state for machine sessions (agent dir and session files). */ stateDir: string; /** Extra arguments for pi, e.g. ["-p", "hello"] to run one prompt non-interactively (tests, scripts). */ piArgs?: string[];};
type Exchange = { run: Run };
/** Open the machine in Pi and block until the user quits. Returns Pi's exit code. */export async function interactive(machine: Machine, workspace: Workspace, options: InteractiveOptions): Promise<number> { const cwd = path.resolve(options.cwd ?? process.cwd()); const model = options.model ?? machine.model; const target = resolvePiModel(model); if (target.tinker && !process.env.TINKER_API_KEY) throw new Error(`${model} runs on Tinker; set TINKER_API_KEY.`);
const sandbox = await directorySandbox(cwd, { isolation: options.isolation }); const session = `${path.basename(cwd).toLowerCase().replace(/[^a-z0-9._-]+/g, "-") || "session"}-${newRunId()}`; const exchanges: Exchange[] = []; let history: Message[] = []; let prompt: string | undefined;
const record = async (messages: PiMessage[]): Promise<Run | undefined> => { const input = userText(messages); if (input === undefined) return undefined; let result; try { result = piResult(input, messages.filter((m) => m.role !== "user")); } catch { return undefined; // a failed attempt Pi will retry or report itself } const usage = result.usage && !result.usage.costUsd && options.price?.(model) ? { ...result.usage, costUsd: costOf(options.price(model)!, result.usage) } : result.usage; const now = new Date().toISOString(); let run: Run = { id: newRunId(), machine: machine.name, ...(options.version ? { version: options.version } : {}), workspace: workspace.name, instructions: machine.instructions, input, history, trajectory: result.trajectory, output: result.output, model, harness: "pi-interactive", usage, stopReason: result.stopReason, startedAt: now, finishedAt: now, session, }; await workspace.store.putRun(run); workspace.events.emit({ type: "run_end", machine: machine.name, run: run.id, model, session, stopReason: run.stopReason, costUsd: usage?.costUsd, input: preview(input), output: preview(run.output) }); history = [...history, ...result.trajectory]; prompt = undefined; await workspace.store.putSession({ machine: machine.name, name: session, messages: history, updatedAt: now }); if (options.evaluate) run = await workspace.evaluate(machine, run); exchanges.push({ run }); return run; };
const last = () => exchanges.at(-1); const addFeedback = async (feedback: Feedback) => { const exchange = last(); if (!exchange) return "Nothing to mark yet: no exchange has finished in this session."; exchange.run = { ...exchange.run, feedback: [...(exchange.run.feedback ?? []), feedback] }; await workspace.store.putRun(exchange.run); workspace.events.emit({ type: "feedback", machine: machine.name, run: exchange.run.id, kind: feedback.kind, note: feedback.note }); return `Marked ${exchange.run.id} ${feedback.kind}${feedback.note ? `: ${feedback.note}` : ""}.`; };
// Tools run here, in the machine process, against the session's sandbox. Calls stream // progress back to Pi (and keep the connection alive) while a long tool runs. const byName = new Map(machine.tools.map((t) => [t.name, t])); const runTool = (body: { name?: string; arguments?: unknown }, send: (line: ToolStreamLine) => void, signal: AbortSignal) => { const tool = byName.get(body.name ?? ""); if (!tool) return Promise.resolve(send({ result: `Unknown tool "${body.name}".`, isError: true })); const conversation = () => [...history, ...(prompt === undefined ? [] : [{ role: "user" as const, content: prompt }])]; return streamToolCall(tool, body.arguments ?? {}, { sandbox, interactive: true, conversation, signal }, send, { onStart: (input) => workspace.events.emit({ type: "tool_call", machine: machine.name, session, name: tool.name, input }), onEnd: (text) => workspace.events.emit({ type: "tool_result", machine: machine.name, session, name: tool.name, preview: preview(text) }), }); };
const control = await serveControl(runTool, async (action, body) => { switch (action) { case "start": prompt = body.note ?? ""; workspace.events.emit({ type: "run_start", machine: machine.name, model, harness: "pi-interactive", session, input: preview(body.note ?? "") }); return ""; case "record": { const run = await record(body.messages ?? []); return run?.scorecard ? `${scoreLine(run)}` : ""; } case "score": { const exchange = last(); if (!exchange) return "Nothing to score yet: no exchange has finished in this session."; exchange.run = await workspace.evaluate(machine, exchange.run); return scorecardText(exchange.run); } case "good": case "bad": return addFeedback({ kind: action, note: body.note || undefined, at: new Date().toISOString() }); case "correct": if (!body.note?.trim()) return "Usage: /correct <what should change>, e.g. /correct too terse"; return addFeedback({ kind: "correction", note: body.note.trim(), at: new Date().toISOString() }); case "machine": { const spent = exchanges.reduce((s, e) => s + (e.run.usage?.costUsd ?? 0) + (e.run.scorecard?.usage?.costUsd ?? 0), 0); return [ `${machine.name}${options.version ? ` ${options.version}` : ""} · ${model}`, machine.intent, `criteria: ${Object.keys(machine.rubric).join(", ")}`, `tools: ${machine.tools.map((t) => t.name).join(", ") || "none"}`, `workspace: ${cwd}${sandbox.isolated ? " (commands isolated)" : " (commands NOT isolated: bubblewrap unavailable)"}`, `session ${session}: ${exchanges.length} exchange${exchanges.length === 1 ? "" : "s"} stored, about $${spent.toFixed(4)}`, ].join("\n"); } default: throw new Error(`unknown action ${action}`); } });
try { const state = path.join(options.stateDir, ...nameSegments(machine.name, "machine name")); const agentDir = path.join(state, "agent"); await mkdir(agentDir, { recursive: true, mode: 0o700 }); const extensionFile = path.join(state, "machine-session.js"); const systemPromptFile = path.join(state, "system-prompt.md"); await writeFile(systemPromptFile, machine.instructions, { mode: 0o600 }); await writeFile( extensionFile, interactiveExtensionSource({ systemPrompt: machine.instructions, tools: machine.tools.map((t) => ({ name: t.name, description: t.description, inputSchema: t.inputSchema })), provider: target.tinker ? tinkerProviderConfig(target.model) : null, }), { mode: 0o600 }, ); const env = { ...piEnvironment(agentDir, undefined, process.env, { attribution: target.provider === "openrouter" }), MACHINE_PI_CONTROL_URL: control.url, MACHINE_PI_CONTROL_TOKEN: control.token, }; const args = [ "--offline", "--no-extensions", "--extension", extensionFile, "--no-skills", "--no-prompt-templates", "--no-themes", "--no-context-files", "--no-approve", "--session-dir", path.join(state, "sessions"), "--system-prompt", systemPromptFile, ...(machine.tools.length > 0 ? ["--tools", machine.tools.map((t) => t.name).join(",")] : ["--no-tools"]), "--provider", target.provider, "--model", target.model, ...(options.continue ? ["--continue"] : []), ...(options.piArgs ?? []), ]; return await new Promise<number>((resolve, reject) => { const proc = spawn(options.command ?? "pi", args, { cwd, env, stdio: "inherit" }); proc.once("error", (error) => reject(new Error(`Could not start pi (is Pi installed and on PATH?): ${error.message}`))); proc.once("exit", (code, signal) => resolve(code ?? (signal ? 1 : 0))); }); } finally { await control.close(); if (exchanges.length) console.error(`${machine.name}: ${exchanges.length} exchange${exchanges.length === 1 ? "" : "s"} stored in session ${session}.`); }}
function userText(messages: PiMessage[]): string | undefined { const user = messages.find((m) => m.role === "user"); if (!user) return undefined; if (typeof user.content === "string") return user.content; return (user.content ?? []).map((c) => (c.type === "text" && typeof (c as { text?: unknown }).text === "string" ? (c as { text: string }).text : "")).join("");}
function scoreLine(run: Run): string { const failing = Object.entries(run.scorecard!.criteria).filter(([, c]) => c.score < 1).map(([id]) => id); return `score ${run.scorecard!.total.toFixed(2)}${failing.length ? ` (below 1: ${failing.join(", ")})` : ""}`;}
function scorecardText(run: Run): string { const sc = run.scorecard!; return [`score ${sc.total.toFixed(2)}`, ...Object.entries(sc.criteria).map(([id, c]) => `${id}: ${c.score.toFixed(2)} (${c.grader}) ${c.reason}`)].join("\n");}
/** One line of a streamed tool call: progress while it runs, a keep-alive tick, then the result. */export type ToolStreamLine = { progress: string } | { tick: true } | { result: string; isError?: boolean };
/** * Run a tool and report it as stream lines: each progress line the tool * reports, a tick every `tickMs` so the connection never sits idle (HTTP * clients time out an idle response), then the result or the error. */export async function streamToolCall( tool: Tool, input: unknown, context: ToolContext, send: (line: ToolStreamLine) => void, hooks: { onStart?: (input: unknown) => void; onEnd?: (text: string) => void; tickMs?: number } = {},): Promise<void> { hooks.onStart?.(input); const tick = setInterval(() => send({ tick: true }), hooks.tickMs ?? 15_000); try { const result = await tool.execute(input, { ...context, progress: (line) => send({ progress: line }) }); const text = typeof result === "string" ? result : JSON.stringify(result); hooks.onEnd?.(text); send({ result: text }); } catch (error) { const text = error instanceof Error ? error.message : String(error); hooks.onEnd?.(text); send({ result: text, isError: true }); } finally { clearInterval(tick); }}
/** * A loopback endpoint the session's Pi extension calls: POST /<action> with a * bearer token, JSON in, text out; POST /tool streams a tool call as JSON lines. */async function serveControl( runTool: (body: { name?: string; arguments?: unknown }, send: (line: ToolStreamLine) => void, signal: AbortSignal) => Promise<void>, handle: (action: string, body: { messages?: PiMessage[]; note?: string }) => Promise<string>,) { const token = randomBytes(24).toString("hex"); const server: Server = createServer(async (req, res) => { if (req.method !== "POST" || req.headers.authorization !== `Bearer ${token}`) { res.writeHead(req.method === "POST" ? 401 : 405).end(); return; } try { const body = JSON.parse((await readBody(req)) || "{}"); const action = (req.url ?? "/").slice(1); if (action === "tool") { const abort = new AbortController(); res.on("close", () => abort.abort()); res.writeHead(200, { "content-type": "application/x-ndjson; charset=utf-8" }); await runTool(body, (line) => void (res.writableEnded || res.write(`${JSON.stringify(line)}\n`)), abort.signal); res.end(); return; } const text = await handle(action, body); res.writeHead(200, { "content-type": "text/plain; charset=utf-8" }).end(text); } catch (error) { if (res.headersSent) res.end(); else res.writeHead(500, { "content-type": "text/plain; charset=utf-8" }).end(error instanceof Error ? error.message : String(error)); } }); await new Promise<void>((resolve) => server.listen(0, "127.0.0.1", resolve)); const address = server.address(); const port = typeof address === "object" && address ? address.port : 0; return { url: `http://127.0.0.1:${port}`, token, close: () => new Promise<void>((resolve) => { server.closeAllConnections?.(); server.close(() => resolve()); }), };}
function readBody(req: IncomingMessage): Promise<string> { return new Promise((resolve, reject) => { let data = ""; req.setEncoding("utf8"); req.on("data", (chunk) => (data += chunk)); req.on("end", () => resolve(data)); req.on("error", reject); });}
function tinkerProviderConfig(model: string) { return { name: TINKER_PROVIDER, config: { name: "Tinker", baseUrl: TINKER_OAI_BASE_URL, apiKey: "$TINKER_API_KEY", api: "openai-completions", models: [ { id: model, name: model, reasoning: false, input: ["text"], cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, contextWindow: 65_536, maxTokens: 8_192, compat: { supportsStore: false, supportsDeveloperRole: false, supportsReasoningEffort: false, supportsStrictMode: false, maxTokensField: "max_tokens" }, }, ], }, };}
/** Source of the session's Pi extension: the machine's prompt, tools, provider, record hook, and slash commands. */export function interactiveExtensionSource(spec: { systemPrompt: string; tools: Array<{ name: string; description: string; inputSchema: Record<string, unknown> }>; provider: { name: string; config: Record<string, unknown> } | null;}): string { return `// Generated by machine for an interactive session; rewritten each time the machine is opened.const SPEC = ${JSON.stringify(spec)};
export default function (pi) { const controlUrl = process.env.MACHINE_PI_CONTROL_URL; const controlToken = process.env.MACHINE_PI_CONTROL_TOKEN; if (SPEC.provider) pi.registerProvider(SPEC.provider.name, SPEC.provider.config);
// The machine's instructions are the whole system prompt. pi.on("before_agent_start", (event) => { control("start", { note: event?.prompt ?? "" }).catch(() => {}); return { systemPrompt: SPEC.systemPrompt }; });
async function control(action, body) { const response = await fetch(controlUrl + "/" + action, { method: "POST", headers: { "content-type": "application/json", authorization: "Bearer " + controlToken }, body: JSON.stringify(body ?? {}), }); const text = await response.text(); if (!response.ok) throw new Error(text || "machine returned HTTP " + response.status); return text; }
// Store each finished exchange as a run. pi.on("agent_end", async (event, ctx) => { try { const note = await control("record", { messages: event.messages }); if (note) ctx.ui.notify(note, "info"); } catch (error) { ctx.ui.notify("machine could not store this exchange: " + (error instanceof Error ? error.message : String(error)), "error"); } });
const commands = { score: "Grade the last exchange against the machine's rubric", good: "Mark the last exchange good", bad: "Mark the last exchange bad", correct: "Record what should change about the last exchange: /correct <note>", machine: "Show the machine, its model, and this session", }; for (const [name, description] of Object.entries(commands)) { pi.registerCommand(name, { description, handler: async (args, ctx) => { try { if (name === "score") ctx.ui.notify("grading…", "info"); ctx.ui.notify(await control(name, { note: (args ?? "").trim() }), "info"); } catch (error) { ctx.ui.notify(error instanceof Error ? error.message : String(error), "error"); } }, }); }
// A tool call streams JSON lines: progress (shown in Pi as it happens), ticks, then the result. async function callTool(name, args, signal, onUpdate) { try { const response = await fetch(controlUrl + "/tool", { method: "POST", headers: { "content-type": "application/json", authorization: "Bearer " + controlToken }, body: JSON.stringify({ name, arguments: args ?? {} }), signal, }); if (!response.ok || !response.body) return { text: JSON.stringify({ error: "machine tool endpoint returned HTTP " + response.status }) }; const reader = response.body.getReader(); const decoder = new TextDecoder(); const progress = []; let buffer = ""; for (;;) { const { value, done } = await reader.read(); if (value) buffer += decoder.decode(value, { stream: true }); let newline; while ((newline = buffer.indexOf("\\n")) >= 0) { const line = buffer.slice(0, newline).trim(); buffer = buffer.slice(newline + 1); if (!line) continue; const message = JSON.parse(line); if (typeof message.progress === "string") { progress.push(message.progress); if (onUpdate) onUpdate({ content: [{ type: "text", text: progress.slice(-12).join("\\n") }], details: {} }); } else if (typeof message.result === "string") { return { text: message.isError ? JSON.stringify({ error: message.result }) : message.result }; } } if (done) return { text: JSON.stringify({ error: "the machine ended the tool call without a result" }) }; } } catch (error) { return { text: JSON.stringify({ error: error instanceof Error ? error.message : String(error) }) }; } }
let queue = Promise.resolve(); for (const tool of SPEC.tools) { pi.registerTool({ name: tool.name, label: tool.name, description: tool.description, parameters: tool.inputSchema, execute(_toolCallId, params, signal, onUpdate) { const result = queue.then(() => callTool(tool.name, params, signal, onUpdate)); queue = result.catch(() => {}); return result.then(({ text }) => ({ content: [{ type: "text", text }], details: {} })); }, }); }}`;}