feat(ai): added prompt cache key and per-provider stream watchdog with idle timeout filtering

- Restored per-provider stream watchdog with idle and first-event timeout support, progress-event filtering, and per-provider timeout overrides via `getStreamIdleTimeoutMs()` and `getStreamFirstEventTimeoutMs()`.
- Fixed silent multi-hour hangs on Codex WebSocket and z.ai/GLM-via-OpenRouter subagent runs by filtering keepalive frames from idle watchdog resets.
- Added `isOpenAICompletionsProgressChunk` export and progress-event filtering to multiple providers (Anthropic, OpenAI Responses/Completions, Azure OpenAI, Codex) to prevent keepalive frames from resetting idle timeouts.
- Un-deprecated `StreamOptions.streamIdleTimeoutMs` and wired it into all built-in providers and lazy stream forwarder with environment variable fallback support.

packages/ai/CHANGELOG.md

@@ -1,10 +1,37 @@
 # Changelog
 
 ## [Unreleased]
+
+### Added
+
+- Added `isOpenAICompletionsProgressChunk` export to identify real progress chunks vs. keepalives in OpenAI completions streams
+- Added per-provider stream watchdog overrides via `getStreamIdleTimeoutMs(fallbackMs)` and `getStreamFirstEventTimeoutMs(idleTimeoutMs, fallbackMs)` to allow providers like Google Gemini CLI to extend first-event timeouts without affecting global defaults
+- Added `promptCacheKey` to `StreamOptions` and passed it through stream option mapping so callers can specify an explicit prompt-cache key separate from `sessionId`
+- Added `promptCacheKey` support to the native server option whitelist so `promptCacheKey` is accepted by `pi-native-server` streams
+- Restored the per-provider stream watchdog (`iterateWithIdleTimeout`) on top of the abortable iterator. The lazy stream forwarder in `register-builtins` now wraps every provider's event stream with the first-event + steady-state idle watchdog (`PI_STREAM_FIRST_EVENT_TIMEOUT_MS`, `PI_STREAM_IDLE_TIMEOUT_MS`; aliases honored), and Anthropic / OpenAI Completions / OpenAI Responses / Azure OpenAI Responses / Codex SSE re-emit their per-provider progress predicates so empty keepalive frames cannot keep a stalled stream alive. Reverts the partial regression from #1392 that left Codex WebSocket subagent runs hanging silently for hours when the broker dropped frames between deltas. The Codex WebSocket transport additionally now resets `lastProgressAt` only on progress events (not keepalives), giving the 300s WS-internal idle ceiling the same liveness semantics as the SSE path.
+
+### Changed
+
+- Changed stream idle watchdog implementation from `iterateUntilAbort` to `iterateWithIdleTimeout`, which now enforces maximum idle gaps between streamed events and distinguishes between first-event and steady-state timeouts
+- Changed Anthropic, OpenAI Responses, OpenAI Completions, Azure OpenAI Responses, and OpenAI Codex Responses providers to use the new idle-timeout iterator with per-provider progress predicates so empty keepalive frames cannot keep a stalled stream alive
+- Changed Codex WebSocket transport to reset `lastProgressAt` only on progress events (not keepalives), giving the 300s WS-internal idle ceiling the same liveness semantics as the SSE path
+- Changed Google Gemini CLI stream forwarding defaults to use a 5-minute first-event floor via per-provider lazy-stream limits to avoid premature first-event timeouts on slow startup
+- Changed OpenAI Responses and OpenAI Codex request handling to keep `sessionId` for provider routing and conversation headers while `promptCacheKey` controls the `prompt_cache_key` payload independently
+- Changed `StreamOptions.streamIdleTimeoutMs` documentation to clarify it is now wired into every built-in provider and the lazy stream forwarder, and that `streamFirstEventTimeoutMs` is honored at both the SDK-request layer and the iterator-watchdog layer
+- Changed OpenAI Responses and OpenAI Codex request handling so `sessionId` continues to drive provider routing and state while `promptCacheKey` controls the `prompt_cache_key` payload
+- Changed Google Gemini CLI stream forwarding defaults to use a 5-minute first-event floor to avoid premature first-event timeouts on slow startup
+- Changed auth-gateway request mapping to preserve incoming `prompt_cache_key` as both `promptCacheKey` and `sessionId` when routing OpenAI-compatible sessions
+- Un-deprecated `StreamOptions.streamIdleTimeoutMs`; the option is wired into every built-in provider and the lazy stream forwarder again. `streamFirstEventTimeoutMs` is now honored at both the SDK-request layer (via `createSdkStreamRequestOptions`) and the iterator-watchdog layer, in cooperation.
+
 ### Removed
 
 - Removed `installH2Fetch` and the `fetch` patch that forced HTTP/2 on HTTPS requests; callers now use the default Bun `fetch` transport
 
+### Fixed
+
+- Fixed silent multi-hour hangs on Codex WebSocket subagent runs when the broker dropped frames between deltas by restoring per-provider stream watchdogs with progress-event filtering
+- Fixed z.ai/GLM-via-OpenRouter subagent stalls where no-op keepalive chunks reset the idle watchdog indefinitely by filtering non-progress items before resetting the deadline
+
 ## [15.4.0] - 2026-05-26
 ### Breaking Changes
 

packages/ai/src/auth-gateway/server.ts

@@ -145,7 +145,9 @@ function buildStreamOptions(parsed: ParsedFormatRequest, api: Api, signal: Abort
 	// Client-supplied `prompt_cache_key` wins; otherwise derive a stable
 	// key from the model + system + tools so prefix caching engages on
 	// Codex-class backends across turns of the same logical conversation.
-	opts.sessionId = options.promptCacheKey ?? deriveSessionId(parsed.modelId, parsed.context);
+	const promptCacheKey = options.promptCacheKey ?? deriveSessionId(parsed.modelId, parsed.context);
+	opts.promptCacheKey = promptCacheKey;
+	opts.sessionId = promptCacheKey;
 	if (options.thinkingBudgets) {
 		opts.thinkingBudgets = { ...(opts.thinkingBudgets ?? {}), ...options.thinkingBudgets };
 	}

packages/ai/src/auth-gateway/types.ts

@@ -67,7 +67,7 @@ export interface AuthGatewayParsedRequestOptions {
 	serviceTier?: ServiceTier;
 	/** Cache retention hint derived from inbound `cache_control` markers. */
 	cacheRetention?: CacheRetention;
-	/** OpenAI Responses `prompt_cache_key`; bridges to pi-ai `sessionId`. */
+	/** OpenAI Responses `prompt_cache_key`; also seeds provider routing when no separate session id exists. */
 	promptCacheKey?: string;
 	/** OpenAI Responses `previous_response_id` for response chaining. */
 	previousResponseId?: string;

packages/ai/src/providers/anthropic.ts

@@ -53,17 +53,16 @@ import {
 	resolveCacheRetention,
 } from "../utils";
 import { createAbortSourceTracker } from "../utils/abort";
-import { iterateUntilAbort } from "../utils/abortable-iterator";
 import { AssistantMessageEventStream } from "../utils/event-stream";
 import { isFoundryEnabled } from "../utils/foundry";
 import { finalizeErrorMessage, type RawHttpRequestDump, rewriteCopilotError } from "../utils/http-inspector";
+import { getStreamFirstEventTimeoutMs, getStreamIdleTimeoutMs, iterateWithIdleTimeout } from "../utils/idle-iterator";
 import { parseJsonWithRepair, parseStreamingJson } from "../utils/json-parse";
 import { parseGitHubCopilotApiKey } from "../utils/oauth/github-copilot";
 import { notifyProviderResponse } from "../utils/provider-response";
 import { isCopilotTransientModelError } from "../utils/retry";
 import { COMBINATOR_KEYS, NO_STRICT, toolWireSchema } from "../utils/schema";
 import { spillToDescription } from "../utils/schema/spill";
-import { createSdkStreamRequestOptions } from "../utils/sdk-stream-timeout";
 import { notifyRawSseEvent, wrapFetchForSseDebug } from "../utils/sse-debug";
 import {
 	buildCopilotDynamicHeaders,
@@ -1089,17 +1088,20 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = (
 				| TextContent
 				| (ToolCall & { partialJson: string })
 			) & { index: number };
+			const idleTimeoutMs = options?.streamIdleTimeoutMs ?? getStreamIdleTimeoutMs();
+			const firstEventTimeoutMs = options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs);
 			const blocks = output.content as Block[];
 			stream.push({ type: "start", partial: output });
 			// Retry loop for transient errors from the stream.
 			// Provider-level transport/rate-limit failures: only before any streamed content starts.
 			// Malformed envelopes/JSON: only before replay-unsafe text/tool events are visible on this stream.
 			let providerRetryAttempt = 0;
+			const firstEventTimeoutAbortError = new Error("Anthropic stream timed out while waiting for the first event");
+			const idleTimeoutAbortError = new Error("Anthropic stream stalled while waiting for the next event");
 			while (true) {
 				activeAbortTracker = createAbortSourceTracker(options?.signal);
 				const { requestSignal } = activeAbortTracker;
-				const requestOptions = createSdkStreamRequestOptions(requestSignal, options?.streamFirstEventTimeoutMs);
-				const anthropicRequest = client.messages.create({ ...params, stream: true }, requestOptions);
+				const anthropicRequest = client.messages.create({ ...params, stream: true }, { signal: requestSignal });
 				let streamedReplayUnsafeContent = false;
 
 				try {
@@ -1117,7 +1119,15 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = (
 					let sawMessageStart = false;
 					let sawTerminalEnvelope = false;
 
-					for await (const event of iterateUntilAbort(anthropicStream, options?.signal)) {
+					for await (const event of iterateWithIdleTimeout(anthropicStream, {
+						idleTimeoutMs,
+						firstItemTimeoutMs: firstEventTimeoutMs,
+						errorMessage: idleTimeoutAbortError.message,
+						firstItemErrorMessage: firstEventTimeoutAbortError.message,
+						onIdle: () => activeAbortTracker.abortLocally(idleTimeoutAbortError),
+						onFirstItemTimeout: () => activeAbortTracker.abortLocally(firstEventTimeoutAbortError),
+						abortSignal: options?.signal,
+					})) {
 						sawEvent = true;
 
 						if (event.type === "message_start") {
@@ -1425,9 +1435,10 @@ export const streamAnthropic: StreamFunction<"anthropic-messages"> = (
 				delete (block as { index?: number }).index;
 				delete (block as { partialJson?: string }).partialJson;
 			}
+			const firstEventTimeoutError = activeAbortTracker.getLocalAbortReason();
 			output.stopReason = activeAbortTracker.wasCallerAbort() ? "aborted" : "error";
 			output.errorStatus = extractHttpStatusFromError(error);
-			output.errorMessage = await finalizeErrorMessage(error, rawRequestDump);
+			output.errorMessage = firstEventTimeoutError?.message ?? (await finalizeErrorMessage(error, rawRequestDump));
 			output.errorMessage = rewriteCopilotError(output.errorMessage, error, model.provider);
 			output.duration = Date.now() - startTime;
 			if (firstTokenTime) output.ttft = firstTokenTime - startTime;

packages/ai/src/providers/azure-openai-responses.ts

@@ -18,11 +18,15 @@ import type {
 } from "../types";
 import { normalizeSystemPrompts } from "../utils";
 import { createAbortSourceTracker } from "../utils/abort";
-import { iterateUntilAbort } from "../utils/abortable-iterator";
 import { AssistantMessageEventStream } from "../utils/event-stream";
 import { finalizeErrorMessage, type RawHttpRequestDump } from "../utils/http-inspector";
+import {
+	createWatchdog,
+	getOpenAIStreamIdleTimeoutMs,
+	getStreamFirstEventTimeoutMs,
+	iterateWithIdleTimeout,
+} from "../utils/idle-iterator";
 import { sanitizeSchemaForOpenAIResponses, toolWireSchema } from "../utils/schema";
-import { createSdkStreamRequestOptions, resolveSdkTimeoutMs } from "../utils/sdk-stream-timeout";
 import { wrapFetchForSseDebug } from "../utils/sse-debug";
 import { mapToOpenAIResponsesToolChoice } from "../utils/tool-choice";
 import { normalizeOpenAIResponsesPromptCacheKey, supportsDeveloperRole } from "./openai-responses";
@@ -39,6 +43,8 @@ import {
 import { transformMessages } from "./transform-messages";
 
 const DEFAULT_AZURE_API_VERSION = "v1";
+const AZURE_OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE =
+	"Azure OpenAI responses stream timed out while waiting for the first event";
 
 function parseDeploymentNameMap(value: string | undefined): Map<string, string> {
 	const map = new Map<string, string>();
@@ -104,7 +110,8 @@ export const streamAzureOpenAIResponses: StreamFunction<"azure-openai-responses"
 		);
 		let rawRequestDump: RawHttpRequestDump | undefined;
 		const abortTracker = createAbortSourceTracker(options?.signal);
-		const { requestSignal } = abortTracker;
+		const firstEventTimeoutAbortError = new Error(AZURE_OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE);
+		const { requestAbortController, requestSignal } = abortTracker;
 
 		try {
 			// Create Azure OpenAI client
@@ -113,6 +120,7 @@ export const streamAzureOpenAIResponses: StreamFunction<"azure-openai-responses"
 			const { baseUrl } = resolveAzureConfig(model, options);
 			const params = buildParams(model, context, options, deploymentName, baseUrl);
 			options?.onPayload?.(params);
+			const idleTimeoutMs = getOpenAIStreamIdleTimeoutMs();
 			rawRequestDump = {
 				provider: model.provider,
 				api: output.api,
@@ -121,15 +129,34 @@ export const streamAzureOpenAIResponses: StreamFunction<"azure-openai-responses"
 				url: `${baseUrl}/responses`,
 				body: params,
 			};
-			const requestOptions = createSdkStreamRequestOptions(requestSignal, options?.streamFirstEventTimeoutMs);
-			const openaiStream = await client.responses.create(params, requestOptions);
+			const openaiStream = await client.responses.create(params, { signal: requestSignal });
+			const firstEventWatchdog = createWatchdog(
+				options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs),
+				() => abortTracker.abortLocally(firstEventTimeoutAbortError),
+			);
 			stream.push({ type: "start", partial: output });
 
-			await processResponsesStream(iterateUntilAbort(openaiStream, options?.signal), output, stream, model, {
-				onFirstToken: () => {
-					if (!firstTokenTime) firstTokenTime = Date.now();
+			await processResponsesStream(
+				iterateWithIdleTimeout(openaiStream, {
+					watchdog: firstEventWatchdog,
+					idleTimeoutMs,
+					errorMessage: "Azure OpenAI responses stream stalled while waiting for the next event",
+					onIdle: () => requestAbortController.abort(),
+				}),
+				output,
+				stream,
+				model,
+				{
+					onFirstToken: () => {
+						if (!firstTokenTime) firstTokenTime = Date.now();
+					},
 				},
-			});
+			);
+
+			const firstEventTimeoutError = abortTracker.getLocalAbortReason();
+			if (firstEventTimeoutError) {
+				throw firstEventTimeoutError;
+			}
 
 			if (abortTracker.wasCallerAbort()) {
 				throw new Error("Request was aborted");
@@ -145,9 +172,10 @@ export const streamAzureOpenAIResponses: StreamFunction<"azure-openai-responses"
 			stream.end();
 		} catch (error) {
 			for (const block of output.content) delete (block as { index?: number }).index;
+			const firstEventTimeoutError = abortTracker.getLocalAbortReason();
 			output.stopReason = abortTracker.wasCallerAbort() ? "aborted" : "error";
 			output.errorStatus = extractHttpStatusFromError(error);
-			output.errorMessage = await finalizeErrorMessage(error, rawRequestDump);
+			output.errorMessage = firstEventTimeoutError?.message ?? (await finalizeErrorMessage(error, rawRequestDump));
 			output.duration = Date.now() - startTime;
 			if (firstTokenTime) output.ttft = firstTokenTime - startTime;
 			stream.push({ type: "error", reason: output.stopReason, error: output });
@@ -218,7 +246,6 @@ function createClient(model: Model<"azure-openai-responses">, apiKey: string, op
 
 	const baseFetch = options?.fetch ?? fetch;
 	const onSseEvent = options?.onSseEvent;
-	const sdkTimeoutMs = resolveSdkTimeoutMs(options?.streamFirstEventTimeoutMs);
 	return new AzureOpenAI({
 		apiKey,
 		apiVersion,
@@ -227,7 +254,6 @@ function createClient(model: Model<"azure-openai-responses">, apiKey: string, op
 		defaultHeaders: headers,
 		baseURL: baseUrl,
 		fetch: onSseEvent ? wrapFetchForSseDebug(baseFetch, event => onSseEvent(event, model)) : baseFetch,
-		...(sdkTimeoutMs !== undefined ? { timeout: sdkTimeoutMs } : {}),
 	});
 }
 
@@ -244,7 +270,7 @@ function buildParams(
 		model: deploymentName,
 		input: messages,
 		stream: true,
-		prompt_cache_key: normalizeOpenAIResponsesPromptCacheKey(options?.sessionId),
+		prompt_cache_key: normalizeOpenAIResponsesPromptCacheKey(options?.promptCacheKey ?? options?.sessionId),
 	};
 
 	applyCommonResponsesSamplingParams(params, options, model.provider);

packages/ai/src/providers/openai-codex-responses.ts

@@ -46,9 +46,9 @@ import {
 	getOpenAIResponsesHistoryPayload,
 	normalizeSystemPrompts,
 } from "../utils";
-import { iterateUntilAbort } from "../utils/abortable-iterator";
 import { AssistantMessageEventStream } from "../utils/event-stream";
 import { finalizeErrorMessage, type RawHttpRequestDump } from "../utils/http-inspector";
+import { getOpenAIStreamIdleTimeoutMs, iterateWithIdleTimeout } from "../utils/idle-iterator";
 import { parseStreamingJson } from "../utils/json-parse";
 import { adaptSchemaForStrict, NO_STRICT, sanitizeSchemaForOpenAIResponses, toolWireSchema } from "../utils/schema";
 import { notifyRawSseEvent } from "../utils/sse-debug";
@@ -118,6 +118,32 @@ const X_REASONING_INCLUDED_HEADER = "x-reasoning-included";
 const CODEX_WEBSOCKET_FATAL_PATTERNS = ["websocket error:", "websocket closed before open", "connection timeout"];
 /** Max total time to spend retrying 429s with server-provided delays (5 minutes). */
 const CODEX_RATE_LIMIT_BUDGET_MS = 5 * 60 * 1000;
+const CODEX_PROGRESS_EVENT_TYPES = new Set([
+	"response.created",
+	"response.output_item.added",
+	"response.reasoning_summary_part.added",
+	"response.reasoning_summary_text.delta",
+	"response.reasoning_summary_part.done",
+	"response.content_part.added",
+	"response.output_text.delta",
+	"response.refusal.delta",
+	"response.function_call_arguments.delta",
+	"response.function_call_arguments.done",
+	"response.custom_tool_call_input.delta",
+	"response.custom_tool_call_input.done",
+	"response.output_item.done",
+	"response.completed",
+	"response.done",
+	"response.incomplete",
+	"response.failed",
+	"error",
+]);
+
+function isCodexStreamProgressEvent(event: unknown): boolean {
+	if (!event || typeof event !== "object") return false;
+	const type = (event as { type?: unknown }).type;
+	return typeof type === "string" && CODEX_PROGRESS_EVENT_TYPES.has(type);
+}
 
 type CodexTransport = "sse" | "websocket";
 type CodexEventItem = ResponseReasoningItem | ResponseOutputMessage | ResponseFunctionToolCall | ResponseCustomToolCall;
@@ -159,6 +185,7 @@ interface CodexRequestContext {
 	baseUrl: string;
 	url: string;
 	requestHeaders: Record<string, string>;
+	transportSessionId?: string;
 	providerSessionState?: CodexProviderSessionState;
 	websocketState?: CodexWebSocketSessionState;
 	transformedBody: RequestBody;
@@ -167,6 +194,8 @@ interface CodexRequestContext {
 
 interface CodexRequestSetup {
 	requestSignal: AbortSignal;
+	wrapCodexSseStream: (source: AsyncGenerator<Record<string, unknown>>) => AsyncGenerator<Record<string, unknown>>;
+	requestAbortController: AbortController;
 }
 
 interface CodexStreamRuntime {
@@ -520,7 +549,21 @@ function removeTransientBlockIndices(output: AssistantMessage): void {
 }
 
 function createRequestSetup(options: OpenAICodexResponsesOptions | undefined): CodexRequestSetup {
-	return { requestSignal: options?.signal ?? new AbortController().signal };
+	const requestAbortController = new AbortController();
+	const requestSignal = options?.signal
+		? AbortSignal.any([options.signal, requestAbortController.signal])
+		: requestAbortController.signal;
+	const wrapCodexSseStream = (
+		source: AsyncGenerator<Record<string, unknown>>,
+	): AsyncGenerator<Record<string, unknown>> =>
+		iterateWithIdleTimeout(source, {
+			idleTimeoutMs: options?.streamIdleTimeoutMs ?? getOpenAIStreamIdleTimeoutMs(),
+			errorMessage: "OpenAI Codex SSE stream stalled while waiting for the next event",
+			onIdle: () => requestAbortController.abort(),
+			abortSignal: options?.signal,
+			isProgressItem: isCodexStreamProgressEvent,
+		});
+	return { requestAbortController, requestSignal, wrapCodexSseStream };
 }
 
 async function buildCodexRequestContext(
@@ -537,8 +580,9 @@ async function buildCodexRequestContext(
 	const accountId = getAccountId(apiKey);
 	const baseUrl = model.baseUrl || CODEX_BASE_URL;
 	const url = resolveCodexResponsesUrl(baseUrl);
-	const promptCacheKey = normalizeOpenAIResponsesPromptCacheKey(options?.sessionId);
-	const transformedBody = await buildTransformedCodexRequestBody(model, context, options);
+	const promptCacheKey = resolveCodexPromptCacheKey(options);
+	const transportSessionId = resolveCodexTransportSessionId(options);
+	const transformedBody = await buildTransformedCodexRequestBody(model, context, options, promptCacheKey);
 	options?.onPayload?.(transformedBody);
 
 	const requestHeaders = { ...(model.headers ?? {}), ...(options?.headers ?? {}) };
@@ -552,20 +596,20 @@ async function buildCodexRequestContext(
 	};
 
 	const providerSessionState = getCodexProviderSessionState(options?.providerSessionState);
-	const sessionKey = getCodexWebSocketSessionKey(promptCacheKey, model, accountId, baseUrl);
-	const publicSessionKey = getCodexPublicSessionKey(promptCacheKey, model, baseUrl);
+	const sessionKey = getCodexWebSocketSessionKey(transportSessionId, model, accountId, baseUrl);
+	const publicSessionKey = getCodexPublicSessionKey(transportSessionId, model, baseUrl);
 	if (sessionKey && publicSessionKey) {
 		providerSessionState?.webSocketPublicToPrivate.set(publicSessionKey, sessionKey);
 	}
 	const websocketState =
 		sessionKey && providerSessionState ? getCodexWebSocketSessionState(sessionKey, providerSessionState) : undefined;
-
 	return {
 		apiKey,
 		accountId,
 		baseUrl,
 		url,
 		requestHeaders,
+		transportSessionId,
 		providerSessionState,
 		websocketState,
 		transformedBody,
@@ -577,12 +621,13 @@ async function buildTransformedCodexRequestBody(
 	model: Model<"openai-codex-responses">,
 	context: Context,
 	options: OpenAICodexResponsesOptions | undefined,
+	promptCacheKey = resolveCodexPromptCacheKey(options),
 ): Promise<RequestBody> {
 	const params: RequestBody = {
 		model: model.id,
 		input: [...convertMessages(model, context)],
 		stream: true,
-		prompt_cache_key: normalizeOpenAIResponsesPromptCacheKey(options?.sessionId),
+		prompt_cache_key: promptCacheKey,
 	};
 
 	if (options?.maxTokens) {
@@ -709,7 +754,7 @@ async function openCodexWebSocketTransport(
 		requestContext.requestHeaders,
 		requestContext.accountId,
 		requestContext.apiKey,
-		requestContext.transformedBody.prompt_cache_key,
+		requestContext.transportSessionId,
 		"websocket",
 		websocketState,
 	);
@@ -748,20 +793,19 @@ async function openCodexSseTransport(
 	requestBodyForState: RequestBody;
 	transport: CodexTransport;
 }> {
-	const eventStream = iterateUntilAbort(
+	const eventStream = requestSetup.wrapCodexSseStream(
 		await openCodexSseEventStream(
 			requestContext.url,
 			requestContext.requestHeaders,
 			requestContext.accountId,
 			requestContext.apiKey,
-			body.prompt_cache_key,
+			requestContext.transportSessionId,
 			body,
 			state,
 			requestSetup.requestSignal,
 			event => options?.onSseEvent?.(event, model),
 			options?.fetch,
 		),
-		requestSetup.requestSignal,
 	);
 	return { eventStream, requestBodyForState: structuredCloneJSON(body), transport: "sse" };
 }
@@ -1659,6 +1703,18 @@ export async function prewarmOpenAICodexResponses(
 	state.prewarmed = true;
 }
 
+function resolveCodexPromptCacheKey(
+	options: Pick<OpenAICodexResponsesOptions, "promptCacheKey" | "sessionId"> | undefined,
+): string | undefined {
+	return normalizeOpenAIResponsesPromptCacheKey(options?.promptCacheKey ?? options?.sessionId);
+}
+
+function resolveCodexTransportSessionId(
+	options: Pick<OpenAICodexResponsesOptions, "sessionId"> | undefined,
+): string | undefined {
+	return normalizeOpenAIResponsesPromptCacheKey(options?.sessionId);
+}
+
 function getCodexWebSocketSessionKey(
 	sessionId: string | undefined,
 	model: Model<"openai-codex-responses">,
@@ -2109,7 +2165,9 @@ class CodexWebSocketConnection {
 					throw createCodexWebSocketTransportError("websocket closed before response completion");
 				}
 				sawFirstEvent = true;
-				lastProgressAt = Date.now();
+				if (isCodexStreamProgressEvent(next)) {
+					lastProgressAt = Date.now();
+				}
 				yield next;
 				const eventType = typeof next.type === "string" ? next.type : "";
 				if (
@@ -2268,7 +2326,7 @@ function createCodexHeaders(
 	initHeaders: Record<string, string> | undefined,
 	accountId: string,
 	accessToken: string,
-	promptCacheKey?: string,
+	sessionId?: string,
 	transport: CodexTransport = "sse",
 	state?: CodexWebSocketSessionState,
 ): Headers {
@@ -2285,10 +2343,10 @@ function createCodexHeaders(
 	headers.set(OPENAI_HEADERS.BETA, betaHeader);
 	headers.set(OPENAI_HEADERS.ORIGINATOR, OPENAI_HEADER_VALUES.ORIGINATOR_CODEX);
 	headers.set("User-Agent", getCodexUserAgent());
-	if (promptCacheKey) {
-		headers.set(OPENAI_HEADERS.CONVERSATION_ID, promptCacheKey);
-		headers.set(OPENAI_HEADERS.SESSION_ID, promptCacheKey);
-		headers.set("x-client-request-id", promptCacheKey);
+	if (sessionId) {
+		headers.set(OPENAI_HEADERS.CONVERSATION_ID, sessionId);
+		headers.set(OPENAI_HEADERS.SESSION_ID, sessionId);
+		headers.set("x-client-request-id", sessionId);
 	} else {
 		headers.delete(OPENAI_HEADERS.CONVERSATION_ID);
 		headers.delete(OPENAI_HEADERS.SESSION_ID);

packages/ai/src/providers/openai-completions.ts

@@ -37,7 +37,6 @@ import {
 } from "../types";
 import { normalizeSystemPrompts } from "../utils";
 import { createAbortSourceTracker } from "../utils/abort";
-import { iterateUntilAbort } from "../utils/abortable-iterator";
 import { AssistantMessageEventStream } from "../utils/event-stream";
 import { toFirepassWireModelId, toFireworksWireModelId } from "../utils/fireworks-model-id";
 import {
@@ -46,13 +45,18 @@ import {
 	type RawHttpRequestDump,
 	rewriteCopilotError,
 } from "../utils/http-inspector";
+import {
+	createWatchdog,
+	getOpenAIStreamIdleTimeoutMs,
+	getStreamFirstEventTimeoutMs,
+	iterateWithIdleTimeout,
+} from "../utils/idle-iterator";
 import { parseStreamingJson } from "../utils/json-parse";
 import { parseGitHubCopilotApiKey } from "../utils/oauth/github-copilot";
 import { getKimiCommonHeaders } from "../utils/oauth/kimi";
 import { notifyProviderResponse } from "../utils/provider-response";
 import { callWithCopilotModelRetry } from "../utils/retry";
 import { adaptSchemaForStrict, NO_STRICT, toolWireSchema } from "../utils/schema";
-import { resolveSdkTimeoutMs } from "../utils/sdk-stream-timeout";
 import { wrapFetchForSseDebug } from "../utils/sse-debug";
 import { type HealedToolCall, modelMayLeakKimiToolCalls, ToolCallHealer } from "../utils/tool-call-healing";
 import { isForcedToolChoice, mapToOpenAICompletionsToolChoice } from "../utils/tool-choice";
@@ -164,6 +168,52 @@ function hasToolHistory(messages: Message[]): boolean {
 	}
 	return false;
 }
+/**
+ * Identify "real progress" stream chunks vs. keepalives, role-only preambles,
+ * and empty `{choices:[]}` no-ops emitted by some OpenAI-compatible endpoints.
+ * Without this filter, every keepalive resets `iterateWithIdleTimeout`'s
+ * deadline, so a provider that streams nothing but pings keeps the watchdog
+ * asleep indefinitely — observed against z.ai/GLM via OpenRouter where a
+ * subagent stalled for hours with no error surfaced.
+ *
+ * A chunk counts as progress when it carries terminal usage, a finish reason,
+ * or any model-produced delta (content / tool calls / reasoning / refusal).
+ * Role-only `delta: { role: "assistant" }` preambles do NOT count; we want the
+ * (longer) first-event timeout to keep governing until real output appears.
+ */
+export function isOpenAICompletionsProgressChunk(chunk: unknown): boolean {
+	if (!chunk || typeof chunk !== "object") return false;
+	const record = chunk as {
+		usage?: unknown;
+		choices?: ReadonlyArray<{
+			finish_reason?: unknown;
+			usage?: unknown;
+			delta?: {
+				content?: unknown;
+				tool_calls?: unknown;
+				reasoning?: unknown;
+				reasoning_content?: unknown;
+				reasoning_text?: unknown;
+				refusal?: unknown;
+			};
+		}>;
+	};
+	if (record.usage) return true;
+	const choice = Array.isArray(record.choices) ? record.choices[0] : undefined;
+	if (!choice) return false;
+	if (choice.finish_reason) return true;
+	if (choice.usage) return true;
+	const delta = choice.delta;
+	if (!delta) return false;
+	const content = delta.content;
+	if (typeof content === "string" ? content.length > 0 : Array.isArray(content) && content.length > 0) return true;
+	if (Array.isArray(delta.tool_calls) && delta.tool_calls.length > 0) return true;
+	if (typeof delta.reasoning === "string" && delta.reasoning.length > 0) return true;
+	if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) return true;
+	if (typeof delta.reasoning_text === "string" && delta.reasoning_text.length > 0) return true;
+	if (typeof delta.refusal === "string" && delta.refusal.length > 0) return true;
+	return false;
+}
 
 export interface OpenAICompletionsOptions extends StreamOptions {
 	toolChoice?: ToolChoice;
@@ -321,6 +371,8 @@ function getTrailingPartialDeepseekToken(text: string): string {
 	if (tail.length > 256) return "";
 	return tail;
 }
+const OPENAI_COMPLETIONS_FIRST_EVENT_TIMEOUT_MESSAGE =
+	"OpenAI completions stream timed out while waiting for the first event";
 
 export const streamOpenAICompletions: StreamFunction<"openai-completions"> = (
 	model: Model<"openai-completions">,
@@ -337,10 +389,12 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = (
 		const output: AssistantMessage = createInitialResponsesAssistantMessage(model.api, model.provider, model.id);
 		let rawRequestDump: RawHttpRequestDump | undefined;
 		const abortTracker = createAbortSourceTracker(options?.signal);
-		const { requestSignal } = abortTracker;
+		const firstEventTimeoutAbortError = new Error(OPENAI_COMPLETIONS_FIRST_EVENT_TIMEOUT_MESSAGE);
+		const { requestAbortController, requestSignal } = abortTracker;
 
 		try {
 			const apiKey = options?.apiKey || getEnvApiKey(model.provider) || "";
+			const idleTimeoutMs = getOpenAIStreamIdleTimeoutMs();
 			const {
 				client,
 				copilotPremiumRequests,
@@ -422,6 +476,10 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = (
 					openaiStream = await createCompletionsStream("none");
 				}
 			}
+			const firstEventWatchdog = createWatchdog(
+				options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs),
+				() => abortTracker.abortLocally(firstEventTimeoutAbortError),
+			);
 			if (premiumRequestsTotal !== undefined) {
 				output.usage.premiumRequests = premiumRequestsTotal;
 			}
@@ -601,7 +659,14 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = (
 				for (const call of calls) emitHealedToolCall(call);
 			};
 
-			for await (const chunk of iterateUntilAbort(openaiStream, options?.signal)) {
+			for await (const chunk of iterateWithIdleTimeout(openaiStream, {
+				watchdog: firstEventWatchdog,
+				idleTimeoutMs,
+				errorMessage: "OpenAI completions stream stalled while waiting for the next event",
+				onIdle: () => requestAbortController.abort(),
+				abortSignal: options?.signal,
+				isProgressItem: isOpenAICompletionsProgressChunk,
+			})) {
 				if (!chunk || typeof chunk !== "object") continue;
 
 				// OpenAI documents ChatCompletionChunk.id as the unique chat completion identifier,
@@ -770,6 +835,10 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = (
 
 			finishCurrentBlock(currentBlock);
 
+			const firstEventTimeoutError = abortTracker.getLocalAbortReason();
+			if (firstEventTimeoutError) {
+				throw firstEventTimeoutError;
+			}
 			if (abortTracker.wasCallerAbort()) {
 				throw new Error("Request was aborted");
 			}
@@ -788,9 +857,12 @@ export const streamOpenAICompletions: StreamFunction<"openai-completions"> = (
 			stream.end();
 		} catch (error) {
 			for (const block of output.content) delete (block as any).index;
+			const firstEventTimeoutError = abortTracker.getLocalAbortReason();
 			output.stopReason = abortTracker.wasCallerAbort() ? "aborted" : "error";
 			output.errorStatus = extractHttpStatusFromError(error) ?? getCapturedErrorResponse?.()?.status;
-			output.errorMessage = await finalizeErrorMessage(error, rawRequestDump, getCapturedErrorResponse?.());
+			output.errorMessage =
+				firstEventTimeoutError?.message ??
+				(await finalizeErrorMessage(error, rawRequestDump, getCapturedErrorResponse?.()));
 			// Some providers via OpenRouter include extra details here.
 			const rawMetadata = (error as { error?: { metadata?: { raw?: string } } })?.error?.metadata?.raw;
 			if (rawMetadata) output.errorMessage += `\n${rawMetadata}`;
@@ -912,7 +984,25 @@ async function createClient(
 		baseFetch.preconnect ? { preconnect: baseFetch.preconnect } : {},
 	);
 	const debugFetch = onSseEvent ? wrapFetchForSseDebug(wrappedFetch, event => onSseEvent(event, model)) : wrappedFetch;
-	const sdkTimeoutMs = resolveSdkTimeoutMs(streamFirstEventTimeoutOverride);
+	// Bound HTTP request timeout to roughly the first-event watchdog window.
+	// The OpenAI SDK's default is 10 minutes per attempt × `maxRetries`, which
+	// turns a stalled-before-headers fetch into a multi-minute hang invisible
+	// to the agent loop (the iterator watchdog only arms AFTER `create()` returns).
+	// Using the first-event timeout keeps both layers aligned: the SDK gives up
+	// before the agent watchdog would have, surfacing a real error to the catch
+	// in the IIFE.
+	// A caller may raise `StreamOptions.streamFirstEventTimeoutMs` for a slow-
+	// before-headers provider; respect it so the SDK doesn't give up before the
+	// wrapping watchdog arms. An explicit `0` disables the first-event watchdog,
+	// and the SDK treats `timeout: 0` as an immediate timeout, so do not pass a
+	// request timeout in that case.
+	const envSdkTimeoutMs = getStreamFirstEventTimeoutMs(getOpenAIStreamIdleTimeoutMs());
+	const sdkTimeoutMs =
+		streamFirstEventTimeoutOverride === 0
+			? undefined
+			: streamFirstEventTimeoutOverride !== undefined
+				? Math.max(envSdkTimeoutMs ?? 0, streamFirstEventTimeoutOverride)
+				: envSdkTimeoutMs;
 	return {
 		client: new OpenAI({
 			apiKey,

packages/ai/src/providers/openai-responses.ts

@@ -30,14 +30,18 @@ import {
 	sanitizeOpenAIResponsesHistoryItemsForReplay,
 } from "../utils";
 import { createAbortSourceTracker } from "../utils/abort";
-import { iterateUntilAbort } from "../utils/abortable-iterator";
 import { AssistantMessageEventStream } from "../utils/event-stream";
 import { finalizeErrorMessage, type RawHttpRequestDump, rewriteCopilotError } from "../utils/http-inspector";
+import {
+	createWatchdog,
+	getOpenAIStreamIdleTimeoutMs,
+	getStreamFirstEventTimeoutMs,
+	iterateWithIdleTimeout,
+} from "../utils/idle-iterator";
 import { parseGitHubCopilotApiKey } from "../utils/oauth/github-copilot";
 import { notifyProviderResponse } from "../utils/provider-response";
 import { callWithCopilotModelRetry } from "../utils/retry";
 import { adaptSchemaForStrict, NO_STRICT, sanitizeSchemaForOpenAIResponses, toolWireSchema } from "../utils/schema";
-import { createSdkStreamRequestOptions, resolveSdkTimeoutMs } from "../utils/sdk-stream-timeout";
 import { wrapFetchForSseDebug } from "../utils/sse-debug";
 import { mapToOpenAIResponsesToolChoice, type OpenAIResponsesToolChoice } from "../utils/tool-choice";
 import {
@@ -96,6 +100,34 @@ export interface OpenAIResponsesOptions extends StreamOptions {
 }
 
 const OPENAI_RESPONSES_PROVIDER_SESSION_STATE_PREFIX = "openai-responses:";
+const OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE =
+	"OpenAI responses stream timed out while waiting for the first event";
+
+const OPENAI_RESPONSES_PROGRESS_EVENT_TYPES = new Set([
+	"response.created",
+	"response.output_item.added",
+	"response.reasoning_summary_part.added",
+	"response.reasoning_summary_text.delta",
+	"response.reasoning_summary_part.done",
+	"response.reasoning_text.delta",
+	"response.content_part.added",
+	"response.output_text.delta",
+	"response.refusal.delta",
+	"response.function_call_arguments.delta",
+	"response.function_call_arguments.done",
+	"response.custom_tool_call_input.delta",
+	"response.custom_tool_call_input.done",
+	"response.output_item.done",
+	"response.completed",
+	"response.failed",
+	"error",
+]);
+
+function isOpenAIResponsesProgressEvent(event: unknown): boolean {
+	if (!event || typeof event !== "object") return false;
+	const type = (event as { type?: unknown }).type;
+	return typeof type === "string" && OPENAI_RESPONSES_PROGRESS_EVENT_TYPES.has(type);
+}
 
 interface OpenAIResponsesProviderSessionState extends ProviderSessionState {
 	nativeHistoryReplayWarmed: boolean;
@@ -165,11 +197,14 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = (
 		);
 		let rawRequestDump: RawHttpRequestDump | undefined;
 		const abortTracker = createAbortSourceTracker(options?.signal);
-		const { requestSignal } = abortTracker;
+		const firstEventTimeoutAbortError = new Error(OPENAI_RESPONSES_FIRST_EVENT_TIMEOUT_MESSAGE);
+		const { requestAbortController, requestSignal } = abortTracker;
 
 		try {
-			// Keep request headers and prompt-cache routing on the same session-derived value.
-			const cacheSessionId = getOpenAIResponsesCacheSessionId(options);
+			// Keep request routing on `sessionId` while allowing callers to pin a
+			// stable prompt-cache key independently. Side-channel calls use this to
+			// avoid perturbing provider conversation state without cold-starting the cache.
+			const routingSessionId = getOpenAIResponsesRoutingSessionId(options);
 			const apiKey = options?.apiKey || getEnvApiKey(model.provider) || "";
 			const { client, copilotPremiumRequests, baseUrl } = createClient(
 				model,
@@ -177,14 +212,14 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = (
 				apiKey,
 				options?.headers,
 				options?.initiatorOverride,
-				cacheSessionId,
+				routingSessionId,
 				options?.onSseEvent,
 				options?.fetch,
-				options?.streamFirstEventTimeoutMs,
 			);
 			const premiumRequestsTotal = copilotPremiumRequests;
 			const providerSessionState = getOpenAIResponsesProviderSessionState(model, options?.providerSessionState);
 			const { params } = buildParams(model, context, options, providerSessionState, baseUrl);
+			const idleTimeoutMs = options?.streamIdleTimeoutMs ?? getOpenAIStreamIdleTimeoutMs();
 			options?.onPayload?.(params);
 			rawRequestDump = {
 				provider: model.provider,
@@ -194,11 +229,10 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = (
 				url: `${baseUrl ?? "https://api.openai.com/v1"}/responses`,
 				body: params,
 			};
-			const requestOptions = createSdkStreamRequestOptions(requestSignal, options?.streamFirstEventTimeoutMs);
 			const openaiStream = await callWithCopilotModelRetry(
 				async () => {
 					const { data, response, request_id } = await client.responses
-						.create(params, requestOptions)
+						.create(params, { signal: requestSignal })
 						.withResponse();
 					await notifyProviderResponse(options, response, model, request_id);
 					return data;
@@ -206,19 +240,40 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = (
 				{ provider: model.provider, signal: requestSignal },
 			);
 			if (premiumRequestsTotal !== undefined) output.usage.premiumRequests = premiumRequestsTotal;
+			const firstEventWatchdog = createWatchdog(
+				options?.streamFirstEventTimeoutMs ?? getStreamFirstEventTimeoutMs(idleTimeoutMs),
+				() => abortTracker.abortLocally(firstEventTimeoutAbortError),
+			);
 			stream.push({ type: "start", partial: output });
 
 			const nativeOutputItems: Array<Record<string, unknown>> = [];
-			await processResponsesStream(iterateUntilAbort(openaiStream, options?.signal), output, stream, model, {
-				onFirstToken: () => {
-					if (!firstTokenTime) firstTokenTime = Date.now();
+			await processResponsesStream(
+				iterateWithIdleTimeout(openaiStream, {
+					idleTimeoutMs,
+					watchdog: firstEventWatchdog,
+					errorMessage: "OpenAI responses stream stalled while waiting for the next event",
+					onIdle: () => requestAbortController.abort(),
+					abortSignal: options?.signal,
+					isProgressItem: isOpenAIResponsesProgressEvent,
+				}),
+				output,
+				stream,
+				model,
+				{
+					onFirstToken: () => {
+						if (!firstTokenTime) firstTokenTime = Date.now();
+					},
+					onOutputItemDone: item => {
+						nativeOutputItems.push(structuredCloneJSON<unknown>(item) as unknown as Record<string, unknown>);
+					},
 				},
-				onOutputItemDone: item => {
-					nativeOutputItems.push(structuredCloneJSON<unknown>(item) as unknown as Record<string, unknown>);
-				},
-			});
+			);
 			if (premiumRequestsTotal !== undefined) output.usage.premiumRequests = premiumRequestsTotal;
 
+			const firstEventTimeoutError = abortTracker.getLocalAbortReason();
+			if (firstEventTimeoutError) {
+				throw firstEventTimeoutError;
+			}
 			if (abortTracker.wasCallerAbort()) {
 				throw new Error("Request was aborted");
 			}
@@ -236,9 +291,10 @@ export const streamOpenAIResponses: StreamFunction<"openai-responses"> = (
 			stream.end();
 		} catch (error) {
 			for (const block of output.content) delete (block as { index?: number }).index;
+			const firstEventTimeoutError = abortTracker.getLocalAbortReason();
 			output.stopReason = abortTracker.wasCallerAbort() ? "aborted" : "error";
 			output.errorStatus = extractHttpStatusFromError(error);
-			output.errorMessage = await finalizeErrorMessage(error, rawRequestDump);
+			output.errorMessage = firstEventTimeoutError?.message ?? (await finalizeErrorMessage(error, rawRequestDump));
 			output.errorMessage = rewriteCopilotError(output.errorMessage, error, model.provider);
 			output.duration = Date.now() - startTime;
 			if (firstTokenTime) output.ttft = firstTokenTime - startTime;
@@ -259,7 +315,6 @@ function createClient(
 	sessionId?: string,
 	onSseEvent?: OpenAIResponsesOptions["onSseEvent"],
 	fetchOverride?: FetchImpl,
-	streamFirstEventTimeoutOverride?: number,
 ): {
 	client: OpenAI;
 	copilotPremiumRequests: number | undefined;
@@ -298,7 +353,6 @@ function createClient(
 		headers["x-client-request-id"] ??= sessionId;
 	}
 	const baseFetch = fetchOverride ?? fetch;
-	const sdkTimeoutMs = resolveSdkTimeoutMs(streamFirstEventTimeoutOverride);
 	return {
 		client: new OpenAI({
 			apiKey,
@@ -307,19 +361,24 @@ function createClient(
 			maxRetries: 5,
 			defaultHeaders: headers,
 			fetch: onSseEvent ? wrapFetchForSseDebug(baseFetch, event => onSseEvent(event, model)) : baseFetch,
-			...(sdkTimeoutMs !== undefined ? { timeout: sdkTimeoutMs } : {}),
 		}),
 		copilotPremiumRequests,
 		baseUrl,
 	};
 }
 
-function getOpenAIResponsesCacheSessionId(
+function getOpenAIResponsesPromptCacheKey(
+	options: Pick<OpenAIResponsesOptions, "cacheRetention" | "promptCacheKey" | "sessionId"> | undefined,
+): string | undefined {
+	if (resolveCacheRetention(options?.cacheRetention) === "none") return undefined;
+	return normalizeOpenAIResponsesPromptCacheKey(options?.promptCacheKey ?? options?.sessionId);
+}
+
+function getOpenAIResponsesRoutingSessionId(
 	options: Pick<OpenAIResponsesOptions, "cacheRetention" | "sessionId"> | undefined,
 ): string | undefined {
-	return resolveCacheRetention(options?.cacheRetention) === "none"
-		? undefined
-		: normalizeOpenAIResponsesPromptCacheKey(options?.sessionId);
+	if (resolveCacheRetention(options?.cacheRetention) === "none") return undefined;
+	return normalizeOpenAIResponsesPromptCacheKey(options?.sessionId);
 }
 
 function buildParams(
@@ -359,7 +418,7 @@ function buildParams(
 	}
 
 	const cacheRetention = resolveCacheRetention(options?.cacheRetention);
-	const promptCacheKey = getOpenAIResponsesCacheSessionId(options);
+	const promptCacheKey = getOpenAIResponsesPromptCacheKey(options);
 	const params: OpenAIResponsesSamplingParams = {
 		model: model.id,
 		input: messages,

packages/ai/src/providers/pi-native-server.ts

@@ -58,6 +58,7 @@ const ALLOWED_OPTION_KEYS: ReadonlySet<keyof SimpleStreamOptions> = new Set([
 	"maxRetryDelayMs",
 	"metadata",
 	"sessionId",
+	"promptCacheKey",
 	"streamFirstEventTimeoutMs",
 	"streamIdleTimeoutMs",
 	"reasoning",

packages/ai/src/providers/register-builtins.ts

@@ -20,8 +20,8 @@ import type {
 	OptionsForApi,
 } from "../types";
 import { type AbortSourceTracker, createAbortSourceTracker } from "../utils/abort";
-import { iterateUntilAbort } from "../utils/abortable-iterator";
 import { AssistantMessageEventStream as EventStreamImpl } from "../utils/event-stream";
+import { getStreamFirstEventTimeoutMs, getStreamIdleTimeoutMs, iterateWithIdleTimeout } from "../utils/idle-iterator";
 import type { BedrockOptions } from "./amazon-bedrock";
 import type { AnthropicOptions } from "./anthropic";
 import type { AzureOpenAIResponsesOptions } from "./azure-openai-responses";
@@ -157,22 +157,70 @@ export function setBedrockProviderModule(module: BedrockProviderModule): void {
 // Stream forwarding / error helpers
 // ---------------------------------------------------------------------------
 
+const LAZY_STREAM_IDLE_TIMEOUT_ERROR = "Provider stream stalled while waiting for the next event";
+const LAZY_STREAM_FIRST_EVENT_TIMEOUT_ERROR = "Provider stream timed out while waiting for the first event";
+
 function hasFinalResult(
 	source: AsyncIterable<AssistantMessageEvent>,
 ): source is AsyncIterable<AssistantMessageEvent> & { result(): Promise<AssistantMessage> } {
 	return typeof (source as { result?: unknown }).result === "function";
 }
 
+/**
+ * Per-provider default overrides for the lazy stream watchdogs. These widen the
+ * floor used when neither caller option nor env var pins a value. The env vars
+ * (`PI_STREAM_FIRST_EVENT_TIMEOUT_MS`, `PI_STREAM_IDLE_TIMEOUT_MS`) still take
+ * precedence; `StreamOptions.streamFirstEventTimeoutMs` / `streamIdleTimeoutMs`
+ * still trump everything.
+ */
+interface LazyStreamLimits {
+	defaultFirstEventTimeoutMs?: number;
+	defaultIdleTimeoutMs?: number;
+}
+
+/**
+ * Cloud Code Assist (google-gemini-cli / google-antigravity) routinely takes
+ * longer than the global 100s default to emit its first SSE event when serving
+ * the heavier Gemini 3.x Pro tiers at high thinking levels. Bump the first-event
+ * floor to five minutes so callers stop seeing spurious "stream timed out while
+ * waiting for the first event" aborts on legitimate cold reasoning starts.
+ * The steady-state idle watchdog stays on the global default since the upstream
+ * emits thinking tokens frequently once it gets going.
+ */
+const GOOGLE_GEMINI_CLI_LAZY_STREAM_LIMITS: LazyStreamLimits = {
+	defaultFirstEventTimeoutMs: 300_000,
+};
+
 function forwardStream<TApi extends Api>(
 	target: EventStreamImpl,
 	source: AsyncIterable<AssistantMessageEvent>,
 	model: Model<TApi>,
 	options: OptionsForApi<TApi>,
 	abortTracker: AbortSourceTracker,
+	limits?: LazyStreamLimits,
 ): void {
 	(async () => {
 		try {
-			for await (const event of iterateUntilAbort(source, options.signal)) {
+			const idleTimeoutMs = options.streamIdleTimeoutMs ?? getStreamIdleTimeoutMs(limits?.defaultIdleTimeoutMs);
+			const watchedSource = iterateWithIdleTimeout(source, {
+				idleTimeoutMs,
+				firstItemTimeoutMs:
+					options.streamFirstEventTimeoutMs ??
+					getStreamFirstEventTimeoutMs(idleTimeoutMs, limits?.defaultFirstEventTimeoutMs),
+				errorMessage: LAZY_STREAM_IDLE_TIMEOUT_ERROR,
+				firstItemErrorMessage: LAZY_STREAM_FIRST_EVENT_TIMEOUT_ERROR,
+				onIdle: () => abortTracker.abortLocally(new Error(LAZY_STREAM_IDLE_TIMEOUT_ERROR)),
+				onFirstItemTimeout: () => abortTracker.abortLocally(new Error(LAZY_STREAM_FIRST_EVENT_TIMEOUT_ERROR)),
+				abortSignal: options.signal,
+				// The synthetic `start` event is yielded immediately by every provider before
+				// the upstream model has emitted any tokens. Treating it as the first "real"
+				// item would flip the watchdog from `firstItemTimeoutMs` to the much shorter
+				// `idleTimeoutMs` while we're still legitimately waiting on the model's
+				// first response (slow first-token from reasoning models, cold proxies, etc.).
+				isProgressItem: event => (event as AssistantMessageEvent).type !== "start",
+			});
+
+			for await (const event of watchedSource) {
 				target.push(event);
 			}
 			if (hasFinalResult(source)) {
@@ -221,6 +269,7 @@ function createLazyLoadErrorMessage<TApi extends Api>(
 
 function createLazyStream<TApi extends Api>(
 	loadModule: () => Promise<LazyProviderModule<TApi>>,
+	limits?: LazyStreamLimits,
 ): (model: Model<TApi>, context: Context, options: OptionsForApi<TApi>) => EventStreamImpl {
 	return (model, context, options) => {
 		const outer = new EventStreamImpl();
@@ -231,7 +280,7 @@ function createLazyStream<TApi extends Api>(
 				const abortTracker = createAbortSourceTracker(streamOptions.signal);
 				const providerOptions = { ...streamOptions, signal: abortTracker.requestSignal } as OptionsForApi<TApi>;
 				const inner = module.stream(model, context, providerOptions);
-				forwardStream(outer, inner, model, streamOptions, abortTracker);
+				forwardStream(outer, inner, model, streamOptions, abortTracker, limits);
 			})
 			.catch(error => {
 				const message = createLazyLoadErrorMessage(model, error);
@@ -349,7 +398,10 @@ function loadBedrockProviderModule(): Promise<LazyProviderModule<"bedrock-conver
 export const streamAnthropic = createLazyStream(loadAnthropicProviderModule);
 export const streamAzureOpenAIResponses = createLazyStream(loadAzureOpenAIResponsesProviderModule);
 export const streamGoogle = createLazyStream(loadGoogleProviderModule);
-export const streamGoogleGeminiCli = createLazyStream(loadGoogleGeminiCliProviderModule);
+export const streamGoogleGeminiCli = createLazyStream(
+	loadGoogleGeminiCliProviderModule,
+	GOOGLE_GEMINI_CLI_LAZY_STREAM_LIMITS,
+);
 export const streamGoogleVertex = createLazyStream(loadGoogleVertexProviderModule);
 export const streamOpenAICodexResponses = createLazyStream(loadOpenAICodexResponsesProviderModule);
 export const streamOpenAICompletions = createLazyStream(loadOpenAICompletionsProviderModule);

packages/ai/src/stream.ts

@@ -564,6 +564,7 @@ function mapOptionsForApi<TApi extends Api>(
 		maxRetryDelayMs: options?.maxRetryDelayMs,
 		metadata: options?.metadata,
 		sessionId: options?.sessionId,
+		promptCacheKey: options?.promptCacheKey,
 		providerSessionState: options?.providerSessionState,
 		onPayload: options?.onPayload,
 		onResponse: options?.onResponse,

packages/ai/src/types.ts

@@ -316,11 +316,17 @@ export interface StreamOptions {
 	 */
 	metadata?: Record<string, unknown>;
 	/**
-	 * Optional session identifier for providers that support session-based caching.
-	 * Providers can use this to enable prompt caching, request routing, or other
-	 * session-aware features. Ignored by providers that don't support it.
+	 * Optional session identifier for providers that support session-based
+	 * routing, request affinity, or transport reuse. Providers may also use this
+	 * as the prompt-cache key when `promptCacheKey` is not set.
 	 */
 	sessionId?: string;
+	/**
+	 * Optional prompt-cache identity. When set, OpenAI Responses-compatible
+	 * providers use this for `prompt_cache_key` while keeping `sessionId` for
+	 * provider routing / conversation headers.
+	 */
+	promptCacheKey?: string;
 	/**
 	 * Provider-scoped mutable state store for this agent session.
 	 * Providers can use this to persist transport/session state between turns.
@@ -346,23 +352,26 @@ export interface StreamOptions {
 	 */
 	onSseEvent?: (event: RawSseEvent, model?: Model<Api>) => void;
 	/**
-	 * Optional SDK/request timeout hint in milliseconds applied to the underlying HTTP
-	 * request when the provider's transport exposes a per-request timeout.
+	 * Optional override for the first-event watchdog in milliseconds. When the
+	 * underlying transport exposes a per-request timeout (OpenAI/Anthropic SDKs),
+	 * this value is also applied to the HTTP layer so the SDK gives up before the
+	 * wrapping iterator watchdog would have. Set to `0` to disable both watchdogs
+	 * for this request. Falls back to `PI_STREAM_FIRST_EVENT_TIMEOUT_MS` and then
+	 * to a 100s default.
 	 *
-	 * Honored by: `openai-completions`, `openai-responses`, `azure-openai-responses`,
-	 *   `anthropic-messages`.
-	 * Ignored by: `openai-codex-responses` (uses its own websocket/SSE transport),
-	 *   `google`, `google-gemini-cli`, `google-vertex`, `bedrock-converse`, `cursor-agent`.
-	 *
-	 * Provider stream silence is never treated as failure on its own — once the request
-	 * has started, callers must abort to interrupt a silent stream.
+	 * Iterator-level honored by: every built-in provider (via the lazy-stream
+	 * forwarder in `register-builtins`). SDK-request honored by:
+	 * `openai-completions`, `openai-responses`, `azure-openai-responses`,
+	 * `anthropic-messages`.
 	 */
 	streamFirstEventTimeoutMs?: number;
 	/**
-	 * @deprecated Stream idle watchdogs were removed; provider streams now wait for
-	 * provider output, provider/socket errors, caller aborts, or request-layer timeouts.
-	 * This field is accepted for backwards compatibility but no longer wired anywhere.
-	 * Will be removed in the next major release.
+	 * Optional override for the maximum idle gap between streamed events in
+	 * milliseconds. Once the first event arrives, this guards against silent
+	 * mid-stream stalls (broker dies, half-open socket, model produces no real
+	 * progress for too long). Set to `0` to disable. Falls back to
+	 * `PI_STREAM_IDLE_TIMEOUT_MS` (alias: `PI_OPENAI_STREAM_IDLE_TIMEOUT_MS`)
+	 * and then to a 120s default.
 	 */
 	streamIdleTimeoutMs?: number;
 	/**

packages/ai/src/utils/idle-iterator.ts

@@ -0,0 +1,250 @@
+import { $env } from "@oh-my-pi/pi-utils";
+
+const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 120_000;
+const DEFAULT_STREAM_FIRST_EVENT_TIMEOUT_MS = 100_000;
+
+function normalizeIdleTimeoutMs(value: string | undefined, fallback: number): number | undefined {
+	if (value === undefined) return fallback;
+	const parsed = Number(value);
+	if (!Number.isFinite(parsed)) return fallback;
+	if (parsed <= 0) return undefined;
+	return Math.trunc(parsed);
+}
+
+/**
+ * Returns the idle timeout used for provider streaming transports.
+ *
+ * `PI_OPENAI_STREAM_IDLE_TIMEOUT_MS` is accepted as a backward-compatible alias.
+ * Set `PI_STREAM_IDLE_TIMEOUT_MS=0` to disable the watchdog.
+ *
+ * Providers that legitimately stream much slower than the global default can pass
+ * `fallbackMs` to widen the floor used when neither env var nor caller option is set.
+ * Caller options still take precedence; env overrides still trump the fallback.
+ */
+export function getStreamIdleTimeoutMs(fallbackMs: number = DEFAULT_STREAM_IDLE_TIMEOUT_MS): number | undefined {
+	return normalizeIdleTimeoutMs($env.PI_STREAM_IDLE_TIMEOUT_MS ?? $env.PI_OPENAI_STREAM_IDLE_TIMEOUT_MS, fallbackMs);
+}
+
+/**
+ * Returns the idle timeout used for OpenAI-family streaming transports.
+ *
+ * Set `PI_OPENAI_STREAM_IDLE_TIMEOUT_MS=0` to disable the watchdog.
+ */
+export function getOpenAIStreamIdleTimeoutMs(): number | undefined {
+	return normalizeIdleTimeoutMs(
+		$env.PI_OPENAI_STREAM_IDLE_TIMEOUT_MS ?? $env.PI_STREAM_IDLE_TIMEOUT_MS,
+		DEFAULT_STREAM_IDLE_TIMEOUT_MS,
+	);
+}
+
+/**
+ * Returns the timeout used while waiting for the first stream event.
+ * The first token can legitimately take longer than later inter-event gaps,
+ * so the default never undershoots the steady-state idle timeout.
+ *
+ * Set `PI_STREAM_FIRST_EVENT_TIMEOUT_MS=0` to disable the watchdog.
+ *
+ * Providers whose first response can legitimately take longer (heavy reasoning,
+ * slow cold-start proxies) can pass `fallbackMs` to widen the floor used when
+ * neither env var nor caller option is set. Caller options still take precedence;
+ * env overrides still trump the fallback.
+ */
+export function getStreamFirstEventTimeoutMs(
+	idleTimeoutMs?: number,
+	fallbackMs: number = DEFAULT_STREAM_FIRST_EVENT_TIMEOUT_MS,
+): number | undefined {
+	const fallback = idleTimeoutMs === undefined ? fallbackMs : Math.max(fallbackMs, idleTimeoutMs);
+	return normalizeIdleTimeoutMs($env.PI_STREAM_FIRST_EVENT_TIMEOUT_MS, fallback);
+}
+
+export type Watchdog = NodeJS.Timeout | undefined;
+
+const dummyWatchdog = setTimeout(() => {}, 1);
+clearTimeout(dummyWatchdog);
+
+/**
+ * Starts a watchdog that aborts a request if no first stream event arrives in time.
+ * Call `markFirstEventReceived()` as soon as the first event is observed.
+ */
+export function createWatchdog(timeoutMs: number | undefined, onTimeout: () => void): Watchdog {
+	if (timeoutMs !== undefined && timeoutMs > 0) {
+		return setTimeout(onTimeout, timeoutMs);
+	}
+	return undefined;
+}
+
+export interface IdleTimeoutIteratorOptions {
+	watchdog?: Watchdog;
+	idleTimeoutMs?: number;
+	firstItemTimeoutMs?: number;
+	errorMessage: string;
+	firstItemErrorMessage?: string;
+	onIdle?: () => void;
+	onFirstItemTimeout?: () => void;
+	/**
+	 * Optional semantic-progress predicate. Non-progress items are still yielded,
+	 * but they do not reset the idle deadline. This prevents provider
+	 * keepalive/no-op events from keeping a stalled tool call alive forever.
+	 */
+	isProgressItem?: (item: unknown) => boolean;
+	/**
+	 * Cancel iteration as soon as this signal aborts. Required for caller-driven
+	 * cancellation (ESC) when the underlying transport does not surface signal
+	 * aborts to the iterator (HTTP/2 proxies, native sockets, mocked fetch).
+	 * Without this, the consumer sleeps on iterator.next() until the idle/first
+	 * -event watchdog fires — observable as the issue #912 "Working… forever"
+	 * symptom on the github-copilot provider.
+	 */
+	abortSignal?: AbortSignal;
+}
+
+/**
+ * Yields items from an async iterable while enforcing a maximum idle gap between items.
+ *
+ * The first item may use a shorter timeout so stuck requests can be aborted and retried
+ * before any user-visible content has streamed.
+ */
+export async function* iterateWithIdleTimeout<T>(
+	iterable: AsyncIterable<T>,
+	options: IdleTimeoutIteratorOptions,
+): AsyncGenerator<T> {
+	let watchdog = options.watchdog;
+	const firstItemTimeoutMs = options.firstItemTimeoutMs ?? options.idleTimeoutMs;
+	const abortSignal = options.abortSignal;
+	const iterator = iterable[Symbol.asyncIterator]();
+
+	const closeIterator = (): void => {
+		const returnPromise = iterator.return?.();
+		if (returnPromise) {
+			void returnPromise.catch(() => {});
+		}
+	};
+
+	if (abortSignal?.aborted) {
+		closeIterator();
+		throw abortReason(abortSignal);
+	}
+
+	const withRacy = <T>(promise: Promise<T>) =>
+		promise.then(
+			result => ({ kind: "next" as const, result }),
+			error => ({ kind: "error" as const, error }),
+		);
+
+	let awaitingFirstItem = true;
+	const markFirstItemReceived = () => {
+		watchdog && clearTimeout(watchdog);
+		watchdog = undefined;
+		awaitingFirstItem = false;
+	};
+	const isProgressItem = (item: T): boolean => {
+		if (!options.isProgressItem) return true;
+		try {
+			return options.isProgressItem(item);
+		} catch {
+			return true;
+		}
+	};
+	let lastProgressAt = Date.now();
+
+	const noTimeoutEnforced =
+		(firstItemTimeoutMs === undefined || firstItemTimeoutMs <= 0) &&
+		(options.idleTimeoutMs === undefined || options.idleTimeoutMs <= 0);
+
+	while (true) {
+		let activeTimeoutMs: number | undefined;
+		if (awaitingFirstItem) {
+			activeTimeoutMs = firstItemTimeoutMs;
+		} else if (options.idleTimeoutMs !== undefined && options.idleTimeoutMs > 0) {
+			activeTimeoutMs = options.idleTimeoutMs - (Date.now() - lastProgressAt);
+			if (activeTimeoutMs <= 0) {
+				options.onIdle?.();
+				closeIterator();
+				throw new Error(options.errorMessage);
+			}
+		}
+
+		const nextResultPromise = withRacy(iterator.next());
+
+		const racers: Array<
+			Promise<
+				| { kind: "next"; result: IteratorResult<T> }
+				| { kind: "error"; error: unknown }
+				| { kind: "timeout" }
+				| { kind: "abort" }
+			>
+		> = [nextResultPromise];
+
+		let timer: NodeJS.Timeout | undefined;
+		let resolveTimeout: ((value: { kind: "timeout" }) => void) | undefined;
+		const enforceTimeout = !noTimeoutEnforced && activeTimeoutMs !== undefined && activeTimeoutMs > 0;
+		if (enforceTimeout) {
+			const { promise, resolve } = Promise.withResolvers<{ kind: "timeout" }>();
+			resolveTimeout = resolve;
+			timer = setTimeout(() => resolve({ kind: "timeout" }), activeTimeoutMs);
+			racers.push(promise);
+		}
+
+		let abortListener: (() => void) | undefined;
+		let resolveAbort: ((value: { kind: "abort" }) => void) | undefined;
+		if (abortSignal) {
+			const { promise, resolve } = Promise.withResolvers<{ kind: "abort" }>();
+			resolveAbort = resolve;
+			abortListener = () => resolve({ kind: "abort" });
+			abortSignal.addEventListener("abort", abortListener, { once: true });
+			racers.push(promise);
+		}
+
+		try {
+			const outcome = await Promise.race(racers);
+			if (outcome.kind === "abort") {
+				closeIterator();
+				throw abortReason(abortSignal!);
+			}
+			if (outcome.kind === "timeout") {
+				if (!awaitingFirstItem) {
+					options.onIdle?.();
+				} else {
+					options.onFirstItemTimeout?.();
+				}
+				closeIterator();
+				throw new Error(
+					!awaitingFirstItem ? options.errorMessage : (options.firstItemErrorMessage ?? options.errorMessage),
+				);
+			}
+			if (outcome.kind === "error") {
+				throw outcome.error;
+			}
+			if (outcome.result.done) {
+				markFirstItemReceived();
+				return;
+			}
+			const item = outcome.result.value;
+			// Non-progress items (e.g. provider keepalives, synthetic `start` events that
+			// arrive before the model has produced any tokens) MUST NOT flip us out of
+			// `awaitingFirstItem`. Otherwise the next iteration switches from the (longer)
+			// first-item watchdog to the (shorter) idle watchdog while we're still waiting
+			// on the model's first real output.
+			if (isProgressItem(item)) {
+				markFirstItemReceived();
+				lastProgressAt = Date.now();
+			}
+			yield item;
+		} finally {
+			if (timer !== undefined) clearTimeout(timer);
+			// Resolve dangling promises so the racers don't leak (Promise.race is one-shot).
+			resolveTimeout?.({ kind: "timeout" });
+			if (abortListener && abortSignal) {
+				abortSignal.removeEventListener("abort", abortListener);
+			}
+			resolveAbort?.({ kind: "abort" });
+		}
+	}
+}
+
+function abortReason(signal: AbortSignal): Error {
+	const reason = signal.reason;
+	if (reason instanceof Error) return reason;
+	if (typeof reason === "string") return new Error(reason);
+	return new Error("Request was aborted");
+}

packages/ai/test/anthropic-stream-timeout.test.ts

@@ -90,13 +90,11 @@ function createSuccessfulAnthropicEvents(text: string): MockAnthropicEvent[] {
 function createAnthropicMockStream({
 	signal,
 	connectDelayMs = 0,
-	firstEventDelayMs = 0,
 	events,
 	hangAfterEvents = false,
 }: {
 	signal: AbortSignal | undefined;
 	connectDelayMs?: number;
-	firstEventDelayMs?: number;
 	events?: MockAnthropicEvent[];
 	hangAfterEvents?: boolean;
 }): MockAnthropicRequest {
@@ -111,9 +109,6 @@ function createAnthropicMockStream({
 				await waitForAbortAndThrowAbortError(signal);
 				return;
 			}
-			if (firstEventDelayMs > 0) {
-				await waitForDelayOrAbort(firstEventDelayMs, signal);
-			}
 			for (const event of events) {
 				yield event;
 			}
@@ -141,15 +136,14 @@ afterEach(() => {
 	// No shared globals to restore; keep hook so the suite stays explicit.
 });
 
-describe("anthropic stream silence", () => {
-	it("waits for delayed first stream events without retrying", async () => {
+describe("anthropic first-event timeout retries", () => {
+	it("retries when the provider never sends the first stream event", async () => {
 		let attempt = 0;
 		const create = ((_body: unknown, requestOptions?: { signal?: AbortSignal }) => {
 			attempt += 1;
 			return createAnthropicMockStream({
 				signal: requestOptions?.signal,
-				firstEventDelayMs: 10,
-				events: createSuccessfulAnthropicEvents("slow first event"),
+				events: attempt === 1 ? undefined : createSuccessfulAnthropicEvents("retry recovered"),
 			}) as never;
 		}) as unknown as Anthropic["messages"]["create"];
 		const client = { messages: { create } } as Anthropic;
@@ -161,10 +155,10 @@ describe("anthropic stream silence", () => {
 			providerRetryWait,
 		}).result();
 
-		expect(attempt).toBe(1);
-		expect(providerRetryWait).not.toHaveBeenCalled();
+		expect(attempt).toBe(2);
+		expect(providerRetryWait).toHaveBeenCalledWith(2000, undefined);
 		expect(result.stopReason).toBe("stop");
-		expect(result.content).toEqual([{ type: "text", text: "slow first event" }]);
+		expect(result.content).toEqual([{ type: "text", text: "retry recovered" }]);
 		expect(result.responseId).toBe("msg_retry_success");
 	});
 
@@ -209,7 +203,7 @@ describe("anthropic stream silence", () => {
 		expect(result.errorMessage).not.toBe("Anthropic stream timed out while waiting for the first event");
 		expect((result.errorMessage ?? "").toLowerCase()).toContain("abort");
 	});
-	it("waits through silent gaps between tool-call events until caller aborts", async () => {
+	it("fails hung Anthropic streams between tool-call events instead of waiting forever", async () => {
 		let attempt = 0;
 		const create = ((_body: unknown, requestOptions?: { signal?: AbortSignal }) => {
 			attempt += 1;
@@ -243,19 +237,16 @@ describe("anthropic stream silence", () => {
 			}) as never;
 		}) as unknown as Anthropic["messages"]["create"];
 		const client = { messages: { create } } as Anthropic;
-		const controller = new AbortController();
-		setTimeout(() => controller.abort(), 10);
 
 		const result = await streamAnthropic(model, context, {
 			client,
-			signal: controller.signal,
 			streamFirstEventTimeoutMs: 5000,
 			streamIdleTimeoutMs: 1,
 		}).result();
 
 		expect(attempt).toBe(1);
-		expect(result.stopReason).toBe("aborted");
-		expect(result.errorMessage).not.toBe("Anthropic stream stalled while waiting for the next event");
+		expect(result.stopReason).toBe("error");
+		expect(result.errorMessage).toBe("Anthropic stream stalled while waiting for the next event");
 		expect(result.content).toEqual([
 			{
 				type: "toolCall",

packages/ai/test/openai-codex-stream.test.ts

@@ -286,7 +286,6 @@ describe("openai-codex streaming", () => {
 		const result = await streamOpenAICodexResponses(model, context, {
 			apiKey: token,
 			signal: controller.signal,
-			streamIdleTimeoutMs: 5,
 		}).result();
 
 		expect(result.stopReason).toBe("aborted");
@@ -964,6 +963,48 @@ describe("openai-codex streaming", () => {
 		const streamResult = streamOpenAICodexResponses(model, context, { apiKey: token, sessionId });
 		await streamResult.result();
 	});
+	it("keeps prompt_cache_key separate from Codex conversation headers", async () => {
+		const tempDir = TempDir.createSync("@pi-codex-stream-");
+		setAgentDir(tempDir.path());
+
+		const token = createCodexTestToken();
+		const model = createCodexTestModel("https://chatgpt.com/backend-api");
+		const sessionId = "side-channel-session";
+		const promptCacheKey = "main-session-cache";
+		let capturedHeaders: Headers | undefined;
+		let capturedBody: Record<string, unknown> | undefined;
+
+		global.fetch = vi.fn(async (input: string | URL, init?: RequestInit) => {
+			const url = typeof input === "string" ? input : input.toString();
+			if (url === "https://api.github.com/repos/openai/codex/releases/latest") {
+				return new Response(JSON.stringify({ tag_name: "rust-v0.0.0" }), { status: 200 });
+			}
+			if (url.startsWith("https://raw.githubusercontent.com/openai/codex/")) {
+				return new Response("PROMPT", { status: 200, headers: { etag: '"etag"' } });
+			}
+			if (url === "https://chatgpt.com/backend-api/codex/responses") {
+				capturedHeaders = init?.headers instanceof Headers ? init.headers : new Headers(init?.headers);
+				capturedBody =
+					typeof init?.body === "string" ? (JSON.parse(init.body) as Record<string, unknown>) : undefined;
+				return new Response(createCompletedCodexSse("Hello"), {
+					status: 200,
+					headers: { "content-type": "text/event-stream" },
+				});
+			}
+			return new Response("not found", { status: 404 });
+		}) as unknown as typeof fetch;
+
+		await streamOpenAICodexResponses(model, createCodexTestContext(), {
+			apiKey: token,
+			sessionId,
+			promptCacheKey,
+		}).result();
+
+		expect(capturedHeaders?.get("conversation_id")).toBe(sessionId);
+		expect(capturedHeaders?.get("session_id")).toBe(sessionId);
+		expect(capturedHeaders?.get("x-client-request-id")).toBe(sessionId);
+		expect(capturedBody?.prompt_cache_key).toBe(promptCacheKey);
+	});
 
 	it("rejects gpt-5.3-codex minimal reasoning effort instead of clamping", async () => {
 		const tempDir = TempDir.createSync("@pi-codex-stream-");

packages/ai/test/openai-completions-progress-chunk.test.ts

@@ -0,0 +1,166 @@
+import { describe, expect, it } from "bun:test";
+import { isOpenAICompletionsProgressChunk } from "../src/providers/openai-completions";
+
+/**
+ * Contract: `isOpenAICompletionsProgressChunk` decides whether a streamed chunk
+ * resets the idle-watchdog deadline in `iterateWithIdleTimeout`. A false
+ * positive (counting a no-op chunk as progress) silently disables the
+ * watchdog and is the root cause of the z.ai/GLM-via-OpenRouter hang where
+ * a subagent stalled for hours with no error surfaced. A false negative is
+ * cheap (delays the watchdog by at most the first-event window).
+ */
+describe("isOpenAICompletionsProgressChunk", () => {
+	describe("non-progress chunks (MUST NOT reset the watchdog)", () => {
+		it("rejects null/non-object", () => {
+			expect(isOpenAICompletionsProgressChunk(null)).toBe(false);
+			expect(isOpenAICompletionsProgressChunk(undefined)).toBe(false);
+			expect(isOpenAICompletionsProgressChunk("hi")).toBe(false);
+			expect(isOpenAICompletionsProgressChunk(42)).toBe(false);
+		});
+
+		it("rejects empty {} keepalives", () => {
+			expect(isOpenAICompletionsProgressChunk({})).toBe(false);
+		});
+
+		it("rejects {choices: []} keepalives", () => {
+			expect(isOpenAICompletionsProgressChunk({ choices: [] })).toBe(false);
+		});
+
+		it("rejects role-only preambles", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { role: "assistant" } }],
+				}),
+			).toBe(false);
+		});
+
+		it("rejects empty-string content deltas", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { content: "" } }],
+				}),
+			).toBe(false);
+		});
+
+		it("rejects empty-array content deltas", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { content: [] } }],
+				}),
+			).toBe(false);
+		});
+
+		it("rejects empty tool_calls arrays", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { tool_calls: [] } }],
+				}),
+			).toBe(false);
+		});
+
+		it("rejects empty reasoning fields", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { reasoning: "" } }],
+				}),
+			).toBe(false);
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { reasoning_content: "" } }],
+				}),
+			).toBe(false);
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { reasoning_text: "" } }],
+				}),
+			).toBe(false);
+		});
+	});
+
+	describe("progress chunks (MUST reset the watchdog)", () => {
+		it("accepts a top-level usage chunk (terminal token report)", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					usage: { prompt_tokens: 12, completion_tokens: 4 },
+				}),
+			).toBe(true);
+		});
+
+		it("accepts choice-level usage", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ usage: { prompt_tokens: 12 } }],
+				}),
+			).toBe(true);
+		});
+
+		it("accepts finish_reason", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ finish_reason: "stop" }],
+				}),
+			).toBe(true);
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ finish_reason: "tool_calls" }],
+				}),
+			).toBe(true);
+		});
+
+		it("accepts text content deltas", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { content: "Hello" } }],
+				}),
+			).toBe(true);
+		});
+
+		it("accepts array-shape content parts (Mistral-style)", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { content: [{ type: "text", text: "Hi" }] } }],
+				}),
+			).toBe(true);
+		});
+
+		it("accepts tool call deltas", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [
+						{
+							delta: {
+								tool_calls: [{ id: "call_1", function: { name: "search", arguments: "{}" } }],
+							},
+						},
+					],
+				}),
+			).toBe(true);
+		});
+
+		it("accepts reasoning deltas in all three field names", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { reasoning: "thinking..." } }],
+				}),
+			).toBe(true);
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { reasoning_content: "thinking..." } }],
+				}),
+			).toBe(true);
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { reasoning_text: "thinking..." } }],
+				}),
+			).toBe(true);
+		});
+
+		it("accepts refusal deltas", () => {
+			expect(
+				isOpenAICompletionsProgressChunk({
+					choices: [{ delta: { refusal: "I can't help with that." } }],
+				}),
+			).toBe(true);
+		});
+	});
+});

packages/ai/test/openai-first-event-timeout.test.ts

@@ -151,20 +151,6 @@ function createDelayedFetch(delayMs: number, responseFactory: () => Response): t
 
 	return Object.assign(mockFetch, { preconnect: originalFetch.preconnect });
 }
-function createAbortObservingDelayedFetch(delayMs: number, onAbort: () => void): typeof fetch {
-	async function mockFetch(input: string | URL | Request, init?: RequestInit): Promise<Response> {
-		const signal = getRequestSignal(input, init);
-		try {
-			await waitForDelayOrAbort(delayMs, signal);
-		} catch (error) {
-			if (signal?.aborted) onAbort();
-			throw error;
-		}
-		return createOpenAIResponsesSuccessResponse();
-	}
-
-	return Object.assign(mockFetch, { preconnect: originalFetch.preconnect });
-}
 
 function createOpenAIResponsesSuccessResponse(): Response {
 	return createSseResponse([
@@ -227,22 +213,16 @@ function createOpenAICompletionsSuccessResponse(modelId: string): Response {
 	]);
 }
 
-async function expectProviderSilenceWaitsForCallerAbort(
-	run: (
-		signal: AbortSignal,
-		streamFirstEventTimeoutMs: number,
-	) => Promise<{ stopReason: string; errorMessage?: string }>,
-	unexpectedMessage: string,
+async function expectFirstEventTimeout(
+	run: (streamFirstEventTimeoutMs: number) => Promise<{ stopReason: string; errorMessage?: string }>,
+	expectedMessage: string,
 ): Promise<void> {
 	global.fetch = createHangingFetch();
-	const controller = new AbortController();
-	setTimeout(() => controller.abort(), 30);
 
-	const result = await run(controller.signal, 5);
+	const result = await run(20);
 
-	expect(result.stopReason).toBe("aborted");
-	expect(result.errorMessage).not.toBe(unexpectedMessage);
-	expect((result.errorMessage ?? "").toLowerCase()).toContain("abort");
+	expect(result.stopReason).toBe("error");
+	expect(result.errorMessage).toBe(expectedMessage);
 }
 
 async function expectCallerAbort(
@@ -270,12 +250,12 @@ function getFirstTextContent(result: { content: unknown[] }): TextContent | unde
 }
 
 async function expectDelayedRequestSetupSucceeds(
-	run: () => Promise<{ stopReason: string; content: unknown[] }>,
+	run: (streamFirstEventTimeoutMs: number) => Promise<{ stopReason: string; content: unknown[] }>,
 	responseFactory: () => Response,
 ): Promise<void> {
 	global.fetch = createDelayedFetch(30, responseFactory);
 
-	const result = await run();
+	const result = await run(20);
 
 	expect(result.stopReason).toBe("stop");
 	expect(getFirstTextContent(result)).toMatchObject({ type: "text", text: "Hello delayed" });
@@ -285,34 +265,30 @@ afterEach(() => {
 	global.fetch = originalFetch;
 });
 
-describe("OpenAI-family provider stream silence", () => {
-	it("waits for caller abort when OpenAI responses emits no first event", async () => {
-		await expectProviderSilenceWaitsForCallerAbort(
-			(signal, streamFirstEventTimeoutMs) =>
+describe("OpenAI-family first-event timeouts", () => {
+	it("surfaces the OpenAI responses first-event timeout message instead of a generic abort", async () => {
+		await expectFirstEventTimeout(
+			streamFirstEventTimeoutMs =>
 				streamOpenAIResponses(openAIResponsesModel, baseContext(), {
 					apiKey: "test-key",
-					signal,
 					streamFirstEventTimeoutMs,
 				}).result(),
 			"OpenAI responses stream timed out while waiting for the first event",
 		);
 	});
 
-	it("waits for caller abort when OpenAI responses streams only no-progress status events", async () => {
+	it("times out OpenAI responses streams that only emit no-progress status events", async () => {
 		global.fetch = ((input: string | URL | Request, init?: RequestInit) =>
 			Promise.resolve(createNoProgressOpenAIResponsesStream(getRequestSignal(input, init)))) as typeof fetch;
-		const controller = new AbortController();
-		setTimeout(() => controller.abort(), 30);
 
 		const result = await streamOpenAIResponses(openAIResponsesModel, baseContext(), {
 			apiKey: "test-key",
-			signal: controller.signal,
 			streamFirstEventTimeoutMs: 1_000,
-			streamIdleTimeoutMs: 5,
+			streamIdleTimeoutMs: 20,
 		}).result();
 
-		expect(result.stopReason).toBe("aborted");
-		expect(result.errorMessage).not.toBe("OpenAI responses stream stalled while waiting for the next event");
+		expect(result.stopReason).toBe("error");
+		expect(result.errorMessage).toBe("OpenAI responses stream stalled while waiting for the next event");
 		expect(result.content as unknown[]).toEqual([
 			{
 				type: "toolCall",
@@ -324,26 +300,24 @@ describe("OpenAI-family provider stream silence", () => {
 		]);
 	});
 
-	it("waits for caller abort when OpenAI completions emits no first event", async () => {
-		await expectProviderSilenceWaitsForCallerAbort(
-			(signal, streamFirstEventTimeoutMs) =>
+	it("surfaces the OpenAI completions first-event timeout message", async () => {
+		await expectFirstEventTimeout(
+			streamFirstEventTimeoutMs =>
 				streamOpenAICompletions(openAICompletionsModel, baseContext(), {
 					apiKey: "test-key",
-					signal,
 					streamFirstEventTimeoutMs,
 				}).result(),
 			"OpenAI completions stream timed out while waiting for the first event",
 		);
 	});
 
-	it("waits for caller abort when Azure OpenAI responses emits no first event", async () => {
-		await expectProviderSilenceWaitsForCallerAbort(
-			(signal, streamFirstEventTimeoutMs) =>
+	it("surfaces the Azure OpenAI responses first-event timeout message", async () => {
+		await expectFirstEventTimeout(
+			streamFirstEventTimeoutMs =>
 				streamAzureOpenAIResponses(azureOpenAIResponsesModel, baseContext(), {
 					apiKey: "test-key",
 					azureBaseUrl: azureOpenAIResponsesModel.baseUrl,
 					azureApiVersion: "v1",
-					signal,
 					streamFirstEventTimeoutMs,
 				}).result(),
 			"Azure OpenAI responses stream timed out while waiting for the first event",
@@ -388,69 +362,38 @@ describe("OpenAI-family provider stream silence", () => {
 		);
 	});
 
-	it("does not apply a default first-event timeout before OpenAI responses stream setup finishes", async () => {
+	it("does not arm the first-event watchdog before OpenAI responses stream setup finishes", async () => {
 		await expectDelayedRequestSetupSucceeds(
-			() =>
+			streamFirstEventTimeoutMs =>
 				streamOpenAIResponses(openAIResponsesModel, baseContext(), {
 					apiKey: "test-key",
+					streamFirstEventTimeoutMs,
 				}).result(),
 			createOpenAIResponsesSuccessResponse,
 		);
 	});
 
-	it("maps explicit OpenAI responses first-event timeout to the SDK request timeout", async () => {
-		let abortObserved = false;
-		global.fetch = createAbortObservingDelayedFetch(1_000, () => {
-			abortObserved = true;
-		});
-
-		const result = await streamOpenAIResponses(openAIResponsesModel, baseContext(), {
-			apiKey: "test-key",
-			streamFirstEventTimeoutMs: 10,
-		}).result();
-
-		expect(abortObserved).toBe(true);
-		expect(result.stopReason).toBe("error");
-		expect(getFirstTextContent(result)).toBeUndefined();
-	});
-
-	it("does not apply a default first-event timeout before OpenAI completions stream setup finishes", async () => {
+	it("does not arm the first-event watchdog before OpenAI completions stream setup finishes", async () => {
 		await expectDelayedRequestSetupSucceeds(
-			() =>
+			streamFirstEventTimeoutMs =>
 				streamOpenAICompletions(openAICompletionsModel, baseContext(), {
 					apiKey: "test-key",
+					streamFirstEventTimeoutMs,
 				}).result(),
 			() => createOpenAICompletionsSuccessResponse(openAICompletionsModel.id),
 		);
 	});
 
-	it("does not apply a default first-event timeout before Azure OpenAI responses setup finishes", async () => {
+	it("does not arm the first-event watchdog before Azure OpenAI responses setup finishes", async () => {
 		await expectDelayedRequestSetupSucceeds(
-			() =>
+			streamFirstEventTimeoutMs =>
 				streamAzureOpenAIResponses(azureOpenAIResponsesModel, baseContext(), {
 					apiKey: "test-key",
 					azureBaseUrl: azureOpenAIResponsesModel.baseUrl,
 					azureApiVersion: "v1",
+					streamFirstEventTimeoutMs,
 				}).result(),
 			createOpenAIResponsesSuccessResponse,
 		);
 	});
-
-	it("maps explicit Azure OpenAI responses first-event timeout to the SDK request timeout", async () => {
-		let abortObserved = false;
-		global.fetch = createAbortObservingDelayedFetch(1_000, () => {
-			abortObserved = true;
-		});
-
-		const result = await streamAzureOpenAIResponses(azureOpenAIResponsesModel, baseContext(), {
-			apiKey: "test-key",
-			azureBaseUrl: azureOpenAIResponsesModel.baseUrl,
-			azureApiVersion: "v1",
-			streamFirstEventTimeoutMs: 10,
-		}).result();
-
-		expect(abortObserved).toBe(true);
-		expect(result.stopReason).toBe("error");
-		expect(getFirstTextContent(result)).toBeUndefined();
-	});
 });

packages/ai/test/openai-responses-cache-affinity.test.ts

@@ -89,6 +89,16 @@ describe("openai-responses cache affinity", () => {
 		expect(captured.clientRequestId).toBe("session-123");
 		expect(captured.body?.prompt_cache_key).toBe("session-123");
 	});
+	it("keeps prompt cache key separate from OpenAI routing headers when both are provided", async () => {
+		const captured = await captureOpenAIResponseHeaders({
+			sessionId: "side-channel-456",
+			promptCacheKey: "session-123",
+		});
+
+		expect(captured.sessionId).toBe("side-channel-456");
+		expect(captured.clientRequestId).toBe("side-channel-456");
+		expect(captured.body?.prompt_cache_key).toBe("session-123");
+	});
 
 	it("lets explicit headers override the default OpenAI session routing headers", async () => {
 		const captured = await captureOpenAIResponseHeaders({

packages/ai/test/register-builtins.test.ts

@@ -93,35 +93,41 @@ describe("register-builtins lazy streams", () => {
 		expect(result.errorMessage).toContain("bedrock exploded");
 	});
 
-	it("forwards lazy provider streams without first-event or idle watchdogs", async () => {
+	it("turns idle lazy provider streams into retryable terminal errors", async () => {
 		const partialMessage = createAssistantMessage("stop");
-		const finalMessage = createAssistantMessage("stop");
-		finalMessage.content = [{ type: "text", text: "slow ok" }];
+		let providerSignal: AbortSignal | undefined;
 		const source = {
 			async *[Symbol.asyncIterator]() {
-				await Bun.sleep(30);
 				yield { type: "start", partial: partialMessage } as const;
-				await Bun.sleep(30);
-				yield { type: "text_delta", contentIndex: 0, delta: "slow ok", partial: finalMessage } as const;
+				yield { type: "text_delta", contentIndex: 0, delta: "hello", partial: partialMessage } as const;
+				const { promise, reject } = Promise.withResolvers<never>();
+				if (providerSignal?.aborted) {
+					reject(new Error("Request was aborted"));
+				}
+				providerSignal?.addEventListener("abort", () => reject(new Error("Request was aborted")), {
+					once: true,
+				});
+				await promise;
 			},
-			result: async () => finalMessage,
 		} as unknown as AssistantMessageEventStream;
 
 		setBedrockProviderModule({
-			streamBedrock: () => source,
+			streamBedrock: (_model, _context, options) => {
+				providerSignal = options.signal;
+				return source;
+			},
 		});
 
-		const stream = streamBedrock(createModel(), baseContext, {
-			streamFirstEventTimeoutMs: 10,
-			streamIdleTimeoutMs: 10,
-		});
+		const stream = streamBedrock(createModel(), baseContext, { streamIdleTimeoutMs: 10 });
 		const result = await Promise.race([stream.result(), Bun.sleep(500).then(() => "timeout" as const)]);
 
 		expect(result).not.toBe("timeout");
 		if (result === "timeout") {
-			throw new Error("Timed out waiting for forwarded slow stream result");
+			throw new Error("Timed out waiting for forwarded stream stall result");
 		}
-		expect(result).toEqual(finalMessage);
+		expect(providerSignal?.aborted).toBe(true);
+		expect(result.stopReason).toBe("error");
+		expect(result.errorMessage).toBe("Provider stream stalled while waiting for the next event");
 	});
 
 	it("preserves caller aborts while forwarding lazy provider streams", async () => {

packages/ai/test/stream-timeout-defaults.test.ts

@@ -0,0 +1,81 @@
+import { afterEach, beforeEach, describe, expect, it } from "bun:test";
+import { getStreamFirstEventTimeoutMs, getStreamIdleTimeoutMs } from "../src/utils/idle-iterator";
+
+/**
+ * Per-provider fallback overrides on the stream-watchdog helpers.
+ *
+ * These are the gear that lets `google-gemini-cli` widen its first-event floor
+ * beyond the 100s global default without forcing every other provider to wait
+ * just as long. Tests pin the precedence contract callers depend on:
+ * caller option > env var > per-provider fallback > base default.
+ */
+
+const ENV_KEYS = [
+	"PI_STREAM_IDLE_TIMEOUT_MS",
+	"PI_OPENAI_STREAM_IDLE_TIMEOUT_MS",
+	"PI_STREAM_FIRST_EVENT_TIMEOUT_MS",
+] as const;
+
+const originalEnv: Partial<Record<(typeof ENV_KEYS)[number], string | undefined>> = {};
+
+beforeEach(() => {
+	for (const key of ENV_KEYS) {
+		originalEnv[key] = Bun.env[key];
+		delete Bun.env[key];
+	}
+});
+
+afterEach(() => {
+	for (const key of ENV_KEYS) {
+		const prior = originalEnv[key];
+		if (prior === undefined) {
+			delete Bun.env[key];
+		} else {
+			Bun.env[key] = prior;
+		}
+	}
+});
+
+describe("getStreamIdleTimeoutMs(fallbackMs)", () => {
+	it("returns the per-provider fallback when env vars are unset", () => {
+		expect(getStreamIdleTimeoutMs(300_000)).toBe(300_000);
+	});
+
+	it("lets PI_STREAM_IDLE_TIMEOUT_MS override the per-provider fallback", () => {
+		Bun.env.PI_STREAM_IDLE_TIMEOUT_MS = "42";
+		expect(getStreamIdleTimeoutMs(300_000)).toBe(42);
+	});
+
+	it("treats PI_STREAM_IDLE_TIMEOUT_MS=0 as a watchdog disable", () => {
+		Bun.env.PI_STREAM_IDLE_TIMEOUT_MS = "0";
+		expect(getStreamIdleTimeoutMs(300_000)).toBeUndefined();
+	});
+});
+
+describe("getStreamFirstEventTimeoutMs(idleTimeoutMs, fallbackMs)", () => {
+	it("returns the per-provider fallback when env unset and idle timeout is undefined", () => {
+		expect(getStreamFirstEventTimeoutMs(undefined, 300_000)).toBe(300_000);
+	});
+
+	it("floors the first-event timeout at the per-provider fallback even when idle is shorter", () => {
+		expect(getStreamFirstEventTimeoutMs(50_000, 300_000)).toBe(300_000);
+	});
+
+	it("never undershoots the steady-state idle timeout", () => {
+		expect(getStreamFirstEventTimeoutMs(500_000, 300_000)).toBe(500_000);
+	});
+
+	it("lets PI_STREAM_FIRST_EVENT_TIMEOUT_MS override the per-provider fallback", () => {
+		Bun.env.PI_STREAM_FIRST_EVENT_TIMEOUT_MS = "42";
+		expect(getStreamFirstEventTimeoutMs(undefined, 300_000)).toBe(42);
+	});
+
+	it("treats PI_STREAM_FIRST_EVENT_TIMEOUT_MS=0 as a watchdog disable", () => {
+		Bun.env.PI_STREAM_FIRST_EVENT_TIMEOUT_MS = "0";
+		expect(getStreamFirstEventTimeoutMs(undefined, 300_000)).toBeUndefined();
+	});
+
+	it("falls back to the 100s global default when no fallback or env is provided", () => {
+		expect(getStreamFirstEventTimeoutMs()).toBe(100_000);
+	});
+});