Something went wrong. Try again.
my pi harness customizations, including a structured, multi-model system of iterative work
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689#!/usr/bin/env node/** * A deliberately small FIFO queue for slow, free advice requests. * * Each queue item is one agentic Pi run. Every assistant message in that run is * one provider request, so the runner records a turn event as it happens and * uses today's total to decide whether launching the next item is safe. The * budget gate runs before each item; it does not interrupt a running item. * * Usage: * node scripts/advice-queue.ts add "prompt..." * node scripts/advice-queue.ts run * node scripts/advice-queue.ts status */
import { spawn } from "node:child_process";import { appendFileSync, existsSync, linkSync, mkdirSync, readFileSync, renameSync, unlinkSync, writeFileSync,} from "node:fs";import { dirname, join, resolve } from "node:path";import { createInterface } from "node:readline";import { fileURLToPath, pathToFileURL } from "node:url";
const MODEL = process.env.ADVICE_QUEUE_MODEL ?? "nvidia/z-ai/glm-5.3-flash";const DAILY_LIMIT = 1000;// Used only until the first item finishes; a real sample replaces it.const DEFAULT_EXPECTED_TURNS = 10;
const agentDir = dirname(dirname(fileURLToPath(import.meta.url)));const stateDir = process.env.ADVICE_QUEUE_DIR ?? join(agentDir, "advice-queue");const queuePath = join(stateDir, "queue.json");const turnsPath = join(stateDir, "turns.jsonl");const outputDir = join(stateDir, "out");const lockPath = join(stateDir, "lock");const piBin = process.env.PI_BIN ?? join(agentDir, "bin", "pi");
type ItemStatus = "pending" | "running" | "done" | "failed";
interface QueueItem { readonly id: string; readonly prompt: string; readonly status: ItemStatus; readonly addedAt: string; readonly turns?: number; readonly outputFile?: string; readonly error?: string; readonly startedAt?: string; readonly finishedAt?: string;}
interface TurnEvent { readonly ts: string; readonly date: string; readonly itemId: string;}
interface TurnStats { readonly count: number; readonly avg: number; readonly median: number; readonly max: number;}
interface Budget { readonly used: number; readonly remaining: number; readonly expected: number; readonly canStart: boolean; readonly atRisk: boolean;}
interface AssistantText { readonly text: string; readonly hasToolCall: boolean;}
interface ItemResult { readonly turns: number; readonly text: string; readonly error?: string;}
// ---------------------------------------------------------------------------// Pure helpers// ---------------------------------------------------------------------------
function isRecord(value: unknown): value is Record<string, unknown> { return typeof value === "object" && value !== null && !Array.isArray(value);}
function isErrno(error: unknown, code: string): boolean { return isRecord(error) && error.code === code;}
function dateKey(date: Date): string { return date.toISOString().slice(0, 10);}
export function computeStats(turns: readonly number[]): TurnStats { const values = turns.filter((value) => Number.isFinite(value) && value >= 0); if (values.length === 0) { return { count: 0, avg: 0, median: 0, max: 0 }; } const sorted = [...values].sort((a, b) => a - b); const middle = Math.floor(sorted.length / 2); const median = sorted.length % 2 === 0 ? (sorted[middle - 1] + sorted[middle]) / 2 : sorted[middle]; return { count: values.length, avg: values.reduce((sum, value) => sum + value, 0) / values.length, median, max: sorted[sorted.length - 1], };}
/** * Start when the expected item still fits. Warn when a worst-case item would * not, so the operator can decide rather than the runner guessing. */export function evaluateBudget(used: number, stats: TurnStats, limit: number): Budget { const remaining = limit - used; const expected = stats.count === 0 ? DEFAULT_EXPECTED_TURNS : Math.ceil(stats.avg); return { used, remaining, expected, canStart: remaining >= Math.max(expected, 1), atRisk: stats.count > 0 && remaining < stats.max, };}
export function isAssistantMessage( message: unknown,): message is { readonly role: "assistant"; readonly content?: unknown } { return isRecord(message) && message.role === "assistant";}
export function isAssistantTurnEvent( event: unknown,): event is { readonly type: "message_end"; readonly message: unknown } { return isRecord(event) && event.type === "message_end" && isAssistantMessage(event.message);}
export function parseAssistantText(message: unknown): AssistantText | undefined { if (!isAssistantMessage(message)) { return undefined; } const content = message.content; if (!Array.isArray(content)) { return undefined; } const textParts = content.flatMap((part) => isRecord(part) && part.type === "text" && typeof part.text === "string" ? [part.text] : [], ); if (textParts.length === 0) { return undefined; } const hasToolCall = content.some( (part) => isRecord(part) && (part.type === "toolCall" || part.type === "tool_call"), ); const text = textParts.join("\n").trim(); return text.length > 0 ? { text, hasToolCall } : undefined;}
export function parseQueue(value: unknown): readonly QueueItem[] { if (!Array.isArray(value)) { throw new Error("queue.json must contain an array"); } return value.map((entry) => { if ( !isRecord(entry) || typeof entry.id !== "string" || typeof entry.prompt !== "string" || typeof entry.status !== "string" || typeof entry.addedAt !== "string" ) { throw new Error("queue.json contains an invalid item"); } const status = entry.status; if (status !== "pending" && status !== "running" && status !== "done" && status !== "failed") { throw new Error(`queue.json item ${entry.id} has an unknown status`); } return { id: entry.id, prompt: entry.prompt, status, addedAt: entry.addedAt, turns: typeof entry.turns === "number" ? entry.turns : undefined, outputFile: typeof entry.outputFile === "string" ? entry.outputFile : undefined, error: typeof entry.error === "string" ? entry.error : undefined, startedAt: typeof entry.startedAt === "string" ? entry.startedAt : undefined, finishedAt: typeof entry.finishedAt === "string" ? entry.finishedAt : undefined, }; });}
// ---------------------------------------------------------------------------// State// ---------------------------------------------------------------------------
function ensureStateDir(): void { mkdirSync(outputDir, { recursive: true });}
function loadQueue(): readonly QueueItem[] { if (!existsSync(queuePath)) { return []; } const parsed: unknown = JSON.parse(readFileSync(queuePath, "utf8")); return parseQueue(parsed);}
function saveQueue(items: readonly QueueItem[]): void { const temporary = `${queuePath}.${process.pid}.${Date.now()}.tmp`; try { writeFileSync(temporary, `${JSON.stringify(items, null, 2)}\n`); renameSync(temporary, queuePath); } finally { try { unlinkSync(temporary); } catch { // The rename already removed it, or the write never created it. } }}
function appendTurn(itemId: string): void { const now = new Date(); const event: TurnEvent = { ts: now.toISOString(), date: dateKey(now), itemId }; appendFileSync(turnsPath, `${JSON.stringify(event)}\n`);}
function readTurnsForToday(): number { if (!existsSync(turnsPath)) { return 0; } const today = dateKey(new Date()); return readFileSync(turnsPath, "utf8") .split("\n") .filter((line) => line.length > 0) .reduce((count, line) => { try { const parsed: unknown = JSON.parse(line); return isRecord(parsed) && parsed.date === today ? count + 1 : count; } catch { return count; } }, 0);}
export function itemTurns(items: readonly QueueItem[]): readonly number[] { return items.flatMap((item) => (item.turns === undefined || item.turns <= 0 ? [] : [item.turns]));}
function writeOutput(item: QueueItem, result: ItemResult, startedAt: string): string { const safeId = item.id.replace(/[^A-Za-z0-9._-]/g, "_"); const outputFile = join(outputDir, `${safeId}.md`); const prompt = item.prompt.replace(/\n/g, "\n> "); const body = [ `# ${item.id}`, "", `- Model: \`${MODEL}\``, `- Started: ${startedAt}`, `- Finished: ${new Date().toISOString()}`, `- Turns: ${result.turns}`, `- Status: ${result.error ? "failed (partial output)" : "done"}`, "", "## Prompt", "", `> ${prompt}`, "", "## Advice", "", result.text, "", ].join("\n"); writeFileSync(outputFile, body); return outputFile;}
// ---------------------------------------------------------------------------// Lock// ---------------------------------------------------------------------------
function processIsAlive(pid: number): boolean { try { process.kill(pid, 0); return true; } catch (error) { return !isErrno(error, "ESRCH"); }}
function tryCreateLock(path: string): boolean { const temporary = `${path}.${process.pid}.${Date.now()}.tmp`; writeFileSync(temporary, String(process.pid)); try { linkSync(temporary, path); return true; } catch (error) { if (isErrno(error, "EEXIST")) { return false; } throw error; } finally { try { unlinkSync(temporary); } catch { // The temporary file is best-effort cleanup. } }}
function readLockPid(path: string): number | undefined { try { const value = Number(readFileSync(path, "utf8").trim()); return Number.isInteger(value) && value > 0 ? value : undefined; } catch { return undefined; }}
function releaseOwnLock(path: string): void { if (readLockPid(path) !== process.pid) { return; } try { unlinkSync(path); } catch { // A missing lock is already released. }}
function acquireRunLock(): () => void { ensureStateDir(); for (let attempt = 0; attempt < 3; attempt += 1) { if (tryCreateLock(lockPath)) { return () => releaseOwnLock(lockPath); } const existingPid = readLockPid(lockPath); if (existingPid !== undefined && processIsAlive(existingPid)) { throw new Error(`another queue runner is active (pid ${existingPid})`); } try { unlinkSync(lockPath); } catch (error) { if (!isErrno(error, "ENOENT")) { throw error; } } } throw new Error("could not acquire the queue lock");}
async function withQueueLock<T>(update: () => T | Promise<T>): Promise<T> { const path = `${queuePath}.lock`; const deadline = Date.now() + 5000; for (;;) { if (tryCreateLock(path)) { break; } const pid = readLockPid(path); if (pid !== undefined && !processIsAlive(pid)) { try { unlinkSync(path); } catch { // Someone else already removed the stale lock. } continue; } if (Date.now() > deadline) { throw new Error("timed out waiting for the queue lock"); } await new Promise((resolveWait) => setTimeout(resolveWait, 50)); } try { return await update(); } finally { releaseOwnLock(path); }}
async function mutateQueue(update: (items: readonly QueueItem[]) => readonly QueueItem[]): Promise<void> { await withQueueLock(() => saveQueue(update(loadQueue())));}
// ---------------------------------------------------------------------------// Runner// ---------------------------------------------------------------------------
let activeChild: ReturnType<typeof spawn> | undefined;
function runItem(item: QueueItem): Promise<ItemResult> { return new Promise((resolveResult) => { let child: ReturnType<typeof spawn>; try { child = spawn( piBin, [ "--mode", "json", "--no-session", "--no-extensions", "--no-skills", "--model", MODEL, "--tools", "read,grep,find,ls", "-p", "--", item.prompt, ], { cwd: process.cwd(), env: process.env, stdio: ["ignore", "pipe", "pipe"] }, ); } catch (error) { resolveResult({ turns: 0, text: "", error: error instanceof Error ? error.message : String(error), }); return; } activeChild = child; let turns = 0; let finalText = ""; let fallbackText = ""; let stderr = ""; let settled = false; const finish = (result: ItemResult): void => { if (!settled) { settled = true; activeChild = undefined; resolveResult(result); } };
if (!child.stdout || !child.stderr) { child.kill(); finish({ turns: 0, text: "", error: "could not capture pi output" }); return; }
const lines = createInterface({ input: child.stdout }); lines.on("line", (line) => { let parsed: unknown; try { parsed = JSON.parse(line); } catch { return; } if (!isAssistantTurnEvent(parsed)) { return; } turns += 1; appendTurn(item.id); process.stderr.write(` ${item.id}: turn ${turns}\n`); const assistant = parseAssistantText(parsed.message); if (!assistant) { return; } fallbackText = assistant.text; if (!assistant.hasToolCall) { finalText = assistant.text; } }); child.stderr.on("data", (chunk: Buffer) => { stderr = `${stderr}${chunk.toString()}`.slice(-2000); }); child.on("error", (error) => { lines.close(); child.kill(); finish({ turns, text: finalText || fallbackText, error: error.message }); }); child.on("close", (code) => { lines.close(); const text = finalText || fallbackText; if (code === 0 && text.length > 0) { finish({ turns, text }); return; } const detail = stderr.trim().split("\n").at(-1) ?? "no stderr"; const error = code === 0 ? "pi produced no final answer" : `pi exited ${code}: ${detail}`; finish({ turns, text, error }); }); });}
function printBudget(used: number, stats: TurnStats): Budget { const budget = evaluateBudget(used, stats, DAILY_LIMIT); const history = stats.count === 0 ? "no completed items yet" : `avg ${stats.avg.toFixed(1)}, median ${stats.median}, max ${stats.max} (n=${stats.count})`; console.log(`turns: ${history}`); console.log( `today: ${used}/${DAILY_LIMIT} requests, ${budget.remaining} remaining, next expected ~${budget.expected}`, ); if (!budget.canStart) { console.log("stop: the expected next item does not fit in today's budget"); } else if (budget.atRisk) { console.log(`warning: an above-average item (max ${stats.max}) would exceed today's limit`); } return budget;}
async function runQueue(): Promise<void> { ensureStateDir(); const release = acquireRunLock(); const stop = (): void => { process.removeListener("SIGINT", stop); process.removeListener("SIGTERM", stop); const child = activeChild; let stopped = false; let timer: ReturnType<typeof setTimeout> | undefined; const finishStop = (): void => { if (stopped) { return; } stopped = true; if (timer !== undefined) { clearTimeout(timer); } release(); process.exit(1); }; if (!child || child.exitCode !== null || child.signalCode !== null) { finishStop(); return; } timer = setTimeout(() => { child.kill("SIGKILL"); finishStop(); }, 2000); child.once("close", finishStop); try { child.kill("SIGTERM"); } catch { finishStop(); } }; process.once("SIGINT", stop); process.once("SIGTERM", stop); try { await mutateQueue((items) => items.map((item) => item.status === "running" ? { ...item, status: "pending" as const, error: "interrupted; requeued" } : item, ), );
for (;;) { const items = loadQueue(); const item = items.find((candidate) => candidate.status === "pending"); if (!item) { console.log("queue empty"); break; } const budget = printBudget(readTurnsForToday(), computeStats(itemTurns(items))); if (!budget.canStart) { break; }
const startedAt = new Date().toISOString(); await mutateQueue((current) => current.map((candidate) => candidate.id === item.id ? { ...candidate, status: "running" as const, startedAt } : candidate, ), ); console.log(`running ${item.id}`); const result = await runItem(item); const finishedAt = new Date().toISOString(); const outputFile = !result.error || result.text.length > 0 ? writeOutput(item, result, startedAt) : undefined;
await mutateQueue((current) => current.map((candidate) => candidate.id === item.id ? { ...candidate, status: result.error ? ("failed" as const) : ("done" as const), turns: result.turns, finishedAt, error: result.error, outputFile, } : candidate, ), ); if (result.error) { console.error(`${item.id} failed after ${result.turns} turns: ${result.error}`); } else { console.log(`${item.id} done in ${result.turns} turns -> ${outputFile}`); } }
const stats = computeStats(itemTurns(loadQueue())); console.log( `stats: n=${stats.count}, avg=${stats.avg.toFixed(1)}, median=${stats.median}, max=${stats.max}`, ); } finally { process.removeListener("SIGINT", stop); process.removeListener("SIGTERM", stop); release(); }}
function showStatus(): void { ensureStateDir(); const items = loadQueue(); const stats = computeStats(itemTurns(items)); const count = (status: ItemStatus): number => items.filter((item) => item.status === status).length; console.log( `queue: ${count("pending")} pending, ${count("running")} running, ${count("done")} done, ${count("failed")} failed`, ); printBudget(readTurnsForToday(), stats); const pending = items.filter((item) => item.status === "pending"); if (pending.length > 0) { console.log("pending:"); for (const item of pending) { console.log(` ${item.id} ${item.prompt.slice(0, 100)}`); } }}
async function addItem(prompt: string): Promise<void> { ensureStateDir(); const now = new Date(); const item: QueueItem = { id: `q-${now.getTime().toString(36)}-${Math.random().toString(36).slice(2, 6)}`, prompt, status: "pending", addedAt: now.toISOString(), }; await mutateQueue((items) => [...items, item]); console.log(`queued ${item.id}`);}
// ---------------------------------------------------------------------------// CLI// ---------------------------------------------------------------------------
function usage(): void { console.log("usage: node scripts/advice-queue.ts <add|run|status> [prompt]");}
async function main(): Promise<void> { const [command, ...rest] = process.argv.slice(2); switch (command) { case "add": { const prompt = rest.join(" ").trim(); if (prompt.length === 0) { usage(); process.exitCode = 1; return; } await addItem(prompt); return; } case "run": await runQueue(); return; case "status": showStatus(); return; default: usage(); process.exitCode = 1; }}
const invokedDirectly = process.argv[1] !== undefined && pathToFileURL(resolve(process.argv[1])).href === import.meta.url;
if (invokedDirectly) { main().catch((error: unknown) => { console.error(error instanceof Error ? error.message : String(error)); process.exitCode = 1; });}