diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 1e16820e7..9ee134278 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed Mistral streams being aborted mid-generation (e.g. during long thinking) after the request timeout; the timeout now applies only to waiting for response headers ([#10609](https://github.com/earendil-works/pi/issues/10609)) + ## [1.1.0] - 2026-10-07 ### Breaking Changes diff --git a/packages/ai/src/api/mistral-conversations.ts b/packages/ai/src/api/mistral-conversations.ts index de7552ad6..b4a54dd06 100644 --- a/packages/ai/src/api/mistral-conversations.ts +++ b/packages/ai/src/api/mistral-conversations.ts @@ -13,6 +13,7 @@ import type { ToolCall, TranscriptContext, } from "../types.ts"; +import { combineAbortSignals } from "../utils/abort-signals.ts"; import { AssistantMessageEventStream } from "../utils/event-stream.ts"; import { shortHash } from "../utils/hash.ts"; import { headersToRecord } from "../utils/headers.ts"; @@ -304,14 +305,27 @@ async function requestMistralStream( baseUrl.pathname = `${baseUrl.pathname.replace(/\/+$/u, "")}/`; const url = new URL("v1/chat/completions", baseUrl); const headers = buildMistralHeaders(model, apiKey, options); - const timeoutSignal = AbortSignal.timeout(options?.timeoutMs ?? 60_000); - const signal = options?.signal ? AbortSignal.any([options.signal, timeoutSignal]) : timeoutSignal; - const response = await (options?.fetch ?? globalThis.fetch)(url, { - method: "POST", - headers, - body: JSON.stringify(toMistralWirePayload(payload)), - signal, - }); + // The timeout covers only the wait for response headers. Long streams (e.g. extended thinking) + // must not be cut off by a fixed deadline; body stalls are left to the HTTP client idle timeout. + const timeoutMs = options?.timeoutMs ?? 60_000; + const headerTimeoutSignal = AbortSignal.timeout(timeoutMs); + const combinedSignal = combineAbortSignals([options?.signal, headerTimeoutSignal]); + let response: Response; + try { + response = await (options?.fetch ?? globalThis.fetch)(url, { + method: "POST", + headers, + body: JSON.stringify(toMistralWirePayload(payload)), + signal: combinedSignal.signal, + }); + } catch (error) { + if (headerTimeoutSignal.aborted && !options?.signal?.aborted) { + throw new Error(`Mistral response headers timed out after ${timeoutMs}ms`); + } + throw error; + } finally { + combinedSignal.cleanup(); + } await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model); @@ -323,7 +337,7 @@ async function requestMistralStream( throw new Error("Mistral response has no body"); } - return readMistralEvents(response.body, signal); + return readMistralEvents(response.body, options?.signal); } class MistralHttpError extends Error { @@ -444,7 +458,7 @@ const MISTRAL_STREAM_DONE = Symbol("mistral-stream-done"); async function* readMistralEvents( body: ReadableStream, - signal: AbortSignal, + signal: AbortSignal | undefined, ): AsyncGenerator { const reader = body.getReader(); const decoder = new TextDecoder(); @@ -452,13 +466,13 @@ async function* readMistralEvents( const onAbort = () => { void reader.cancel().catch(() => {}); }; - signal.addEventListener("abort", onAbort, { once: true }); + signal?.addEventListener("abort", onAbort, { once: true }); try { while (true) { - if (signal.aborted) throw signal.reason; + if (signal?.aborted) throw signal.reason; const { done, value } = await reader.read(); - if (signal.aborted) throw signal.reason; + if (signal?.aborted) throw signal.reason; buffer += done ? decoder.decode() : decoder.decode(value, { stream: true }); let boundary = findMistralEventBoundary(buffer); @@ -478,7 +492,7 @@ async function* readMistralEvents( if (event !== MISTRAL_STREAM_DONE && event) yield event; } } finally { - signal.removeEventListener("abort", onAbort); + signal?.removeEventListener("abort", onAbort); try { await reader.cancel(); } catch {} diff --git a/packages/ai/test/mistral-http-transport.test.ts b/packages/ai/test/mistral-http-transport.test.ts index 9356e1dd9..6afe745f5 100644 --- a/packages/ai/test/mistral-http-transport.test.ts +++ b/packages/ai/test/mistral-http-transport.test.ts @@ -469,18 +469,15 @@ describe("Mistral HTTP transport", () => { expect(message.stopReason).toBe("aborted"); }); - it("applies the request timeout while waiting for an SSE chunk", async () => { + it("applies the request timeout while waiting for response headers", async () => { const model = getModel("mistral", "mistral-large-latest"); const context = normalizeContext({ messages: [{ role: "user", content: "hello", timestamp: 1 }], }); - const fetch: FetchFunction = async () => - new Response( - new ReadableStream({ - start() {}, - }), - { headers: { "content-type": "text/event-stream" } }, - ); + const fetch: FetchFunction = (_url, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => reject(init.signal?.reason), { once: true }); + }); const message = await streamMistral(model, context, { apiKey: "test", @@ -489,7 +486,44 @@ describe("Mistral HTTP transport", () => { }).result(); expect(message.stopReason).toBe("error"); - expect(message.errorMessage).toMatch(/timeout/i); + expect(message.errorMessage).toBe("Mistral response headers timed out after 5ms"); + }); + + // Regression test for #10609: an active stream must not be cut off after timeoutMs. + it("does not abort an active stream that lasts longer than the request timeout", async () => { + const model = getModel("mistral", "mistral-large-latest"); + const context = normalizeContext({ + messages: [{ role: "user", content: "hello", timestamp: 1 }], + }); + const encoder = new TextEncoder(); + const thinkingEvent = { + choices: [{ index: 0, delta: { content: [{ type: "thinking", thinking: [{ type: "text", text: "x" }] }] } }], + }; + const fetch: FetchFunction = async () => + new Response( + new ReadableStream({ + async start(controller) { + for (let i = 0; i < 5; i++) { + controller.enqueue(encoder.encode(`data: ${JSON.stringify(thinkingEvent)}\n\n`)); + await new Promise((resolve) => setTimeout(resolve, 10)); + } + controller.enqueue( + encoder.encode(`data: ${JSON.stringify(createTerminalEvent())}\n\ndata: [DONE]\n\n`), + ); + controller.close(); + }, + }), + { headers: { "content-type": "text/event-stream" } }, + ); + + const message = await streamMistral(model, context, { + apiKey: "test", + fetch, + timeoutMs: 20, + }).result(); + + expect(message.stopReason).toBe("stop"); + expect(message.content).toEqual([{ type: "thinking", thinking: "xxxxx" }]); }); it("preserves HTTP status and response bodies in errors", async () => { diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index b753720ed..6bc19cb08 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -10,6 +10,10 @@ - Changed the `showHardwareCursor` setting to use only the terminal cursor instead of also drawing Pi's reverse-video cursor +### Fixed + +- Fixed Mistral streams being aborted mid-generation (e.g. during long thinking) after the request timeout; the timeout now applies only to waiting for response headers ([#10609](https://github.com/earendil-works/pi/issues/10609)) + ## [1.1.0] - 2026-10-07 ### New Features