From f865343c904799d6bd08701ab489672f5f5d6209 Mon Sep 17 00:00:00 2001 From: Yaowei Zheng Date: Wed, 29 Jul 2026 16:13:02 +0800 Subject: [PATCH] fix(core): a request whose upstream ignores its AbortSignal can no longer wedge the Session (#107) Co-authored-by: Claude Opus 5 (1M context) --- packages/core/src/llm/generative-model.ts | 32 +++++++++++++++-- packages/core/test/llm.test.ts | 42 +++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/packages/core/src/llm/generative-model.ts b/packages/core/src/llm/generative-model.ts index 6407b55..79ab7ae 100644 --- a/packages/core/src/llm/generative-model.ts +++ b/packages/core/src/llm/generative-model.ts @@ -1133,6 +1133,26 @@ export class GenerativeModel implements LLMInterface { }, this.requestTimeoutMs); }; + /** + * Settles as soon as the run must stop, whatever upstream is doing: the user aborted, or + * the idle timer fired (both go through `ac`). `it.next()` is raced against this because + * an upstream that does not honour its AbortSignal leaves that promise pending **forever** + * — and once `ac` is aborted the idle timer's own `ac.abort()` is a no-op, so nothing is + * left to unwedge the loop. Observed against Kimi: pressing Stop mid-request left the + * Session running with no way to send, compact or interrupt it again, short of a restart. + * + * The pre-loop `userSignal?.aborted` check below covers the *other* half of this (aborting + * while suspended at a `yield`); it cannot help here, since the loop never gets back to it. + */ + const STOPPED = Symbol("stopped"); + const stopped: Promise = new Promise((resolve) => { + if (ac.signal.aborted) { + resolve(STOPPED); + return; + } + ac.signal.addEventListener("abort", () => resolve(STOPPED), { once: true }); + }); + // Terminal-state classification: timeout (timed out/network drop) / malformed (response // parse error) / aborted (user) / failed (other). null means it ended normally. let outcome: LLMOutcome | null = null; @@ -1159,12 +1179,20 @@ export class GenerativeModel implements LLMInterface { // time), measuring upstream idleness — this avoids a slow consumer (e.g. a slow Trace // sink) falsely triggering the timeout. armTimer(); - let res: IteratorResult; + let res: IteratorResult | typeof STOPPED; try { - res = await it.next(); + res = await Promise.race([it.next(), stopped]); } finally { clearTimer(); } + if (res === STOPPED) { + // Upstream never settled after the abort. Abandon it rather than await it: ask it to + // close (best effort — a stream that ignored the signal may ignore this too, so the + // rejection is swallowed and the promise is not awaited) and classify by trigger. + void Promise.resolve(it.return?.(undefined)).catch(() => undefined); + outcome = userSignal?.aborted ? { status: "aborted" } : { status: "timeout" }; + break; + } if (res.done) break; if (userSignal?.aborted) { outcome = { status: "aborted" }; diff --git a/packages/core/test/llm.test.ts b/packages/core/test/llm.test.ts index 054b4ad..c383e1f 100644 --- a/packages/core/test/llm.test.ts +++ b/packages/core/test/llm.test.ts @@ -1418,6 +1418,27 @@ describe("GenerativeModel.streamGenerate outcome classification (PRN-013)", () = const abortError = (): Error => Object.assign(new Error("aborted"), { name: "AbortError" }); + /** + * An upstream that IGNORES its AbortSignal: it never yields and never rejects, whatever + * happens to the signal. This is the shape that wedged real Sessions (observed with Kimi): + * the SDK's stream promise simply never settles, so `await it.next()` hangs forever and the + * idle timer cannot rescue it either, because by then `ac` is already aborted and its + * `ac.abort()` is a no-op. + */ + async function* deaf(): AsyncGenerator { + await new Promise(() => {}); // never settles, never observes the signal + } + + /** Fails the test loudly instead of letting a regression hang the whole suite. */ + function withTimeout(p: Promise, ms: number): Promise { + return Promise.race([ + p, + new Promise((_, reject) => + setTimeout(() => reject(new Error(`streamGenerate did not settle within ${ms}ms`)), ms), + ), + ]); + } + // Never yields any event, and only ends with an AbortError once the signal aborts // (simulates idle/hanging). async function* hang(signal: AbortSignal): AsyncGenerator { @@ -1584,6 +1605,27 @@ describe("GenerativeModel.streamGenerate outcome classification (PRN-013)", () = expect(outcome2.status).toBe("malformed"); }); + it("a user abort ends the run even when upstream ignores the signal and never settles", async () => { + const model = new SeamModel(() => deaf()); + const controller = new AbortController(); + setTimeout(() => controller.abort(), 20); // the user presses Stop mid-request + // Without the race this never settles and the Session stays "running" forever. + const { outcome } = await withTimeout( + drain(model.streamGenerate({ newMessages: [userText("go")], signal: controller.signal })), + 2000, + ); + expect(outcome.status).toBe("aborted"); + }); + + it("the idle timeout still ends the run when upstream ignores the signal", async () => { + const model = new SeamModel(() => deaf(), 50); // 50ms idle budget + const { outcome } = await withTimeout( + drain(model.streamGenerate({ newMessages: [userText("go")] })), + 2000, + ); + expect(outcome.status).toBe("timeout"); + }); + it("classifies a credentials failure as its own terminal status auth; params stay failed", async () => { const model = new SeamModel(() => authError()); const { messages, outcome } = await drain(