diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index c86b3176c..b6a83feb0 100644 --- a/packages/ai/CHANGELOG.md +++ b/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 diff --git a/packages/ai/src/auth-gateway/server.ts b/packages/ai/src/auth-gateway/server.ts index 47233f994..4292bfd21 100644 --- a/packages/ai/src/auth-gateway/server.ts +++ b/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 }; } diff --git a/packages/ai/src/auth-gateway/types.ts b/packages/ai/src/auth-gateway/types.ts index 34e5c4b3e..7d9b6fefe 100644 --- a/packages/ai/src/auth-gateway/types.ts +++ b/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; diff --git a/packages/ai/src/providers/anthropic.ts b/packages/ai/src/providers/anthropic.ts index 1ba4eb02c..2069b48a2 100644 --- a/packages/ai/src/providers/anthropic.ts +++ b/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; diff --git a/packages/ai/src/providers/azure-openai-responses.ts b/packages/ai/src/providers/azure-openai-responses.ts index af1eb779a..949caa197 100644 --- a/packages/ai/src/providers/azure-openai-responses.ts +++ b/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 { const map = new Map(); @@ -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); diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index b6eda8ece..ce8297b47 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/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; + transportSessionId?: string; providerSessionState?: CodexProviderSessionState; websocketState?: CodexWebSocketSessionState; transformedBody: RequestBody; @@ -167,6 +194,8 @@ interface CodexRequestContext { interface CodexRequestSetup { requestSignal: AbortSignal; + wrapCodexSseStream: (source: AsyncGenerator>) => AsyncGenerator>; + 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>, + ): AsyncGenerator> => + 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 { 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 | undefined, +): string | undefined { + return normalizeOpenAIResponsesPromptCacheKey(options?.promptCacheKey ?? options?.sessionId); +} + +function resolveCodexTransportSessionId( + options: Pick | 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 | 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); diff --git a/packages/ai/src/providers/openai-completions.ts b/packages/ai/src/providers/openai-completions.ts index 9e30cc716..4b01b5fea 100644 --- a/packages/ai/src/providers/openai-completions.ts +++ b/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, diff --git a/packages/ai/src/providers/openai-responses.ts b/packages/ai/src/providers/openai-responses.ts index bb22d3cd8..b976791f1 100644 --- a/packages/ai/src/providers/openai-responses.ts +++ b/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> = []; - 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(item) as unknown as Record); + }, }, - onOutputItemDone: item => { - nativeOutputItems.push(structuredCloneJSON(item) as unknown as Record); - }, - }); + ); 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 | undefined, +): string | undefined { + if (resolveCacheRetention(options?.cacheRetention) === "none") return undefined; + return normalizeOpenAIResponsesPromptCacheKey(options?.promptCacheKey ?? options?.sessionId); +} + +function getOpenAIResponsesRoutingSessionId( options: Pick | 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, diff --git a/packages/ai/src/providers/pi-native-server.ts b/packages/ai/src/providers/pi-native-server.ts index 2be0e9f7e..6092aeca4 100644 --- a/packages/ai/src/providers/pi-native-server.ts +++ b/packages/ai/src/providers/pi-native-server.ts @@ -58,6 +58,7 @@ const ALLOWED_OPTION_KEYS: ReadonlySet = new Set([ "maxRetryDelayMs", "metadata", "sessionId", + "promptCacheKey", "streamFirstEventTimeoutMs", "streamIdleTimeoutMs", "reasoning", diff --git a/packages/ai/src/providers/register-builtins.ts b/packages/ai/src/providers/register-builtins.ts index 64a90d9a9..9e88eb499 100644 --- a/packages/ai/src/providers/register-builtins.ts +++ b/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, ): source is AsyncIterable & { result(): Promise } { 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( target: EventStreamImpl, source: AsyncIterable, model: Model, options: OptionsForApi, 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( function createLazyStream( loadModule: () => Promise>, + limits?: LazyStreamLimits, ): (model: Model, context: Context, options: OptionsForApi) => EventStreamImpl { return (model, context, options) => { const outer = new EventStreamImpl(); @@ -231,7 +280,7 @@ function createLazyStream( const abortTracker = createAbortSourceTracker(streamOptions.signal); const providerOptions = { ...streamOptions, signal: abortTracker.requestSignal } as OptionsForApi; 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( maxRetryDelayMs: options?.maxRetryDelayMs, metadata: options?.metadata, sessionId: options?.sessionId, + promptCacheKey: options?.promptCacheKey, providerSessionState: options?.providerSessionState, onPayload: options?.onPayload, onResponse: options?.onResponse, diff --git a/packages/ai/src/types.ts b/packages/ai/src/types.ts index d2c256781..0a3e3ab01 100644 --- a/packages/ai/src/types.ts +++ b/packages/ai/src/types.ts @@ -316,11 +316,17 @@ export interface StreamOptions { */ metadata?: Record; /** - * 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) => 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; /** diff --git a/packages/ai/src/utils/idle-iterator.ts b/packages/ai/src/utils/idle-iterator.ts new file mode 100644 index 000000000..1c0ab903c --- /dev/null +++ b/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( + iterable: AsyncIterable, + options: IdleTimeoutIteratorOptions, +): AsyncGenerator { + 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 = (promise: Promise) => + 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 } + | { 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"); +} diff --git a/packages/ai/test/anthropic-stream-timeout.test.ts b/packages/ai/test/anthropic-stream-timeout.test.ts index fda30d3d8..e745b698f 100644 --- a/packages/ai/test/anthropic-stream-timeout.test.ts +++ b/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", diff --git a/packages/ai/test/openai-codex-stream.test.ts b/packages/ai/test/openai-codex-stream.test.ts index 6b565ca1c..ef97fc9c9 100644 --- a/packages/ai/test/openai-codex-stream.test.ts +++ b/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 | 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) : 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-"); diff --git a/packages/ai/test/openai-completions-progress-chunk.test.ts b/packages/ai/test/openai-completions-progress-chunk.test.ts new file mode 100644 index 000000000..6c685a959 --- /dev/null +++ b/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); + }); + }); +}); diff --git a/packages/ai/test/openai-first-event-timeout.test.ts b/packages/ai/test/openai-first-event-timeout.test.ts index aa7c88eb7..78e7de631 100644 --- a/packages/ai/test/openai-first-event-timeout.test.ts +++ b/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 { - 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 { 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 { 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(); - }); }); diff --git a/packages/ai/test/openai-responses-cache-affinity.test.ts b/packages/ai/test/openai-responses-cache-affinity.test.ts index 31a8cb7dc..93ae55db4 100644 --- a/packages/ai/test/openai-responses-cache-affinity.test.ts +++ b/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({ diff --git a/packages/ai/test/register-builtins.test.ts b/packages/ai/test/register-builtins.test.ts index f4dfc51f4..8e6628786 100644 --- a/packages/ai/test/register-builtins.test.ts +++ b/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(); + 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 () => { diff --git a/packages/ai/test/stream-timeout-defaults.test.ts b/packages/ai/test/stream-timeout-defaults.test.ts new file mode 100644 index 000000000..15efa15a4 --- /dev/null +++ b/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> = {}; + +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); + }); +});