From 60327f5368f9461d402af4a7f3f6cc4bf19d418e Mon Sep 17 00:00:00 2001 From: Canitas Date: Sun, 18 Oct 2026 02:41:35 -0700 Subject: [PATCH 1/2] fix: refuse a prompt stopped while it waits to start A prompt waits for the session's startup and for a manual compaction before it owns a controller, and abort() only stopped what was already running, so a prompt stopped while it waited later ran with a fresh controller. It is now admitted, and stoppable, before its first wait, and refused when stopped; its admission becomes the run's controller. Print mode takes that refusal as the operator's stop rather than an error. Implemented by Sol (gpt-6.1-sol), who found it reading loop.ts; reproduced and checked by Claude, who added the print-mode case. --- packages/core/src/loop.ts | 22 +++-- packages/core/tests/stop-admission.test.ts | 94 ++++++++++++++++++++++ 2 files changed, 111 insertions(+), 5 deletions(-) create mode 100644 packages/core/tests/stop-admission.test.ts diff --git a/packages/core/src/loop.ts b/packages/core/src/loop.ts index 849bde764..8ff91a161 100644 --- a/packages/core/src/loop.ts +++ b/packages/core/src/loop.ts @@ -406,6 +406,8 @@ export function createLoop(options: LoopOptions): LoopSession { }, options.extensions ?? [], options.verifier); tools.restore(options.resumed?.activeTools); const started = host.sessionStart({ type: "session_start", reason: options.resumed === undefined ? "startup" : "resume" }); + /** Prompts waiting for startup or manual compaction, which a stop refuses before they become runs. */ + const admitting = new Set(); let controller: AbortController | undefined; let running: Promise | undefined; /** Settles once the running request is in the ledger, so no amendment comes before it. */ @@ -794,16 +796,24 @@ export function createLoop(options: LoopOptions): LoopSession { return { async prompt(text, { streamingBehavior, source = "interactive" } = {}) { - await started; - // Checked again after each wait, with no wait between the last check and the run's start, so a compaction - // asked for meanwhile cannot start under it. - while (compacting !== undefined) await compacting.done.catch(() => undefined); + const admission = new AbortController(); + admitting.add(admission); + try { + await started; + if (admission.signal.aborted) throw new Error(stoppedReason); + // Checked again after each wait, with no wait between the last check and the run's start, so a compaction + // asked for meanwhile cannot start under it. + while (compacting !== undefined) { + await compacting.done.catch(() => undefined); + if (admission.signal.aborted) throw new Error(stoppedReason); + } + } finally { admitting.delete(admission); } if (running !== undefined) { if (streamingBehavior === undefined) throw new Error("A prompt is already running; steer it, queue a follow-up, abort it or wait."); await amend(runs, text, streamingBehavior, source, controller?.signal); return; } - controller = new AbortController(); + controller = admission; queue = []; waiting = []; closed = false; @@ -820,6 +830,7 @@ export function createLoop(options: LoopOptions): LoopSession { } }, async abort() { + for (const admission of admitting) admission.abort(); controller?.abort(); compacting?.controller.abort(); await running?.catch(() => undefined); @@ -848,6 +859,7 @@ export function createLoop(options: LoopOptions): LoopSession { stats: () => ({ tokens: { ...tokens } }), outcome: () => outcome, dispose: () => disposed ??= (async () => { + for (const admission of admitting) admission.abort(); controller?.abort(); compacting?.controller.abort(); await running?.catch(() => undefined); diff --git a/packages/core/tests/stop-admission.test.ts b/packages/core/tests/stop-admission.test.ts new file mode 100644 index 000000000..c7987d890 --- /dev/null +++ b/packages/core/tests/stop-admission.test.ts @@ -0,0 +1,94 @@ +import { createModels, fauxAssistantMessage, fauxProvider, type FauxResponseStep } from "@earendil-works/pi-ai"; +import { afterEach, expect, test } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createSession, type EngineSession, type SessionOptions } from "../src/session.ts"; +import { accepts, scriptedReply } from "./scripted.ts"; + +const open: EngineSession[] = []; +afterEach(async () => { await Promise.all(open.splice(0).map((session) => session.dispose())); }); + +function session(responses: readonly FauxResponseStep[], options: Partial = {}): EngineSession { + const faux = fauxProvider(); + faux.setResponses([...responses]); + const models = createModels(); + models.setProvider(faux.provider); + const created = createSession({ cwd: mkdtempSync(join(tmpdir(), "tesota-core-admission-")), + folder: mkdtempSync(join(tmpdir(), "tesota-core-sessions-")), models, model: faux.getModel(), verifier: accepts, ...options }); + open.push(created); + return created; +} + +test("a prompt stopped while waiting for startup is refused before it runs; a later prompt runs", async () => { + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const inputs: string[] = []; + const created = session([ + scriptedReply({ tool: "write", args: { path: "ran.md", content: "ran" } }), scriptedReply("done"), + ], { extensions: [(pi) => { + pi.on("session_start", async () => { entered.resolve(); await release.promise; }); + pi.on("input", (event) => { inputs.push(event.text); }); + }] }); + const pending = created.prompt("go").catch((error: unknown) => error); + try { + await entered.promise; + await created.abort(); + } finally { release.resolve(); } + const refused = await pending; + expect(existsSync(join(created.workspace.cwd, "ran.md"))).toBe(false); + expect(refused).toMatchObject({ message: "the operator stopped the run" }); + expect(inputs).toEqual([]); + expect(created.messages()).toEqual([]); + expect(created.ledger()).toEqual([]); + expect(readFileSync(created.file, "utf8").trimEnd().split("\n")).toHaveLength(1); + expect(created.outcome()).toBeUndefined(); + + await created.prompt("after the stop"); + expect(inputs).toEqual(["after the stop"]); + expect(readFileSync(join(created.workspace.cwd, "ran.md"), "utf8")).toBe("ran"); + expect(created.ledger().map(({ run, text }) => ({ run, text }))).toEqual([{ run: 1, text: "after the stop" }]); + expect(created.outcome()).toBe("completed"); +}); + +test("a prompt stopped while waiting for manual compaction is refused; a prompt after the stop runs", async () => { + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const inputs: string[] = []; + const created = session([ + ...Array.from({ length: 3 }, (_, k) => scriptedReply({ tool: "write", args: { path: `note-${k}.txt`, content: "note ".repeat(300) } })), + scriptedReply("done"), + async () => { entered.resolve(); await release.promise; return fauxAssistantMessage("## Goal\nwrite notes"); }, + scriptedReply({ tool: "write", args: { path: "ran.md", content: "ran" } }), scriptedReply("done"), + ], { compaction: { reserve: 1_000, keep: 600 }, extensions: [(pi) => { + pi.on("input", (event) => { inputs.push(event.text); }); + }] }); + await created.prompt("write notes"); + const compacting = created.compact().catch((error: unknown) => error); + await entered.promise; + const before = readFileSync(created.file, "utf8"); + const pending = created.prompt("go").catch((error: unknown) => error); + // Startup has finished; let the prompt enter its compaction wait before stopping it. + await Promise.resolve(); + const stopping = created.abort(); + release.resolve(); + await stopping; + expect(await compacting).toMatchObject({ message: "the operator stopped the run" }); + const refused = await pending; + expect(existsSync(join(created.workspace.cwd, "ran.md"))).toBe(false); + expect(refused).toMatchObject({ message: "the operator stopped the run" }); + expect(inputs).toEqual(["write notes"]); + expect(created.ledger().map(({ text }) => text)).toEqual(["write notes"]); + // The stopped summary may record its usage, but the refused prompt adds no run to the file. + expect(readFileSync(created.file, "utf8").slice(before.length).trimEnd().split("\n") + .map((line) => (JSON.parse(line) as { type: string }).type)).toEqual(["summary_usage"]); + expect(created.outcome()).toBe("completed"); + + await created.prompt("after the stop"); + expect(inputs).toEqual(["write notes", "after the stop"]); + expect(readFileSync(join(created.workspace.cwd, "ran.md"), "utf8")).toBe("ran"); + expect(created.ledger().map(({ run, text }) => ({ run, text }))).toEqual([ + { run: 1, text: "write notes" }, { run: 2, text: "after the stop" }, + ]); + expect(created.outcome()).toBe("completed"); +}); From 9612a1ae2a35a367281518ddefae31813a8b09f3 Mon Sep 17 00:00:00 2001 From: Canitas Date: Sun, 18 Oct 2026 02:42:10 -0700 Subject: [PATCH 2/2] fix: take a prompt refused for the stop as the stop in print mode With a prompt stopped before it starts now refused, print mode would print that refusal with its stack after a Ctrl+C or a termination. It ends without a run and exits 1, as a stop does. --- packages/core/src/print.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/packages/core/src/print.ts b/packages/core/src/print.ts index 1779f3f1d..da1c16c70 100644 --- a/packages/core/src/print.ts +++ b/packages/core/src/print.ts @@ -109,7 +109,8 @@ export async function runPrint(args: readonly string[], cwd: string, write: Prin process.once("SIGTERM", stop); try { servers = await serversFor(session, mcp); - await session.prompt(asked.request); + // A request stopped before it began is refused rather than run; that is the stop, not a failure to report. + await session.prompt(asked.request).catch((error: unknown) => { if (!stopped) throw error; }); write.out(`${session.lastAssistantText() ?? ""}\n`); } finally { process.off("SIGINT", stop);