From 6b07b4e57269ebc4b3fe2b7cd3caba2d95e26cc0 Mon Sep 17 00:00:00 2001 From: David Brailovsky Date: Thu, 8 Oct 2026 10:20:44 +0200 Subject: [PATCH] fix(ai): apply Mistral request timeout only to response headers Long Mistral streams (e.g. extended thinking) were aborted once timeoutMs elapsed, even while data was still arriving. The timeout now covers only waiting for response headers; body stalls are left to the HTTP client idle timeout. fixes #10609 --- packages/ai/CHANGELOG.md | 4 ++ packages/ai/src/api/mistral-conversations.ts | 42 ++++++++++----- .../ai/test/mistral-http-transport.test.ts | 52 +++++++++++++++---- packages/coding-agent/CHANGELOG.md | 4 ++ 4 files changed, 79 insertions(+), 23 deletions(-) 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