rocko
4bdc7e0e86
Unify halt and fire cancellation by verified run identity with durable first-owner receipts and shared deadlines. Integrate the instance control with the current host, library and tester-socket paths. Focused nine-file evidence is 62 passing tests; broad suites, mutants and review remain pending.
461 lines
20 KiB
JavaScript
461 lines
20 KiB
JavaScript
// The engine: one `pi --mode rpc` child per binding, one conversation, one
|
|
// turn at a time from the connector's point of view. A prompt sent while pi
|
|
// is busy is held here and sent when pi settles, so a second Discord
|
|
// message during a turn is neither lost nor run concurrently. It is not
|
|
// sent as a pi follow-up (streamingBehavior followUp): pi folds a follow-up
|
|
// into the running agent loop and closes both answers with one agent_end,
|
|
// which lost the first answer live on 2026-09-17 (the second message's
|
|
// reply was posted as the first message's, the second failed as settled
|
|
// without a turn). One prompt per run keeps the events unambiguous.
|
|
//
|
|
// Each prompt resolves on the `agent_end` event that closes its run (one
|
|
// run per prompt, in the order prompts were sent). A run holds one or more
|
|
// pi turns: with tools, an assistant message that only calls tools ends a
|
|
// turn and the next turn carries the answer. The reply is the last
|
|
// assistant message of the run; every tool call in between is collected
|
|
// from `tool_execution_start`/`tool_execution_end` into the result so the
|
|
// turn record shows what was read. An `agent_end` with `willRetry` is not
|
|
// the end of the run. A timeout sends `abort` and fails that turn; the
|
|
// process stays. The failed turn holds later prompts back until its
|
|
// agent_end or a settle. If pi has not started it within ABORT_GRACE_MS, the
|
|
// engine stops pi instead of sending again: pi's events carry no prompt id,
|
|
// so a late run of the failed prompt would be taken for the next one's. A
|
|
// run pi did start holds later prompts until it ends, as any run does. A
|
|
// malformed JSONL line from pi fails the current turn (its outcome is now
|
|
// unknowable) and the process stays. Process exit fails every pending turn
|
|
// and is reported through `onExit`.
|
|
//
|
|
// Framing follows pi's RPC doc: split on "\n" only, strip a trailing "\r".
|
|
// Node readline is not used because it also splits on U+2028/U+2029.
|
|
//
|
|
// This module can be replaced by the CHAT-03 conversation controller later
|
|
// without the connector noticing: the contract is start(), prompt(), stop().
|
|
|
|
import { spawn as nodeSpawn } from "node:child_process";
|
|
import { fileURLToPath } from "node:url";
|
|
import { DiscordError } from "./errors.mjs";
|
|
import { enabledToolNames } from "./tools.mjs";
|
|
import { assertProcessing, processingStatus } from "../../processing-control/src/state.mjs";
|
|
import { registerRuntime, updateRuntime, unregisterRuntime, processIdentity, runtime, runtimeGroupState, createRuntimeSignaller } from "../../processing-control/src/runtimes.mjs";
|
|
|
|
export const PI_FIXED_ARGS = Object.freeze([
|
|
"--mode", "rpc", "--no-extensions", "--no-context-files", "--no-skills",
|
|
"--no-prompt-templates", "--no-themes", "--offline",
|
|
]);
|
|
// Without tools: pi's own tools off, no extension. With tools: pi's
|
|
// built-in tools off, our extension loaded explicitly, and an allowlist of
|
|
// exactly its tool names. --no-extensions stays in both cases; it disables
|
|
// discovery, not an explicit --extension.
|
|
export const PI_NO_TOOLS_ARGS = Object.freeze(["--no-tools"]);
|
|
export const TOOLS_EXTENSION = fileURLToPath(new URL("../extension/tools.mjs", import.meta.url));
|
|
// Older name, kept for the tests and docs that still use it.
|
|
export const READONLY_TOOLS_EXTENSION = TOOLS_EXTENSION;
|
|
|
|
export function buildPiArgs({ provider, model, thinking, sessionDir, appendSystemPromptFile, continueSession, tools = null }) {
|
|
const args = [...PI_FIXED_ARGS];
|
|
if (tools) args.push("--no-builtin-tools", "--extension", TOOLS_EXTENSION, "--tools", enabledToolNames(tools).join(","));
|
|
else args.push(...PI_NO_TOOLS_ARGS);
|
|
args.push("--provider", provider, "--model", model);
|
|
if (thinking) args.push("--thinking", thinking);
|
|
args.push("--session-dir", sessionDir, "--append-system-prompt", appendSystemPromptFile);
|
|
if (continueSession) args.push("--continue");
|
|
return args;
|
|
}
|
|
|
|
export function assistantText(message) {
|
|
if (!message || !Array.isArray(message.content)) return "";
|
|
return message.content
|
|
.filter((c) => c && c.type === "text" && typeof c.text === "string")
|
|
.map((c) => c.text)
|
|
.join("")
|
|
.trim();
|
|
}
|
|
|
|
// How long a turn that failed here (timeout, protocol error) may wait for
|
|
// pi's agent_start before the engine stops pi. Without a bound, a prompt pi
|
|
// accepted but never ran would hold every later prompt until restart.
|
|
export const ABORT_GRACE_MS = 30000;
|
|
|
|
export function createEngine({
|
|
command, args, cwd, env = {},
|
|
spawn = nodeSpawn,
|
|
setTimeoutImpl = globalThis.setTimeout, clearTimeoutImpl = globalThis.clearTimeout,
|
|
log = () => {},
|
|
onExit = () => {},
|
|
abortGraceMs = ABORT_GRACE_MS,
|
|
processingRoot = null, binding = null,
|
|
} = {}) {
|
|
if (typeof command !== "string" || command.length === 0) throw new DiscordError("engine: command required", 1);
|
|
if (!Array.isArray(args)) throw new DiscordError("engine: args required", 1);
|
|
|
|
// pending: prompts sent to pi, oldest first. held: prompts waiting for pi
|
|
// to settle before they are sent, oldest first.
|
|
// wedged: set when the engine gave up on pi and is stopping it. Nothing
|
|
// is sent to that child again.
|
|
const state = { child: null, buffer: "", pending: [], held: [], responses: new Map(), nextId: 1, busy: false, exited: null, wedged: false };
|
|
let managed = null;
|
|
const processing = () => {
|
|
if (processingRoot) return assertProcessing(processingRoot, managed?.generation ?? null);
|
|
};
|
|
const releaseRuntime = () => {
|
|
if (!managed) return;
|
|
try {
|
|
if (!processingStatus(processingRoot).halted && runtimeGroupState(runtime(processingRoot, managed.id)) === false)
|
|
unregisterRuntime(processingRoot, managed.id);
|
|
}
|
|
catch { log('engine: processing receipt cleanup refused'); }
|
|
};
|
|
|
|
// A turn that fails on the client side (timeout, protocol error) stays in
|
|
// the pending queue, marked done, until pi's own turn_end for it arrives.
|
|
// Otherwise that turn_end would be attributed to the next prompt. It holds
|
|
// later prompts back; if pi has not started it within abortGraceMs, the
|
|
// engine stops pi.
|
|
function failTurn(turn, code, message) {
|
|
if (turn.done) return;
|
|
turn.done = true;
|
|
if (turn.timer !== null) clearTimeoutImpl(turn.timer);
|
|
turn.timer = null;
|
|
if (state.pending.includes(turn)) {
|
|
turn.grace = setTimeoutImpl(() => {
|
|
turn.grace = null;
|
|
// A run pi started keeps its place until its agent_end or a settle.
|
|
if (state.busy) return;
|
|
// No agent_start yet. Pi may never run this prompt, or its events
|
|
// may still be on the way; with no prompt id in them, nothing sent
|
|
// now could be told apart from it. Stop pi: held prompts fail, and
|
|
// the exit fails the rest and reaches onExit.
|
|
log(`engine: no agent_start ${abortGraceMs} ms after a failed turn; stopping pi`);
|
|
wedge();
|
|
}, abortGraceMs);
|
|
}
|
|
turn.reject(new DiscordError(message, 1, { code }));
|
|
}
|
|
|
|
function wedge() {
|
|
if (state.wedged || state.exited !== null) return;
|
|
state.wedged = true;
|
|
for (const h of state.held.splice(0)) failTurn(h.turn, "engine-wedged", "engine stopped: pi did not start an aborted turn");
|
|
stopChild();
|
|
}
|
|
|
|
// Call when a turn leaves the pending queue.
|
|
function release(turn) {
|
|
if (turn.grace !== null) clearTimeoutImpl(turn.grace);
|
|
turn.grace = null;
|
|
}
|
|
|
|
function settleTurn(turn, value) {
|
|
if (turn.done) return;
|
|
turn.done = true;
|
|
if (turn.timer !== null) clearTimeoutImpl(turn.timer);
|
|
turn.timer = null;
|
|
turn.resolve(value);
|
|
}
|
|
|
|
function failAll(code, message) {
|
|
const pending = state.pending.splice(0);
|
|
for (const t of pending) {
|
|
release(t);
|
|
failTurn(t, code, message);
|
|
}
|
|
for (const h of state.held.splice(0)) failTurn(h.turn, code, message);
|
|
for (const [, r] of state.responses) r.reject(new DiscordError(message, 1, { code }));
|
|
state.responses.clear();
|
|
}
|
|
|
|
function handleLine(line) {
|
|
let event;
|
|
try {
|
|
event = JSON.parse(line);
|
|
} catch {
|
|
log("engine: malformed JSONL line from pi");
|
|
const head = state.pending.find((t) => !t.done);
|
|
if (head) failTurn(head, "engine-protocol", "engine emitted a malformed line during the turn");
|
|
return;
|
|
}
|
|
if (!event || typeof event !== "object") return;
|
|
if (event.type === "response") {
|
|
const waiter = event.id !== undefined ? state.responses.get(event.id) : undefined;
|
|
if (waiter) {
|
|
state.responses.delete(event.id);
|
|
if (event.success === false) waiter.reject(new DiscordError(`engine refused ${event.command}: ${event.error || "unknown error"}`, 1, { code: "engine-refused" }));
|
|
else waiter.resolve(event.data);
|
|
}
|
|
return;
|
|
}
|
|
if (event.type === "agent_start") state.busy = true;
|
|
// Tool and turn events belong to the run pi is executing, which is the
|
|
// oldest queued prompt: agent_end shifts it off, and a prompt that failed
|
|
// client-side (a timeout) stays at the front, marked done, until then.
|
|
// Events while that front prompt is done belong to the dead run and are
|
|
// dropped, so they never become another prompt's evidence.
|
|
const front = state.pending[0] || null;
|
|
const head = front && !front.done ? front : null;
|
|
if (event.type === "tool_execution_start" && head) {
|
|
head.tools.set(event.toolCallId, { name: event.toolName, startedAt: Date.now(), args: event.args || {} });
|
|
return;
|
|
}
|
|
if (event.type === "tool_execution_end" && head) {
|
|
const open = head.tools.get(event.toolCallId) || { name: event.toolName, startedAt: Date.now(), args: {} };
|
|
const d = (event.result && event.result.details) || {};
|
|
head.tools.set(event.toolCallId, {
|
|
...open, done: true,
|
|
record: {
|
|
name: event.toolName, root: d.root ?? (typeof open.args.root === "string" ? open.args.root : null),
|
|
path: d.path ?? (typeof open.args.path === "string" ? open.args.path : null),
|
|
...(d.url !== undefined || typeof open.args.url === "string" ? { url: d.url ?? open.args.url } : {}),
|
|
...(d.query !== undefined || typeof open.args.query === "string" ? { query: d.query ?? open.args.query } : {}),
|
|
...(d.status !== undefined ? { status: d.status } : {}),
|
|
...(d.hash !== undefined ? { hash: d.hash } : {}),
|
|
...(d.pushed !== undefined ? { pushed: d.pushed } : {}),
|
|
...(d.paths !== undefined ? { paths: d.paths } : {}),
|
|
...(d.requester !== undefined ? { requester: d.requester } : {}),
|
|
...(d.id !== undefined ? { id: d.id } : {}),
|
|
...(d.request !== undefined ? { request: d.request } : {}),
|
|
...(d.verb !== undefined ? { verb: d.verb } : {}),
|
|
...(d.key !== undefined ? { key: d.key } : {}),
|
|
...(d.revision !== undefined ? { revision: d.revision } : {}),
|
|
...(d.code !== undefined ? { code: d.code } : {}),
|
|
ok: event.isError ? false : d.ok !== false, reason: d.reason ?? (event.isError ? "tool error" : null),
|
|
bytes: d.bytes ?? null, ms: d.ms ?? Date.now() - open.startedAt,
|
|
},
|
|
});
|
|
return;
|
|
}
|
|
if (event.type === "turn_end" && head) {
|
|
head.turns += 1;
|
|
head.last = event.message || head.last;
|
|
return;
|
|
}
|
|
if (event.type === "agent_end") {
|
|
if (event.willRetry === true) return;
|
|
// Attribute the run to the head even if it failed client-side, so the
|
|
// next prompt's agent_end is not taken for this one.
|
|
const run = state.pending.shift();
|
|
if (run) release(run);
|
|
if (!run || run.done) return;
|
|
const messages = Array.isArray(event.messages) ? event.messages.filter((m) => m && m.role === "assistant") : [];
|
|
const message = messages.length > 0 ? messages[messages.length - 1] : run.last;
|
|
const tools = [...run.tools.values()].filter((t) => t.done).map((t) => t.record);
|
|
const text = assistantText(message);
|
|
const stopReason = message && message.stopReason;
|
|
if (stopReason === "error" || stopReason === "aborted") {
|
|
failTurn(run, `engine-${stopReason}`, `engine turn ended with ${stopReason}: ${(message && message.errorMessage) || ""}`.trim());
|
|
return;
|
|
}
|
|
settleTurn(run, { text, message, tools, turns: run.turns, usage: (message && message.usage) || null, model: message ? message.model : null, provider: message ? message.provider : null });
|
|
return;
|
|
}
|
|
if (event.type === "agent_settled") {
|
|
state.busy = false;
|
|
// A settle means pi has nothing queued. A turn that was accepted before
|
|
// this settle and still has no agent_end will never get one: fail it now
|
|
// instead of waiting for its timeout. Turns whose prompt response has
|
|
// not arrived yet belong to a later run and stay.
|
|
const dropped = [];
|
|
const keep = [];
|
|
for (const t of state.pending) (t.done || t.accepted ? dropped : keep).push(t);
|
|
state.pending = keep;
|
|
for (const t of dropped) {
|
|
release(t);
|
|
failTurn(t, "engine-settled-without-turn", "engine settled without answering this prompt");
|
|
}
|
|
sendHeld();
|
|
}
|
|
}
|
|
|
|
// Send one prompt to pi and queue it as pending. Only called when pi is
|
|
// idle from our point of view, so no streamingBehavior is ever needed.
|
|
function send(turn, command) {
|
|
state.pending.push(turn);
|
|
request(command).then(() => {
|
|
turn.accepted = true;
|
|
}, (err) => {
|
|
// Never accepted: pi will not emit a turn_end for it, so remove it.
|
|
const i = state.pending.indexOf(turn);
|
|
if (i !== -1) state.pending.splice(i, 1);
|
|
release(turn);
|
|
failTurn(turn, (err.details && err.details.code) || "engine-refused", err.message);
|
|
sendHeld();
|
|
});
|
|
}
|
|
|
|
// Pi is busy from our side while any sent prompt is still queued, even one
|
|
// that already failed here: a turn that timed out before its agent_start
|
|
// was read leaves state.busy false while pi runs it, and sending then would
|
|
// be refused as streaming. It leaves the queue on its agent_end, on a
|
|
// settle, on a refused send, or at process exit.
|
|
const engineBusy = () => state.busy || state.pending.length > 0;
|
|
|
|
// After a settle (or a refused send) the oldest held prompt goes out.
|
|
function sendHeld() {
|
|
if (state.exited !== null || state.wedged) return;
|
|
if (engineBusy()) return;
|
|
const next = state.held.shift();
|
|
if (next) send(next.turn, next.command);
|
|
}
|
|
|
|
function write(command) {
|
|
if (command.type === 'prompt') processing();
|
|
if (!state.child || state.exited !== null || state.wedged) throw new DiscordError("engine is not running", 1, { code: "engine-down" });
|
|
state.child.stdin.write(JSON.stringify(command) + "\n");
|
|
}
|
|
|
|
function request(command) {
|
|
const id = `r${state.nextId++}`;
|
|
return new Promise((resolve, reject) => {
|
|
state.responses.set(id, { resolve, reject });
|
|
try {
|
|
write({ ...command, id });
|
|
} catch (err) {
|
|
state.responses.delete(id);
|
|
reject(err);
|
|
}
|
|
});
|
|
}
|
|
|
|
function stopChild({ graceMs = 5000 } = {}) {
|
|
const child = state.child;
|
|
if (!child || (state.exited !== null && !managed)) return Promise.resolve(state.exited);
|
|
if (managed) {
|
|
let record;
|
|
try { record = runtime(processingRoot, managed.id); } catch { return Promise.resolve(state.exited); }
|
|
const signal = createRuntimeSignaller(record, members => updateRuntime(processingRoot, managed.id, { members }));
|
|
signal('SIGTERM');
|
|
const began = Date.now();
|
|
return new Promise(resolve => {
|
|
let killed = false;
|
|
const poll = setInterval(() => {
|
|
let live;
|
|
try { live = runtimeGroupState(runtime(processingRoot, managed.id)); } catch { live = false; }
|
|
if (live === false || Date.now() - began > graceMs + 2000) {
|
|
clearInterval(poll); releaseRuntime(); resolve(state.exited); return;
|
|
}
|
|
if (!killed && Date.now() - began >= graceMs) { signal('SIGKILL'); killed = true; }
|
|
}, 20);
|
|
});
|
|
}
|
|
return new Promise((resolve) => {
|
|
const timer = setTimeoutImpl(() => {
|
|
try {
|
|
child.kill("SIGKILL");
|
|
} catch {
|
|
// already gone
|
|
}
|
|
}, graceMs);
|
|
child.once("exit", () => {
|
|
clearTimeoutImpl(timer);
|
|
resolve(state.exited);
|
|
});
|
|
try {
|
|
child.stdin.end();
|
|
child.kill("SIGTERM");
|
|
} catch {
|
|
// already gone
|
|
}
|
|
});
|
|
}
|
|
|
|
return {
|
|
start() {
|
|
if (state.child) throw new DiscordError("engine already started", 1);
|
|
processing();
|
|
if (processingRoot) managed = registerRuntime(processingRoot, { kind: 'discord', business: 'instance', run: binding });
|
|
let child;
|
|
try {
|
|
processing();
|
|
child = spawn(command, args, { cwd, env, stdio: ["pipe", "pipe", "pipe"], ...(processingRoot ? { detached: true } : {}) });
|
|
if (managed) {
|
|
const identity = processIdentity(child.pid);
|
|
if (!identity) throw new DiscordError('engine process identity unavailable', 1);
|
|
updateRuntime(processingRoot, managed.id, { ...identity, group: true, phase: 'running' });
|
|
processing();
|
|
}
|
|
} catch (e) {
|
|
if (child?.pid) child.kill('SIGKILL');
|
|
releaseRuntime();
|
|
throw e;
|
|
}
|
|
state.child = child;
|
|
child.stdout.setEncoding("utf8");
|
|
child.stdout.on("data", (chunk) => {
|
|
state.buffer += chunk;
|
|
let idx;
|
|
while ((idx = state.buffer.indexOf("\n")) !== -1) {
|
|
let line = state.buffer.slice(0, idx);
|
|
state.buffer = state.buffer.slice(idx + 1);
|
|
if (line.endsWith("\r")) line = line.slice(0, -1);
|
|
if (line.length > 0) handleLine(line);
|
|
}
|
|
});
|
|
child.stderr.setEncoding("utf8");
|
|
child.stderr.on("data", (chunk) => log(`pi: ${chunk.trimEnd()}`));
|
|
child.on("error", (err) => {
|
|
releaseRuntime();
|
|
log(`engine spawn error: ${err.message}`);
|
|
state.exited = { code: null, signal: null, error: err.message };
|
|
failAll("engine-down", `engine failed: ${err.message}`);
|
|
onExit(state.exited);
|
|
});
|
|
child.on("exit", (code, signal) => {
|
|
releaseRuntime();
|
|
state.exited = { code, signal };
|
|
failAll("engine-down", `engine exited (code ${code}, signal ${signal})`);
|
|
onExit(state.exited);
|
|
});
|
|
return child;
|
|
},
|
|
|
|
// Resolves {text, message, tools, turns, usage, model, provider}. Rejects
|
|
// with DiscordError carrying details.code for the turn record.
|
|
prompt(text, { timeoutMs = 180000 } = {}) {
|
|
try { processing(); } catch (e) { return Promise.reject(e); }
|
|
if (typeof text !== "string" || text.length === 0) throw new DiscordError("prompt text required", 1);
|
|
const turn = { resolve: null, reject: null, timer: null, grace: null, done: false, accepted: false, tools: new Map(), turns: 0, last: null };
|
|
const done = new Promise((resolve, reject) => {
|
|
turn.resolve = resolve;
|
|
turn.reject = reject;
|
|
});
|
|
const command = { type: "prompt", message: text };
|
|
// The clock starts on arrival, held time included: a message that
|
|
// waits behind a long turn still fails after timeoutMs, and a held
|
|
// turn that times out is simply never sent.
|
|
turn.timer = setTimeoutImpl(() => {
|
|
if (turn.done) return;
|
|
const heldAt = state.held.findIndex((h) => h.turn === turn);
|
|
if (heldAt !== -1) {
|
|
state.held.splice(heldAt, 1);
|
|
failTurn(turn, "timeout", `turn timed out after ${timeoutMs} ms while waiting for the engine`);
|
|
return;
|
|
}
|
|
log(`engine: turn timed out after ${timeoutMs} ms, aborting`);
|
|
try {
|
|
write({ type: "abort" });
|
|
} catch (err) {
|
|
log(`engine: abort failed: ${err.message}`);
|
|
}
|
|
failTurn(turn, "timeout", `turn timed out after ${timeoutMs} ms`);
|
|
}, timeoutMs);
|
|
if (state.exited !== null || state.wedged) {
|
|
failTurn(turn, "engine-down", "engine is not running");
|
|
return done;
|
|
}
|
|
if (engineBusy() || state.held.length > 0) state.held.push({ turn, command });
|
|
else send(turn, command);
|
|
return done;
|
|
},
|
|
|
|
get busy() {
|
|
return engineBusy() || state.held.length > 0;
|
|
},
|
|
get pendingCount() {
|
|
return state.pending.filter((t) => !t.done).length + state.held.length;
|
|
},
|
|
|
|
stop(options) {
|
|
return stopChild(options);
|
|
},
|
|
};
|
|
}
|