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

@@ -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

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<Uint8Array>,
-	signal: AbortSignal,
+	signal: AbortSignal | undefined,
 ): AsyncGenerator<MistralCompletionEvent> {
 	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 {}

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<Response>((_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 () => {

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