mosaicstack/stack
25
0
Fork 1
Code Issues 340 Pull Requests 13 Actions Packages Projects Releases 15 Wiki Activity

fix(discord): engine tests stop in finally; a turn pi never started stops pi instead of guessing (#1509)

Browse Source 
Every engine test stops its engine in finally, and commands() tolerates a
log that doesn't exist yet, so a failed assertion no longer leaks a fake pi
and hangs the six-package union. A turn that failed client-side stays at
the front of the queue and holds the next prompt. If pi has sent no
agent_start ABORT_GRACE_MS (30 s) after the failure, the engine marks
itself wedged, fails held prompts with engine-wedged, refuses new ones with
engine-down, and stops pi. The exit reaches onExit, the connector exits 1,
and the unit restarts it. Pi's events carry no prompt id, so R1's approach,
dropping the turn and sending on, let a late run answer the next prompt.
Rocko rejected R1 and approved R2.

Tests: engine 17/17 (R1 fails 4, HEAD fails 5). Union 408/408 and the eight
suites green on the committed index. Record:
agents/darkwing/work/discord-engine-busy/.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
jason.woltje
2026-09-26 15:50:37 -05:00
co-authored by Claude Opus 5.5
parent f55b94e866
commit 3edb15eb96
11 changed files with 1843 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/discord/src/engine-pi.mjs
+98 -40
View File
@@ -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);
},
};
}