diff --git a/packages/discord/src/engine-pi.mjs b/packages/discord/src/engine-pi.mjs index 5c8fd0a9..46ef1f88 100644 --- a/packages/discord/src/engine-pi.mjs +++ b/packages/discord/src/engine-pi.mjs @@ -16,9 +16,14 @@ // 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. 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`. +// 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. @@ -64,31 +69,67 @@ export function assistantText(message) { .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, } = {}) { 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. - const state = { child: null, buffer: "", pending: [], held: [], responses: new Map(), nextId: 1, busy: false, exited: null }; + // 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 }; // 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. + // 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; @@ -99,7 +140,10 @@ export function createEngine({ function failAll(code, message) { const pending = state.pending.splice(0); - for (const t of pending) failTurn(t, code, message); + 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(); @@ -174,6 +218,7 @@ export function createEngine({ // 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; @@ -193,13 +238,14 @@ export function createEngine({ // 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) { - if (t.done) continue; - if (t.accepted) failTurn(t, "engine-settled-without-turn", "engine settled without answering this prompt"); - else keep.push(t); - } + 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(); } } @@ -214,21 +260,29 @@ export function createEngine({ // 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) return; - if (state.busy || state.pending.some((t) => !t.done)) return; + 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 (!state.child || state.exited !== null) throw new DiscordError("engine is not running", 1, { code: "engine-down" }); + 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"); } @@ -245,6 +299,30 @@ export function createEngine({ }); } + function stopChild({ graceMs = 5000 } = {}) { + const child = state.child; + if (!child || state.exited !== null) return Promise.resolve(state.exited); + 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); @@ -281,7 +359,7 @@ export function createEngine({ // with DiscordError carrying details.code for the turn record. prompt(text, { timeoutMs = 180000 } = {}) { if (typeof text !== "string" || text.length === 0) throw new DiscordError("prompt text required", 1); - const turn = { resolve: null, reject: null, timer: null, done: false, accepted: false, tools: new Map(), turns: 0, last: null }; + 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; @@ -306,44 +384,24 @@ export function createEngine({ } failTurn(turn, "timeout", `turn timed out after ${timeoutMs} ms`); }, timeoutMs); - if (state.exited !== null) { + if (state.exited !== null || state.wedged) { failTurn(turn, "engine-down", "engine is not running"); return done; } - if (state.busy || state.pending.some((t) => !t.done) || state.held.length > 0) state.held.push({ turn, command }); + if (engineBusy() || state.held.length > 0) state.held.push({ turn, command }); else send(turn, command); return done; }, get busy() { - return state.busy || state.pending.some((t) => !t.done) || state.held.length > 0; + return engineBusy() || state.held.length > 0; }, get pendingCount() { return state.pending.filter((t) => !t.done).length + state.held.length; }, - stop({ graceMs = 5000 } = {}) { - const child = state.child; - if (!child || state.exited !== null) return Promise.resolve(state.exited); - 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 - } - }); + stop(options) { + return stopChild(options); }, }; } diff --git a/packages/discord/tests/engine.test.mjs b/packages/discord/tests/engine.test.mjs index 59674d1e..f3bd7525 100644 --- a/packages/discord/tests/engine.test.mjs +++ b/packages/discord/tests/engine.test.mjs @@ -4,6 +4,8 @@ import { readFileSync } from "node:fs"; import { join } from "node:path"; import { createEngine, buildPiArgs, PI_FIXED_ARGS, TOOLS_EXTENSION, READONLY_TOOLS_EXTENSION, assistantText } from "../src/engine-pi.mjs"; import { existsSync } from "node:fs"; +import { EventEmitter } from "node:events"; +import { PassThrough } from "node:stream"; import { makeRoot } from "./helpers.mjs"; const fakePi = join(import.meta.dirname, "fake-pi.mjs"); @@ -28,7 +30,20 @@ function start(root, extra = {}) { log: (m) => logs.push(m), ...extra, }); engine.start(); - return { engine, logs, commands: () => readFileSync(logPath, "utf8").trim().split("\n").filter(Boolean).map((l) => JSON.parse(l)) }; + // The fake creates its log on the first command; until then there are none. + const commands = () => (existsSync(logPath) ? readFileSync(logPath, "utf8").trim().split("\n").filter(Boolean).map((l) => JSON.parse(l)) : []); + return { engine, logs, commands }; +} + +// Every test stops its engine in finally: a fake pi left running after a +// failed assertion keeps the test file from exiting. +async function withEngine(extra, body) { + const started = start(makeRoot(), extra); + try { + await body(started); + } finally { + await started.engine.stop(); + } } test("engine: buildPiArgs carries the fixed flags, engine settings, session dir and prompt file", () => { @@ -56,8 +71,7 @@ test("engine: with tools, buildPiArgs turns pi's own tools off, loads the extens assert.equal(rw[rw.indexOf("--tools") + 1], "list_dir,read_file,search,write_file,edit_file", "a writable root adds exactly the two write tools"); }); -test("engine: a run with tool turns settles once, on the answer, with every tool call in the result", async () => { - const { engine } = start(makeRoot()); +test("engine: a run with tool turns settles once, on the answer, with every tool call in the result", () => withEngine({}, async ({ engine }) => { const r = await engine.prompt("tools 3"); assert.equal(r.text, "read 3 file(s)"); assert.equal(r.turns, 2); @@ -71,45 +85,38 @@ test("engine: a run with tool turns settles once, on the answer, with every tool assert.equal(plain.turns, 1); await idle(engine); assert.equal(engine.busy, false); - await engine.stop(); -}); +})); -test("engine: a run that ends on a tool-only turn fails the prompt as empty; a retried run settles on the real end", async () => { - const { engine } = start(makeRoot()); +test("engine: a run that ends on a tool-only turn fails the prompt as empty; a retried run settles on the real end", () => withEngine({}, async ({ engine }) => { const r = await engine.prompt("toolonly"); assert.equal(r.text, "", "no text: the connector turns this into engine-empty"); assert.equal(r.tools.length, 1); const again = await engine.prompt("retry"); assert.equal(again.text, "after retry"); - await engine.stop(); -}); +})); -test("engine: one prompt, one turn, text and usage come back", async () => { - const { engine } = start(makeRoot()); - try { - const r = await engine.prompt("hello"); - assert.equal(r.text, "echo: hello"); - assert.deepEqual(r.usage, { input: 3, output: 2 }); - await idle(engine); - assert.equal(engine.busy, false); - } finally { - await engine.stop(); - } -}); +test("engine: one prompt, one turn, text and usage come back", () => withEngine({}, async ({ engine }) => { + const r = await engine.prompt("hello"); + assert.equal(r.text, "echo: hello"); + assert.deepEqual(r.usage, { input: 3, output: 2 }); + await idle(engine); + assert.equal(engine.busy, false); +})); -test("engine: a prompt while streaming is held until pi settles, then sent as its own run, and answered in order", async () => { - const { engine, commands } = start(makeRoot()); - const first = engine.prompt("slow 150"); - await new Promise((r) => setTimeout(r, 20)); +test("engine: a prompt while streaming is held until pi settles, then sent as its own run, and answered in order", () => withEngine({}, async ({ engine, commands }) => { + const first = engine.prompt("slow 300"); assert.equal(engine.busy, true); const second = engine.prompt("second"); assert.equal(engine.pendingCount, 2); - await new Promise((r) => setTimeout(r, 20)); - assert.equal(commands().filter((c) => c.type === "prompt").length, 1, "the second prompt is not sent while pi is busy"); + const prompted = () => commands().filter((c) => c.type === "prompt"); + assert.ok(await until(() => prompted().length > 0), "the first prompt reached pi"); + assert.equal(prompted().length, 1, "the second prompt is not sent while pi is busy"); + // The fake refuses a prompt without streamingBehavior while it runs one, so + // an answered second prompt also proves it was not sent early. const [r1, r2] = await Promise.all([first, second]); assert.equal(r1.text, "slow reply"); assert.equal(r2.text, "echo: second"); - const prompts = commands().filter((c) => c.type === "prompt"); + const prompts = prompted(); assert.equal(prompts.length, 2); // Never a pi follow-up: pi would fold it into the first run and close both // answers with one agent_end (the live loss of 2026-09-17). @@ -117,11 +124,9 @@ test("engine: a prompt while streaming is held until pi settles, then sent as it assert.equal(prompts[1].streamingBehavior, undefined); await idle(engine); assert.equal(engine.busy, false); - await engine.stop(); -}); +})); -test("engine: a held prompt that times out before pi settles fails on its own and is never sent", async () => { - const { engine, commands } = start(makeRoot()); +test("engine: a held prompt that times out before pi settles fails on its own and is never sent", () => withEngine({}, async ({ engine, commands }) => { const first = engine.prompt("slow 200"); await new Promise((r) => setTimeout(r, 20)); await assert.rejects(engine.prompt("late one", { timeoutMs: 50 }), (e) => e.details.code === "timeout" && /waiting for the engine/.test(e.message)); @@ -130,50 +135,177 @@ test("engine: a held prompt that times out before pi settles fails on its own an await idle(engine); assert.deepEqual(commands().filter((c) => c.type === "prompt").map((c) => c.message), ["slow 200"]); assert.deepEqual(commands().filter((c) => c.type === "abort"), [], "a held turn is not aborted; pi never had it"); - await engine.stop(); -}); +})); -test("engine: timeout sends abort and fails only that turn; the process stays", async () => { - const { engine, commands, logs } = start(makeRoot()); +test("engine: timeout sends abort and fails only that turn; the process stays", () => withEngine({}, async ({ engine, commands, logs }) => { await assert.rejects(engine.prompt("slow 5000", { timeoutMs: 100 }), (err) => err.details.code === "timeout"); assert.ok(await until(() => commands().some((c) => c.type === "abort")), "abort reached pi"); assert.ok(logs.some((l) => /timed out/.test(l))); const r = await engine.prompt("again"); assert.equal(r.text, "echo: again"); - await engine.stop(); +})); + +test("engine: tool events from a run that outlived its timeout never land in the next prompt's record", () => withEngine({}, async ({ engine }) => { + await assert.rejects(engine.prompt("late 200", { timeoutMs: 40 }), (err) => err.details.code === "timeout"); + const r = await engine.prompt("after late"); + assert.equal(r.text, "echo: after late"); + assert.deepEqual(r.tools, [], "the dead run's read is not this prompt's evidence"); + assert.equal(r.turns, 1, "the dead run's turns are not counted here"); +})); + +// The turn timer is fired by hand, before the engine has read any event from +// pi, so the timed-out run is still pi's and state.busy is still false when +// the next prompt arrives. Under load a real timer does the same. +const TURN_MS = 60000; +const manualTurnTimer = (fire) => ({ + setTimeoutImpl: (fn, ms) => (ms === TURN_MS ? fire.push(fn) : setTimeout(fn, ms)), + clearTimeoutImpl: (id) => { if (typeof id !== "number") clearTimeout(id); }, }); -test("engine: tool events from a run that outlived its timeout never land in the next prompt's record", async () => { - const { engine } = start(makeRoot()); - try { - await assert.rejects(engine.prompt("late 200", { timeoutMs: 40 }), (err) => err.details.code === "timeout"); - const r = await engine.prompt("after late"); +test("engine: a prompt after a turn that timed out before its agent_start waits for pi to settle instead of being refused", () => { + const fire = []; + return withEngine(manualTurnTimer(fire), async ({ engine, commands }) => { + const late = engine.prompt("late 100", { timeoutMs: TURN_MS }); + fire.shift()(); + assert.equal(engine.busy, true, "pi is still running the prompt that timed out"); + const next = engine.prompt("after late", { timeoutMs: 5000 }); + assert.equal(engine.pendingCount, 1, "only the new prompt is live"); + await assert.rejects(late, (err) => err.details.code === "timeout"); + const r = await next; assert.equal(r.text, "echo: after late"); assert.deepEqual(r.tools, [], "the dead run's read is not this prompt's evidence"); - assert.equal(r.turns, 1, "the dead run's turns are not counted here"); - } finally { - await engine.stop(); - } + assert.equal(r.turns, 1); + assert.deepEqual(commands().map((c) => (c.type === "prompt" ? c.message : c.type)), ["late 100", "abort", "after late"]); + await idle(engine); + assert.equal(engine.busy, false); + }); +}); + +// "mute" is accepted and never run, so no agent_start, agent_end or settle +// ever comes for it. Unbounded, it would hold every later prompt. +test("engine: when pi has not started a timed-out turn by the end of the abort grace, the engine stops pi and fails held prompts", async () => { + let exited = null; + await withEngine({ abortGraceMs: 150, onExit: (e) => (exited = e) }, async ({ engine, commands, logs }) => { + await assert.rejects(engine.prompt("mute", { timeoutMs: 50 }), (err) => err.details.code === "timeout"); + assert.equal(engine.busy, true, "pi might still be running it"); + const started = Date.now(); + await assert.rejects(engine.prompt("after mute", { timeoutMs: 5000 }), (err) => err.details.code === "engine-wedged"); + assert.ok(Date.now() - started >= 100, "held for the grace, not failed at once"); + await assert.rejects(engine.prompt("later"), (err) => err.details.code === "engine-down"); + assert.ok(await until(() => exited !== null), "pi exits and onExit hears of it"); + assert.ok(logs.some((l) => /stopping pi/.test(l))); + assert.deepEqual(commands().map((c) => (c.type === "prompt" ? c.message : c.type)), ["mute", "abort"]); + }); +}); + +// Rocko's 6b R1 case: pi is stuck before agent_start, then runs the old +// prompt and only afterwards reads the next one. The events carry no prompt +// id, so a prompt sent after the grace would get the old run's answer. +test("engine: a timed-out turn pi starts only after the grace never answers a later prompt", async () => { + let exited = null; + await withEngine({ abortGraceMs: 150, onExit: (e) => (exited = e) }, async ({ engine, commands }) => { + await assert.rejects(engine.prompt("stall 400", { timeoutMs: 50 }), (err) => err.details.code === "timeout"); + await assert.rejects(engine.prompt("after stall", { timeoutMs: 5000 }), (err) => err.details.code === "engine-wedged"); + assert.ok(await until(() => exited !== null), "pi exits and onExit hears of it"); + await new Promise((res) => setTimeout(res, 400)); + assert.ok(!commands().some((c) => c.message === "after stall"), "nothing was sent after the grace"); + }); }); -test("engine: a malformed JSONL line fails the turn, not the process", async () => { - const { engine, logs } = start(makeRoot()); +// The same case in memory, after Rocko's reproducer: the old run's events +// arrive after the grace while pi is still exiting. They land on the failed +// turn, nothing more is written to pi, and only the exit ends the engine. +// Pi's response to the old prompt comes either before its timeout or only +// with the late events. +for (const lateResponse of [false, true]) test(`engine: late events of a run past its grace, before pi exits, answer nothing and nothing more is sent (${lateResponse ? "late" : "early"} prompt response)`, async () => { + const timers = []; + const written = []; + const kills = []; + const child = new EventEmitter(); + child.stdout = new PassThrough(); + child.stderr = new PassThrough(); + child.stdin = { write: (s) => { written.push(JSON.parse(s)); return true; }, end: () => {} }; + child.kill = (signal) => { kills.push(signal); return true; }; + let exited = null; + const engine = createEngine({ + command: "memory-only", args: [], spawn: () => child, abortGraceMs: 150, onExit: (e) => (exited = e), + setTimeoutImpl: (fn, ms) => { const t = { fn, ms, active: true }; timers.push(t); return t; }, + clearTimeoutImpl: (t) => { t.active = false; }, + }); + const emit = (x) => child.stdout.write(JSON.stringify(x) + "\n"); + const fire = (ms) => { const t = timers.find((x) => x.ms === ms && x.active); assert.ok(t, `timer ${ms}`); t.active = false; t.fn(); }; + const message = (text) => ({ role: "assistant", content: [{ type: "text", text }], stopReason: "stop" }); + const tick = () => new Promise((res) => setImmediate(res)); + // Checked after a tick instead of awaited, so a regression fails here + // rather than hanging on a promise nothing will settle. + const outcome = (p) => { + const o = { state: "pending", code: null, text: null }; + p.then((v) => Object.assign(o, { state: "resolved", text: v.text }), (e) => Object.assign(o, { state: "rejected", code: e.details && e.details.code })); + return o; + }; + engine.start(); + const first = engine.prompt("old", { timeoutMs: 50 }); + const accept = () => emit({ type: "response", id: written[0].id, command: "prompt", success: true }); + if (!lateResponse) accept(); + await tick(); + fire(50); + await assert.rejects(first, (err) => err.details.code === "timeout"); + const next = outcome(engine.prompt("new", { timeoutMs: 2000 })); + fire(150); + await tick(); + assert.deepEqual(next, { state: "rejected", code: "engine-wedged", text: null }); + assert.deepEqual(kills, ["SIGTERM"]); + if (lateResponse) accept(); + emit({ type: "agent_start" }); + emit({ type: "tool_execution_start", toolCallId: "old-call", toolName: "read_file", args: { root: "docs", path: "old.md" } }); + emit({ type: "tool_execution_end", toolCallId: "old-call", toolName: "read_file", result: { details: { root: "docs", path: "old.md", ok: true } } }); + emit({ type: "turn_end", message: message("OLD RUN ANSWER") }); + emit({ type: "agent_end", messages: [message("OLD RUN ANSWER")] }); + emit({ type: "agent_settled" }); + await tick(); + const after = outcome(engine.prompt("after settle", { timeoutMs: 2000 })); + await tick(); + assert.deepEqual(after, { state: "rejected", code: "engine-down", text: null }); + assert.deepEqual(next, { state: "rejected", code: "engine-wedged", text: null }, "the old answer did not reach the new prompt"); + assert.deepEqual(written.map((c) => (c.type === "prompt" ? c.message : c.type)), ["old", "abort"], "no prompt reached pi after the grace"); + assert.equal(exited, null); + fire(5000); + assert.deepEqual(kills, ["SIGTERM", "SIGKILL"]); + child.emit("exit", null, "SIGKILL"); + assert.deepEqual(exited, { code: null, signal: "SIGKILL" }); +}); + +test("engine: a timed-out run pi did start outlives the grace; the next prompt goes out when it ends", async () => { + let exited = null; + await withEngine({ abortGraceMs: 150, onExit: (e) => (exited = e) }, async ({ engine, commands }) => { + await assert.rejects(engine.prompt("late 400", { timeoutMs: 50 }), (err) => err.details.code === "timeout"); + const r = await engine.prompt("after late", { timeoutMs: 5000 }); + assert.equal(r.text, "echo: after late"); + assert.deepEqual(r.tools, []); + assert.equal(exited, null, "pi was not stopped"); + assert.deepEqual(commands().map((c) => (c.type === "prompt" ? c.message : c.type)), ["late 400", "abort", "after late"]); + }); +}); + +test("engine: a malformed JSONL line fails the turn, not the process", () => withEngine({}, async ({ engine, logs }) => { await assert.rejects(engine.prompt("garbage"), (err) => err.details.code === "engine-protocol"); assert.ok(logs.some((l) => /malformed/.test(l))); const r = await engine.prompt("still here"); assert.equal(r.text, "echo: still here"); - await engine.stop(); -}); +})); test("engine: a turn that ends in error rejects with the error code; process exit fails pending turns", async () => { - const root = makeRoot(); let exited = null; - const { engine } = start(root, { onExit: (e) => (exited = e) }); - await assert.rejects(engine.prompt("error"), (err) => err.details.code === "engine-error" && /fake provider error/.test(err.message)); - const pending = engine.prompt("slow 5000"); - await new Promise((r) => setTimeout(r, 20)); - await engine.stop(); - await assert.rejects(pending, (err) => err.details.code === "engine-down"); - assert.ok(exited); - await assert.rejects(engine.prompt("x"), /not running/); + const { engine } = start(makeRoot(), { onExit: (e) => (exited = e) }); + try { + await assert.rejects(engine.prompt("error"), (err) => err.details.code === "engine-error" && /fake provider error/.test(err.message)); + const pending = engine.prompt("slow 5000"); + await new Promise((r) => setTimeout(r, 20)); + await engine.stop(); + await assert.rejects(pending, (err) => err.details.code === "engine-down"); + assert.ok(exited); + await assert.rejects(engine.prompt("x"), /not running/); + } finally { + await engine.stop(); + } }); diff --git a/packages/discord/tests/fake-pi.mjs b/packages/discord/tests/fake-pi.mjs index 94919306..bf94c013 100644 --- a/packages/discord/tests/fake-pi.mjs +++ b/packages/discord/tests/fake-pi.mjs @@ -8,9 +8,13 @@ // then a second turn that answers "read file(s)" // "toolonly" a run whose only turn calls a tool and never answers // "retry" an agent_end with willRetry, then the real answer +// "mute" accept the prompt and emit nothing, staying idle // "late " ignore abort; after emit a tool pair and a tool turn, // then answer "late reply", like a run that outlives its // client-side timeout +// "stall " accept the prompt, then read nothing for (pi stuck +// before agent_start); then run it, answering "echo: +// stalled", and only then read what came in meanwhile // anything else answer "echo: " immediately // A prompt received while busy without streamingBehavior is refused, as pi // does. A prompt with streamingBehavior followUp is folded into the running @@ -30,6 +34,7 @@ function assistant(text, stopReason = "stop") { } function run(text) { + if (text === "mute") return; busy = true; out({ type: "agent_start" }); out({ type: "turn_start" }); @@ -109,11 +114,16 @@ function run(text) { let current = null; let buffer = ""; +let stalled = false; process.stdin.setEncoding("utf8"); process.stdin.on("data", (chunk) => { buffer += chunk; + drain(); +}); + +function drain() { let idx; - while ((idx = buffer.indexOf("\n")) !== -1) { + while (!stalled && (idx = buffer.indexOf("\n")) !== -1) { const line = buffer.slice(0, idx); buffer = buffer.slice(idx + 1); if (!line) continue; @@ -125,7 +135,15 @@ process.stdin.on("data", (chunk) => { continue; } out({ id: cmd.id, type: "response", command: "prompt", success: true }); - if (busy) queue.push(cmd.message); + const sm = /^stall (\d+)$/.exec(cmd.message); + if (sm) { + stalled = true; + setTimeout(() => { + run("stalled"); + stalled = false; + drain(); + }, Number(sm[1])); + } else if (busy) queue.push(cmd.message); else run(cmd.message); } else if (cmd.type === "abort") { out({ id: cmd.id, type: "response", command: "abort", success: true }); @@ -139,5 +157,5 @@ process.stdin.on("data", (chunk) => { out({ id: cmd.id, type: "response", command: cmd.type, success: true, data: {} }); } } -}); +} process.stdin.on("end", () => process.exit(0));