From bb61d8430bd14f75e7054f2099c2c1eb71c221d5 Mon Sep 17 00:00:00 2001 From: Bogdan Date: Wed, 1 Jul 2026 01:56:50 +0200 Subject: [PATCH] fix(openai-shim): wire stream controller abort (#1828) * fix(openai-shim): wire stream controller abort * test(openai-shim): guard Ollama abort fixture cleanup --- src/services/api/codexShim.test.ts | 163 ++++++ src/services/api/codexShim.ts | 552 ++++++++++-------- src/services/api/openaiShim.test.ts | 862 ++++++++++++++++++++++++++++ src/services/api/openaiShim.ts | 425 +++++++++++--- 4 files changed, 1687 insertions(+), 315 deletions(-) diff --git a/src/services/api/codexShim.test.ts b/src/services/api/codexShim.test.ts index 22bb88676..df7134660 100644 --- a/src/services/api/codexShim.test.ts +++ b/src/services/api/codexShim.test.ts @@ -70,6 +70,64 @@ async function collectStreamEventTypes(responseText: string): Promise return events } +async function waitForPromise( + promise: Promise, + timeoutMs: number, + message: string, +): Promise { + let timer: ReturnType | undefined + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(message)), timeoutMs) + }), + ]) + } finally { + if (timer !== undefined) { + clearTimeout(timer) + } + } +} + +function makeStallingCodexResponse(firstChunk: string): { + response: Response + cancelReasons: unknown[] + close: () => void +} { + const encoder = new TextEncoder() + const cancelReasons: unknown[] = [] + let streamController: ReadableStreamDefaultController | undefined + let closed = false + + const response = new Response( + new ReadableStream({ + start(controller) { + streamController = controller + controller.enqueue(encoder.encode(firstChunk)) + }, + cancel(reason) { + closed = true + cancelReasons.push(reason) + }, + }), + ) + + return { + response, + cancelReasons, + close: () => { + if (closed) return + closed = true + try { + streamController?.close() + } catch { + // The test may already have cancelled the stream. + } + }, + } +} + async function importFreshProviderConfigModule() { return import(`./providerConfig.js?ts=${Date.now()}-${Math.random()}`) } @@ -1016,6 +1074,111 @@ describe('Codex request translation', () => { ]) }) + test('Codex stream: abort signal cancels source while paused after message_start', async () => { + const stalled = makeStallingCodexResponse([ + 'event: response.output_text.delta', + 'data: {"type":"response.output_text.delta","content_index":0,"delta":"partial","item_id":"msg_1","output_index":0,"sequence_number":0}', + '', + ].join('\n')) + const controller = new AbortController() + const iterator = codexStreamToAnthropic( + stalled.response, + 'gpt-5.4', + controller.signal, + )[Symbol.asyncIterator]() + + try { + const first = await waitForPromise( + iterator.next(), + 500, + 'Codex stream did not produce message_start', + ) + expect(first.done).toBe(false) + expect(first.value?.type).toBe('message_start') + + controller.abort() + await waitForPromise( + (async () => { + for (let i = 0; i < 10; i++) { + if (stalled.cancelReasons.length > 0) return + await new Promise(resolve => setTimeout(resolve, 0)) + } + throw new Error('Codex stream did not cancel source on abort') + })(), + 500, + 'Codex stream did not cancel source on abort', + ) + + expect(stalled.cancelReasons).toHaveLength(1) + expect((stalled.cancelReasons[0] as { name?: unknown }).name).toBe('AbortError') + } finally { + await Promise.resolve(iterator.return?.(undefined)).catch(() => {}) + stalled.close() + } + }) + + test('Codex stream: abort signal stops buffered events after emitted delta', async () => { + const stalled = makeStallingCodexResponse([ + 'event: response.output_text.delta', + 'data: {"type":"response.output_text.delta","content_index":0,"delta":"first","item_id":"msg_1","output_index":0,"sequence_number":0}', + '', + 'event: response.output_text.delta', + 'data: {"type":"response.output_text.delta","content_index":0,"delta":"second","item_id":"msg_1","output_index":0,"sequence_number":1}', + '', + ].join('\n')) + const controller = new AbortController() + const iterator = codexStreamToAnthropic( + stalled.response, + 'gpt-5.4', + controller.signal, + )[Symbol.asyncIterator]() + + try { + const messageStart = await waitForPromise( + iterator.next(), + 500, + 'Codex stream did not produce message_start', + ) + expect(messageStart.done).toBe(false) + expect(messageStart.value?.type).toBe('message_start') + + const blockStart = await waitForPromise( + iterator.next(), + 500, + 'Codex stream did not produce content_block_start', + ) + expect(blockStart.done).toBe(false) + expect(blockStart.value?.type).toBe('content_block_start') + + const firstDelta = await waitForPromise( + iterator.next(), + 500, + 'Codex stream did not produce first delta', + ) + expect(firstDelta.done).toBe(false) + expect(firstDelta.value?.type).toBe('content_block_delta') + expect((firstDelta.value as { delta?: { text?: string } }).delta?.text).toBe('first') + + controller.abort() + const afterAbort = await waitForPromise( + iterator.next().then( + value => ({ status: 'resolved' as const, value }), + error => ({ status: 'rejected' as const, error }), + ), + 500, + 'Codex stream did not stop after abort', + ) + + if (afterAbort.status !== 'rejected') { + throw new Error(`Codex stream yielded after abort: ${JSON.stringify(afterAbort.value)}`) + } + expect((afterAbort.error as { name?: unknown }).name).toBe('AbortError') + } finally { + await Promise.resolve(iterator.return?.(undefined)).catch(() => {}) + stalled.close() + } + }) + test('strips tag block from Codex SSE text stream', async () => { const responseText = [ 'event: response.output_item.added', diff --git a/src/services/api/codexShim.ts b/src/services/api/codexShim.ts index 87fdd1226..09e0cc0f8 100644 --- a/src/services/api/codexShim.ts +++ b/src/services/api/codexShim.ts @@ -81,6 +81,42 @@ type CodexSseEvent = { data: Record } +function createStreamAbortError(): DOMException { + return new DOMException('Aborted', 'AbortError') +} + +function throwIfStreamAborted(signal?: AbortSignal): void { + if (signal?.aborted) { + throw createStreamAbortError() + } +} + +function createReaderCanceller( + reader: ReadableStreamDefaultReader, + signal?: AbortSignal, +): { + cancel: (error?: unknown) => void + cleanup: () => void + } { + let cancelled = false + const cancel = (error: unknown = createStreamAbortError()) => { + if (cancelled) return + cancelled = true + void reader.cancel(error).catch(() => {}) + } + const onAbort = () => cancel(createStreamAbortError()) + + signal?.addEventListener('abort', onAbort, { once: true }) + if (signal?.aborted) { + onAbort() + } + + return { + cancel, + cleanup: () => signal?.removeEventListener('abort', onAbort), + } +} + function makeUsage(usage?: Record): AnthropicUsage { // Single source of truth for raw → Anthropic shape. Lives in // cacheMetrics.ts alongside the raw-shape extractor so any new @@ -671,11 +707,13 @@ export async function performCodexRequest(options: { async function* readSseEvents(response: Response, signal?: AbortSignal): AsyncGenerator { const reader = response.body?.getReader() if (!reader) return + const readerCanceller = createReaderCanceller(reader, signal) const decoder = new TextDecoder() let buffer = '' const STREAM_IDLE_TIMEOUT_MS = 120_000 // 2 minutes without data let lastDataTime = Date.now() + let streamComplete = false /** * Read from the stream with an idle timeout. Respects the caller's @@ -684,72 +722,104 @@ async function* readSseEvents(response: Response, signal?: AbortSignal): AsyncGe */ async function readWithTimeout(): Promise>> { return new Promise((resolve, reject) => { - const timeoutId = setTimeout(() => { + let settled = false + let timeoutId: ReturnType | undefined = setTimeout(() => { const elapsed = Math.round((Date.now() - lastDataTime) / 1000) - reject(new Error( + cancelAndReject(new Error( `Codex SSE stream idle for ${elapsed}s (limit: ${STREAM_IDLE_TIMEOUT_MS / 1000}s). Connection likely dropped.`, )) }, STREAM_IDLE_TIMEOUT_MS) - let abortCleanup: (() => void) | undefined - if (signal) { - abortCleanup = () => { + const cleanup = () => { + if (timeoutId !== undefined) { clearTimeout(timeoutId) + timeoutId = undefined } - signal.addEventListener('abort', abortCleanup, { once: true }) + signal?.removeEventListener('abort', onAbort) + } + const finishResolve = ( + value: Bun.ReadableStreamDefaultReadResult>, + ) => { + if (settled) return + settled = true + cleanup() + if (value.value) lastDataTime = Date.now() + resolve(value) + } + const finishReject = (error: unknown) => { + if (settled) return + settled = true + cleanup() + reject(error) + } + const cancelAndReject = (error: unknown) => { + readerCanceller.cancel(error) + finishReject(error) + } + const onAbort = () => cancelAndReject(createStreamAbortError()) + + signal?.addEventListener('abort', onAbort, { once: true }) + if (signal?.aborted) { + onAbort() + return } // reader is guarded non-null above; hoisted function escapes TS narrowing. reader!.read().then( - result => { - clearTimeout(timeoutId) - if (signal && abortCleanup) signal.removeEventListener('abort', abortCleanup) - if (result.value) lastDataTime = Date.now() - resolve(result) - }, - err => { - clearTimeout(timeoutId) - if (signal && abortCleanup) signal.removeEventListener('abort', abortCleanup) - reject(err) - }, + result => finishResolve(result), + err => finishReject(err), ) }) } - while (true) { - const { done, value } = await readWithTimeout() - if (done) break - - buffer += decoder.decode(value, { stream: true }) - const chunks = buffer.split('\n\n') - buffer = chunks.pop() ?? '' - - for (const chunk of chunks) { - const lines = chunk - .split('\n') - .map(line => line.trim()) - .filter(Boolean) - if (lines.length === 0) continue - - const eventLine = lines.find(line => line.startsWith('event: ')) - const dataLines = lines.filter(line => line.startsWith('data: ')) - if (!eventLine || dataLines.length === 0) continue - - const event = eventLine.slice(7).trim() - const rawData = dataLines.map(line => line.slice(6)).join('\n') - if (rawData === '[DONE]') continue - - let data: Record - try { - const parsed = JSON.parse(rawData) - if (!parsed || typeof parsed !== 'object') continue - data = parsed as Record - } catch { - continue + try { + while (true) { + const { done, value } = await readWithTimeout() + if (done) { + streamComplete = true + break } - yield { event, data } + throwIfStreamAborted(signal) + buffer += decoder.decode(value, { stream: true }) + const chunks = buffer.split('\n\n') + buffer = chunks.pop() ?? '' + + for (const chunk of chunks) { + throwIfStreamAborted(signal) + const lines = chunk + .split('\n') + .map(line => line.trim()) + .filter(Boolean) + if (lines.length === 0) continue + + const eventLine = lines.find(line => line.startsWith('event: ')) + const dataLines = lines.filter(line => line.startsWith('data: ')) + if (!eventLine || dataLines.length === 0) continue + + const event = eventLine.slice(7).trim() + const rawData = dataLines.map(line => line.slice(6)).join('\n') + if (rawData === '[DONE]') continue + + let data: Record + try { + const parsed = JSON.parse(rawData) + if (!parsed || typeof parsed !== 'object') continue + data = parsed as Record + } catch { + continue + } + + throwIfStreamAborted(signal) + yield { event, data } + } } + } finally { + if (!streamComplete || signal?.aborted) { + readerCanceller.cancel(createStreamAbortError()) + } + readerCanceller.cleanup() + reader.releaseLock() } } @@ -823,11 +893,17 @@ export async function* codexStreamToAnthropic( let nextContentBlockIndex = 0 let sawToolUse = false let finalResponse: Record | undefined + let streamComplete = false + const cancelResponseBody = () => { + void response.body?.cancel(createStreamAbortError()).catch(() => {}) + } + signal?.addEventListener('abort', cancelResponseBody, { once: true }) const closeActiveTextBlock = async function* () { if (activeTextBlockIndex === null) return const tail = thinkFilter.flush() if (tail) { + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: activeTextBlockIndex, @@ -837,6 +913,7 @@ export async function* codexStreamToAnthropic( }, } } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: activeTextBlockIndex, @@ -847,6 +924,7 @@ export async function* codexStreamToAnthropic( const startTextBlockIfNeeded = async function* () { if (activeTextBlockIndex !== null) return activeTextBlockIndex = nextContentBlockIndex++ + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: activeTextBlockIndex, @@ -854,205 +932,229 @@ export async function* codexStreamToAnthropic( } } - yield { - type: 'message_start', - message: { - id: messageId, - type: 'message', - role: 'assistant', - content: [], - model, - stop_reason: null, - stop_sequence: null, - usage: makeUsage(), - }, - } + try { + throwIfStreamAborted(signal) - for await (const event of readSseEvents(response, signal)) { - const payload = event.data - - if (event.event === 'response.output_item.added') { - const item = payload.item - if (item?.type === 'function_call') { - yield* closeActiveTextBlock() - const blockIndex = nextContentBlockIndex++ - const toolUseId = item.call_id ?? item.id ?? `call_${blockIndex}` - const initialArgs = - typeof item.arguments === 'string' ? item.arguments : '' - toolBlocksByItemId.set(String(item.id ?? toolUseId), { - index: blockIndex, - toolUseId, - emittedArgs: initialArgs, - }) - sawToolUse = true - - yield { - type: 'content_block_start', - index: blockIndex, - content_block: { - type: 'tool_use', - id: toolUseId, - name: item.name ?? 'tool', - input: {}, - }, - } - - if (initialArgs) { - yield { - type: 'content_block_delta', - index: blockIndex, - delta: { - type: 'input_json_delta', - partial_json: initialArgs, - }, - } - } - } - continue + yield { + type: 'message_start', + message: { + id: messageId, + type: 'message', + role: 'assistant', + content: [], + model, + stop_reason: null, + stop_sequence: null, + usage: makeUsage(), + }, } - if (event.event === 'response.content_part.added') { - if (payload.part?.type === 'output_text') { - yield* startTextBlockIfNeeded() - } - continue - } + for await (const event of readSseEvents(response, signal)) { + throwIfStreamAborted(signal) + const payload = event.data - if (event.event === 'response.output_text.delta') { - yield* startTextBlockIfNeeded() - if (activeTextBlockIndex !== null) { - const visible = thinkFilter.feed(payload.delta ?? '') - if (visible) { - yield { - type: 'content_block_delta', - index: activeTextBlockIndex, - delta: { - type: 'text_delta', - text: visible, - }, - } - } - } - continue - } - - if (event.event === 'response.function_call_arguments.delta') { - const toolBlock = toolBlocksByItemId.get(String(payload.item_id ?? '')) - if (toolBlock) { - const delta = typeof payload.delta === 'string' ? payload.delta : '' - if (delta) { - toolBlock.emittedArgs += delta - yield { - type: 'content_block_delta', - index: toolBlock.index, - delta: { - type: 'input_json_delta', - partial_json: delta, - }, - } - } - } - continue - } - - // Some Codex Responses backends (codexspark / gpt-5.3-codex-spark) deliver - // the *complete* function-call arguments only via the terminal - // `response.function_call_arguments.done` event, with zero - // `response.function_call_arguments.delta` events in between. Without - // handling `done`, the tool block closed with `input: {}` and downstream - // tool validation failed with "required parameter X is missing" (#1259). - if (event.event === 'response.function_call_arguments.done') { - const toolBlock = toolBlocksByItemId.get(String(payload.item_id ?? '')) - if (toolBlock) { - const fullArgs = - typeof payload.arguments === 'string' ? payload.arguments : '' - if (fullArgs && !toolBlock.emittedArgs) { - toolBlock.emittedArgs = fullArgs - yield { - type: 'content_block_delta', - index: toolBlock.index, - delta: { - type: 'input_json_delta', - partial_json: fullArgs, - }, - } - } - } - continue - } - - if (event.event === 'response.output_item.done') { - const item = payload.item - if (item?.type === 'function_call') { - const toolBlock = toolBlocksByItemId.get(String(item.id ?? '')) - if (toolBlock) { - // Backstop for backends that skip the dedicated `function_call_arguments.done` - // event entirely and only put the full arguments on `output_item.done`. - // Same #1259 failure mode; trust whichever channel actually carried the data. - const finalArgs = + if (event.event === 'response.output_item.added') { + const item = payload.item + if (item?.type === 'function_call') { + yield* closeActiveTextBlock() + throwIfStreamAborted(signal) + const blockIndex = nextContentBlockIndex++ + const toolUseId = item.call_id ?? item.id ?? `call_${blockIndex}` + const initialArgs = typeof item.arguments === 'string' ? item.arguments : '' - if (finalArgs && !toolBlock.emittedArgs) { - toolBlock.emittedArgs = finalArgs + toolBlocksByItemId.set(String(item.id ?? toolUseId), { + index: blockIndex, + toolUseId, + emittedArgs: initialArgs, + }) + sawToolUse = true + + throwIfStreamAborted(signal) + yield { + type: 'content_block_start', + index: blockIndex, + content_block: { + type: 'tool_use', + id: toolUseId, + name: item.name ?? 'tool', + input: {}, + }, + } + + if (initialArgs) { + throwIfStreamAborted(signal) + yield { + type: 'content_block_delta', + index: blockIndex, + delta: { + type: 'input_json_delta', + partial_json: initialArgs, + }, + } + } + } + continue + } + + if (event.event === 'response.content_part.added') { + if (payload.part?.type === 'output_text') { + yield* startTextBlockIfNeeded() + } + continue + } + + if (event.event === 'response.output_text.delta') { + yield* startTextBlockIfNeeded() + if (activeTextBlockIndex !== null) { + throwIfStreamAborted(signal) + const visible = thinkFilter.feed(payload.delta ?? '') + if (visible) { + throwIfStreamAborted(signal) + yield { + type: 'content_block_delta', + index: activeTextBlockIndex, + delta: { + type: 'text_delta', + text: visible, + }, + } + } + } + continue + } + + if (event.event === 'response.function_call_arguments.delta') { + const toolBlock = toolBlocksByItemId.get(String(payload.item_id ?? '')) + if (toolBlock) { + const delta = typeof payload.delta === 'string' ? payload.delta : '' + if (delta) { + toolBlock.emittedArgs += delta + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: toolBlock.index, delta: { type: 'input_json_delta', - partial_json: finalArgs, + partial_json: delta, }, } } - yield { - type: 'content_block_stop', - index: toolBlock.index, - } - toolBlocksByItemId.delete(String(item.id)) } - } else if (item?.type === 'message') { - yield* closeActiveTextBlock() + continue + } + + // Some Codex Responses backends (codexspark / gpt-5.3-codex-spark) deliver + // the *complete* function-call arguments only via the terminal + // `response.function_call_arguments.done` event, with zero + // `response.function_call_arguments.delta` events in between. Without + // handling `done`, the tool block closed with `input: {}` and downstream + // tool validation failed with "required parameter X is missing" (#1259). + if (event.event === 'response.function_call_arguments.done') { + const toolBlock = toolBlocksByItemId.get(String(payload.item_id ?? '')) + if (toolBlock) { + const fullArgs = + typeof payload.arguments === 'string' ? payload.arguments : '' + if (fullArgs && !toolBlock.emittedArgs) { + toolBlock.emittedArgs = fullArgs + throwIfStreamAborted(signal) + yield { + type: 'content_block_delta', + index: toolBlock.index, + delta: { + type: 'input_json_delta', + partial_json: fullArgs, + }, + } + } + } + continue + } + + if (event.event === 'response.output_item.done') { + const item = payload.item + if (item?.type === 'function_call') { + const toolBlock = toolBlocksByItemId.get(String(item.id ?? '')) + if (toolBlock) { + // Backstop for backends that skip the dedicated `function_call_arguments.done` + // event entirely and only put the full arguments on `output_item.done`. + // Same #1259 failure mode; trust whichever channel actually carried the data. + const finalArgs = + typeof item.arguments === 'string' ? item.arguments : '' + if (finalArgs && !toolBlock.emittedArgs) { + toolBlock.emittedArgs = finalArgs + throwIfStreamAborted(signal) + yield { + type: 'content_block_delta', + index: toolBlock.index, + delta: { + type: 'input_json_delta', + partial_json: finalArgs, + }, + } + } + throwIfStreamAborted(signal) + yield { + type: 'content_block_stop', + index: toolBlock.index, + } + toolBlocksByItemId.delete(String(item.id)) + } + } else if (item?.type === 'message') { + yield* closeActiveTextBlock() + } + continue + } + + if ( + event.event === 'response.completed' || + event.event === 'response.incomplete' + ) { + finalResponse = payload.response + break + } + + if (event.event === 'response.failed') { + const msg = payload?.response?.error?.message ?? + payload?.error?.message ?? 'Codex response failed' + throw APIError.generate(500, undefined, msg, new Headers()) } - continue } - if ( - event.event === 'response.completed' || - event.event === 'response.incomplete' - ) { - finalResponse = payload.response - break + throwIfStreamAborted(signal) + yield* closeActiveTextBlock() + for (const toolBlock of toolBlocksByItemId.values()) { + throwIfStreamAborted(signal) + yield { + type: 'content_block_stop', + index: toolBlock.index, + } } - if (event.event === 'response.failed') { - const msg = payload?.response?.error?.message ?? - payload?.error?.message ?? 'Codex response failed' - throw APIError.generate(500, undefined, msg, new Headers()) - } - } - - yield* closeActiveTextBlock() - for (const toolBlock of toolBlocksByItemId.values()) { + throwIfStreamAborted(signal) yield { - type: 'content_block_stop', - index: toolBlock.index, + type: 'message_delta', + delta: { + stop_reason: determineStopReason(finalResponse, sawToolUse), + stop_sequence: null, + }, + // Delegate to the shared normalizer so the streaming message_delta + // path uses the same raw→Anthropic conversion as makeUsage() above + // and the non-streaming response converter below. Previously this + // block had its own inline subtraction that missed Kimi / DeepSeek + // / Gemini raw shapes that the shared helper handles. + usage: makeUsage( + finalResponse?.usage as Record | undefined, + ), } + throwIfStreamAborted(signal) + yield { type: 'message_stop' } + streamComplete = true + } finally { + if (!streamComplete || signal?.aborted) { + cancelResponseBody() + } + signal?.removeEventListener('abort', cancelResponseBody) } - - yield { - type: 'message_delta', - delta: { - stop_reason: determineStopReason(finalResponse, sawToolUse), - stop_sequence: null, - }, - // Delegate to the shared normalizer so the streaming message_delta - // path uses the same raw→Anthropic conversion as makeUsage() above - // and the non-streaming response converter below. Previously this - // block had its own inline subtraction that missed Kimi / DeepSeek - // / Gemini raw shapes that the shared helper handles. - usage: makeUsage( - finalResponse?.usage as Record | undefined, - ), - } - yield { type: 'message_stop' } } export function convertCodexResponseToAnthropicMessage( diff --git a/src/services/api/openaiShim.test.ts b/src/services/api/openaiShim.test.ts index 8abf6a52b..438837d1e 100644 --- a/src/services/api/openaiShim.test.ts +++ b/src/services/api/openaiShim.test.ts @@ -94,6 +94,255 @@ function makeSseResponse(lines: string[]): Response { ) } +function withResponseUrl(response: Response, url: string): Response { + Object.defineProperty(response, 'url', { + value: url, + configurable: true, + }) + return response +} + +type StallingResponse = { + response: Response + cancelReasons: unknown[] + close: () => void +} + +function makeStallingResponse( + firstChunk: string, + url = 'https://api.example.test/v1/chat/completions', + contentType = 'text/event-stream', +): StallingResponse { + const encoder = new TextEncoder() + const cancelReasons: unknown[] = [] + let streamController: ReadableStreamDefaultController | undefined + let closed = false + + const response = new Response( + new ReadableStream({ + start(controller) { + streamController = controller + controller.enqueue(encoder.encode(firstChunk)) + }, + cancel(reason) { + closed = true + cancelReasons.push(reason) + }, + }), + { + headers: { + 'Content-Type': contentType, + }, + }, + ) + + return { + response: withResponseUrl(response, url), + cancelReasons, + close: () => { + if (closed) return + closed = true + try { + streamController?.close() + } catch { + // The test may already have cancelled the stream. + } + }, + } +} + +type ShimStream = AsyncIterable> & { + controller: AbortController +} + +type StreamDrainOutcome = + | { status: 'completed'; events: Array> } + | { + status: 'rejected' + events: Array> + error: unknown + } + +async function waitForPromise( + promise: Promise, + timeoutMs: number, + message: string, +): Promise { + let timer: ReturnType | undefined + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(message)), timeoutMs) + }), + ]) + } finally { + if (timer !== undefined) { + clearTimeout(timer) + } + } +} + +async function expectAbortStopsStream({ + abort, + cancelReasons, + expectedEventsBeforeAbort, + label, + stream, +}: { + abort: () => void + cancelReasons: unknown[] + expectedEventsBeforeAbort: number + label: string + stream: ShimStream +}): Promise { + const events: Array> = [] + let resolveReady!: () => void + const ready = new Promise(resolve => { + resolveReady = resolve + }) + + const drain = (async (): Promise => { + try { + for await (const event of stream) { + events.push(event) + if (events.length >= expectedEventsBeforeAbort) { + resolveReady() + } + } + return { status: 'completed', events } + } catch (error) { + return { status: 'rejected', events, error } + } + })() + + await waitForPromise( + ready, + 500, + `${label} did not produce initial stream events`, + ) + // Let the for-await loop ask the stream reader for the next chunk, so the + // abort has to wake a real pending read rather than only flipping a flag. + await Promise.resolve() + await Promise.resolve() + + abort() + + const outcome = await waitForPromise( + drain, + 500, + `${label} did not stop promptly after abort`, + ) + expect(cancelReasons).toHaveLength(1) + expect(outcome.status).toBe('rejected') + if (outcome.status === 'rejected') { + expect((outcome.error as { name?: unknown }).name).toBe('AbortError') + } + return outcome +} + +async function expectPausedAbortCancelsStream({ + cancelReasons, + label, + stream, +}: { + cancelReasons: unknown[] + label: string + stream: ShimStream +}): Promise>> { + const iterator = stream[Symbol.asyncIterator]() + const first = await waitForPromise( + iterator.next(), + 500, + `${label} did not produce first stream event`, + ) + expect(first.done).toBe(false) + + stream.controller.abort() + await waitForPromise( + (async () => { + for (let i = 0; i < 10; i++) { + if (cancelReasons.length > 0) return + await Promise.resolve() + } + throw new Error(`${label} did not cancel source on controller abort`) + })(), + 500, + `${label} did not cancel source on controller abort`, + ) + + const returned = await waitForPromise( + Promise.resolve(iterator.return?.()), + 500, + `${label} did not return promptly after abort while paused`, + ) + expect(cancelReasons).toHaveLength(1) + return returned as IteratorResult> +} + +async function expectBufferedAbortRejectsNext({ + expectedText, + label, + stream, +}: { + expectedText?: string + label: string + stream: ShimStream +}): Promise { + const iterator = stream[Symbol.asyncIterator]() + + try { + let firstDelta: Record | undefined + for (let i = 0; i < 5; i++) { + const next = await waitForPromise( + iterator.next(), + 500, + `${label} did not produce expected pre-abort events`, + ) + expect(next.done).toBe(false) + if (next.value?.type === 'content_block_delta') { + firstDelta = next.value + break + } + } + + expect(firstDelta).toBeDefined() + if (expectedText !== undefined) { + expect((firstDelta as { delta?: { text?: string } }).delta?.text).toBe(expectedText) + } + + stream.controller.abort() + const afterAbort = await waitForPromise( + iterator.next().then( + value => ({ status: 'resolved' as const, value }), + error => ({ status: 'rejected' as const, error }), + ), + 500, + `${label} did not stop after abort`, + ) + + if (afterAbort.status !== 'rejected') { + throw new Error(`${label} yielded after abort: ${JSON.stringify(afterAbort.value)}`) + } + expect((afterAbort.error as { name?: unknown }).name).toBe('AbortError') + } finally { + await Promise.resolve(iterator.return?.()).catch(() => {}) + } +} + +function makeOpenAIStreamFrame( + delta: Record, + finishReason: string | null = null, +): string { + return `data: ${JSON.stringify({ + id: 'chatcmpl-abort-test', + object: 'chat.completion.chunk', + created: 1_780_000_000, + model: 'test-model', + choices: [{ index: 0, delta, finish_reason: finishReason }], + })}\n\n` +} + function makeStreamChunks(chunks: unknown[]): string[] { return [ ...chunks.map(chunk => `data: ${JSON.stringify(chunk)}\n\n`), @@ -1226,6 +1475,619 @@ test('preserves usage from final OpenAI stream chunk with empty choices', async expect(usageEvent?.usage?.output_tokens).toBe(45) }) +test('controller abort reaches generic OpenAI SSE converter', async () => { + const stalled = makeStallingResponse( + makeOpenAIStreamFrame({ role: 'assistant', content: 'partial' }), + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + const outcome = await expectAbortStopsStream({ + abort: () => stream.controller.abort(), + cancelReasons: stalled.cancelReasons, + expectedEventsBeforeAbort: 3, + label: 'generic OpenAI SSE stream', + stream, + }) + + expect(outcome.events.some(event => event.type === 'content_block_delta')).toBe(true) + } finally { + stalled.close() + } +}) + +test('controller abort cancels generic OpenAI SSE before iteration starts', async () => { + const stalled = makeStallingResponse( + makeOpenAIStreamFrame({ role: 'assistant', content: 'partial' }), + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + stream.controller.abort() + await waitForPromise( + (async () => { + for (let i = 0; i < 10; i++) { + if (stalled.cancelReasons.length > 0) return + await Promise.resolve() + } + throw new Error('pre-iteration OpenAI SSE stream did not cancel source') + })(), + 500, + 'pre-iteration OpenAI SSE stream did not cancel source', + ) + expect(stalled.cancelReasons).toHaveLength(1) + } finally { + stalled.close() + } +}) + +test('controller abort cancels generic OpenAI SSE when paused after message_start', async () => { + const stalled = makeStallingResponse( + makeOpenAIStreamFrame({ role: 'assistant', content: 'partial' }), + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + await expectPausedAbortCancelsStream({ + cancelReasons: stalled.cancelReasons, + label: 'paused generic OpenAI SSE stream', + stream, + }) + } finally { + stalled.close() + } +}) + +test('controller abort stops buffered generic OpenAI SSE events', async () => { + const stalled = makeStallingResponse( + makeOpenAIStreamFrame({ role: 'assistant', content: 'first' }) + + makeOpenAIStreamFrame({ content: 'second' }), + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + await expectBufferedAbortRejectsNext({ + expectedText: 'first', + label: 'buffered generic OpenAI SSE stream', + stream, + }) + expect(stalled.cancelReasons).toHaveLength(1) + } finally { + stalled.close() + } +}) + +test('controller abort reaches Anthropic messages SSE passthrough', async () => { + const stalled = makeStallingResponse( + `data: ${JSON.stringify({ + type: 'message_start', + message: { + id: 'msg_passthrough_abort', + type: 'message', + role: 'assistant', + content: [], + model: 'passthrough-model', + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 0, output_tokens: 0 }, + }, + })}\n\n`, + 'https://api.anthropic-shaped.example.com/v1/messages', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'passthrough-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + const outcome = await expectAbortStopsStream({ + abort: () => stream.controller.abort(), + cancelReasons: stalled.cancelReasons, + expectedEventsBeforeAbort: 1, + label: 'Anthropic messages passthrough stream', + stream, + }) + + expect(outcome.events[0]?.type).toBe('message_start') + } finally { + stalled.close() + } +}) + +test('controller abort cancels Anthropic messages SSE when paused after event', async () => { + const stalled = makeStallingResponse( + `data: ${JSON.stringify({ + type: 'message_start', + message: { + id: 'msg_paused_passthrough_abort', + type: 'message', + role: 'assistant', + content: [], + model: 'passthrough-model', + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 0, output_tokens: 0 }, + }, + })}\n\n`, + 'https://api.anthropic-shaped.example.com/v1/messages', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'passthrough-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + await expectPausedAbortCancelsStream({ + cancelReasons: stalled.cancelReasons, + label: 'paused Anthropic messages passthrough stream', + stream, + }) + } finally { + stalled.close() + } +}) + +test('controller abort stops buffered Anthropic messages SSE events', async () => { + const stalled = makeStallingResponse( + [ + `data: ${JSON.stringify({ + type: 'message_start', + message: { + id: 'msg_buffered_passthrough_abort', + type: 'message', + role: 'assistant', + content: [], + model: 'passthrough-model', + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 0, output_tokens: 0 }, + }, + })}`, + '', + `data: ${JSON.stringify({ + type: 'content_block_start', + index: 0, + content_block: { type: 'text', text: '' }, + })}`, + '', + '', + ].join('\n'), + 'https://api.anthropic-shaped.example.com/v1/messages', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'passthrough-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + const iterator = stream[Symbol.asyncIterator]() + + try { + const first = await waitForPromise( + iterator.next(), + 500, + 'buffered Anthropic messages passthrough did not produce first event', + ) + expect(first.done).toBe(false) + expect(first.value?.type).toBe('message_start') + + stream.controller.abort() + const afterAbort = await waitForPromise( + iterator.next().then( + value => ({ status: 'resolved' as const, value }), + error => ({ status: 'rejected' as const, error }), + ), + 500, + 'buffered Anthropic messages passthrough did not stop after abort', + ) + + if (afterAbort.status !== 'rejected') { + throw new Error(`buffered Anthropic messages passthrough yielded after abort: ${JSON.stringify(afterAbort.value)}`) + } + expect((afterAbort.error as { name?: unknown }).name).toBe('AbortError') + expect(stalled.cancelReasons).toHaveLength(1) + } finally { + await Promise.resolve(iterator.return?.()).catch(() => {}) + stalled.close() + } +}) + +test('parent signal abort still reaches OpenAI SSE converter', async () => { + const stalled = makeStallingResponse( + makeOpenAIStreamFrame({ role: 'assistant', content: 'partial' }), + ) + const parent = new AbortController() + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create( + { + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }, + { signal: parent.signal }, + ) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + const outcome = await expectAbortStopsStream({ + abort: () => parent.abort(), + cancelReasons: stalled.cancelReasons, + expectedEventsBeforeAbort: 3, + label: 'parent-aborted OpenAI SSE stream', + stream, + }) + + expect(outcome.events.some(event => event.type === 'content_block_delta')).toBe(true) + } finally { + stalled.close() + } +}) + +test('parent signal abort cancels OpenAI SSE before iteration starts', async () => { + const stalled = makeStallingResponse( + makeOpenAIStreamFrame({ role: 'assistant', content: 'partial' }), + ) + const parent = new AbortController() + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create( + { + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }, + { signal: parent.signal }, + ) + .withResponse() + expect(result.data).toBeDefined() + + try { + parent.abort() + await waitForPromise( + (async () => { + for (let i = 0; i < 10; i++) { + if (stalled.cancelReasons.length > 0) return + await Promise.resolve() + } + throw new Error('pre-iteration parent-aborted OpenAI SSE stream did not cancel source') + })(), + 500, + 'pre-iteration parent-aborted OpenAI SSE stream did not cancel source', + ) + expect(stalled.cancelReasons).toHaveLength(1) + } finally { + stalled.close() + } +}) + +test('controller abort reaches Codex responses stream converter', async () => { + const stalled = makeStallingResponse( + `event: response.output_text.delta\ndata: ${JSON.stringify({ delta: 'partial' })}\n\n`, + 'https://api.example.test/v1/responses', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'gpt-5.4', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + const outcome = await expectAbortStopsStream({ + abort: () => stream.controller.abort(), + cancelReasons: stalled.cancelReasons, + expectedEventsBeforeAbort: 3, + label: 'Codex responses stream', + stream, + }) + + expect(outcome.events.some(event => event.type === 'content_block_delta')).toBe(true) + } finally { + stalled.close() + } +}) + +test('controller abort cancels Codex responses stream when paused after message_start', async () => { + const stalled = makeStallingResponse( + `event: response.output_text.delta\ndata: ${JSON.stringify({ delta: 'partial' })}\n\n`, + 'https://api.example.test/v1/responses', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'gpt-5.4', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + await expectPausedAbortCancelsStream({ + cancelReasons: stalled.cancelReasons, + label: 'paused Codex responses stream', + stream, + }) + } finally { + stalled.close() + } +}) + +test('controller abort reaches Gemini SSE converter', async () => { + const stalled = makeStallingResponse( + `data: ${JSON.stringify({ + candidates: [ + { + content: { + role: 'model', + parts: [{ text: 'partial' }], + }, + }, + ], + })}\n\n`, + 'https://generativelanguage.googleapis.com/v1beta/models/gemini-3.1-pro-preview:streamGenerateContent?alt=sse', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'google/gemini-3.1-pro-preview', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + const outcome = await expectAbortStopsStream({ + abort: () => stream.controller.abort(), + cancelReasons: stalled.cancelReasons, + expectedEventsBeforeAbort: 3, + label: 'Gemini SSE stream', + stream, + }) + + expect(outcome.events.some(event => event.type === 'content_block_delta')).toBe(true) + } finally { + stalled.close() + } +}) + +test('controller abort stops buffered Gemini SSE events', async () => { + const makeGeminiFrame = (text: string) => + `data: ${JSON.stringify({ + candidates: [ + { + content: { + role: 'model', + parts: [{ text }], + }, + }, + ], + })}\n\n` + const stalled = makeStallingResponse( + makeGeminiFrame('first') + makeGeminiFrame('second'), + 'https://generativelanguage.googleapis.com/v1beta/models/gemini-3.1-pro-preview:streamGenerateContent?alt=sse', + ) + + globalThis.fetch = (async () => stalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'google/gemini-3.1-pro-preview', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + try { + await expectBufferedAbortRejectsNext({ + expectedText: 'first', + label: 'buffered Gemini SSE stream', + stream, + }) + expect(stalled.cancelReasons).toHaveLength(1) + } finally { + stalled.close() + } +}) + +test('controller abort reaches native Ollama converted stream', async () => { + const previousBaseUrl = process.env.OPENAI_BASE_URL + let stalled: StallingResponse | undefined + + try { + process.env.OPENAI_BASE_URL = 'http://localhost:11434/v1' + stalled = makeStallingResponse( + `${JSON.stringify({ + model: 'llama3.1:8b', + message: { role: 'assistant', content: 'partial' }, + done: false, + })}\n`, + 'http://localhost:11434/api/chat', + 'application/x-ndjson', + ) + const activeStalled = stalled + + globalThis.fetch = (async () => activeStalled.response) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'llama3.1:8b', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + const stream = result.data as unknown as ShimStream + + const outcome = await expectAbortStopsStream({ + abort: () => stream.controller.abort(), + cancelReasons: activeStalled.cancelReasons, + expectedEventsBeforeAbort: 1, + label: 'native Ollama converted stream', + stream, + }) + + expect(outcome.events[0]?.type).toBe('message_start') + } finally { + stalled?.close() + restoreEnv('OPENAI_BASE_URL', previousBaseUrl) + } +}) + +test('normal OpenAI SSE stream still completes after controller wiring', async () => { + globalThis.fetch = (async () => + makeSseResponse(makeStreamChunks([ + { + id: 'chatcmpl-normal-stream', + object: 'chat.completion.chunk', + model: 'fake-model', + choices: [ + { + index: 0, + delta: { role: 'assistant', content: 'complete' }, + finish_reason: null, + }, + ], + }, + { + id: 'chatcmpl-normal-stream', + object: 'chat.completion.chunk', + model: 'fake-model', + choices: [ + { + index: 0, + delta: {}, + finish_reason: 'stop', + }, + ], + }, + ]))) as unknown as FetchType + + const client = createOpenAIShimClient({}) as OpenAIShimClient + const result = await client.beta.messages + .create({ + model: 'fake-model', + messages: [{ role: 'user', content: 'hello' }], + max_tokens: 64, + stream: true, + }) + .withResponse() + + const textDeltas: string[] = [] + for await (const event of result.data) { + const delta = (event as { delta?: { type?: string; text?: string } }).delta + if (delta?.type === 'text_delta' && typeof delta.text === 'string') { + textDeltas.push(delta.text) + } + } + + expect(textDeltas.join('')).toBe('complete') + expect((result.data as unknown as ShimStream).controller.signal.aborted).toBe(false) +}) + test('uses max_tokens instead of max_completion_tokens for local providers', async () => { process.env.OPENAI_BASE_URL = 'http://localhost:11434/v1' diff --git a/src/services/api/openaiShim.ts b/src/services/api/openaiShim.ts index c6017ed03..07b310b17 100644 --- a/src/services/api/openaiShim.ts +++ b/src/services/api/openaiShim.ts @@ -93,6 +93,7 @@ import { import { sanitizeSchemaForOpenAICompat } from '../../utils/schemaSanitizer.js' import { redactSecretValueForDisplay, type SecretValueSource } from '../../utils/providerProfile.js' import { shouldRedactUrlQueryParam } from '../../utils/urlRedaction.js' +import { createCombinedAbortSignal } from '../../utils/combinedAbortSignal.js' import { normalizeToolArguments, hasToolFieldMapping, @@ -129,6 +130,90 @@ function isCopilotTokenExpiredError(text: string): boolean { return lower.includes('token expired') || lower.includes('token has expired') } +function createStreamAbortError(): DOMException { + return new DOMException('Aborted', 'AbortError') +} + +function throwIfStreamAborted(signal?: AbortSignal): void { + if (signal?.aborted) { + throw createStreamAbortError() + } +} + +type StreamReadResult = Awaited< + ReturnType['read']> +> + +function createReaderCanceller( + reader: ReadableStreamDefaultReader, + signal?: AbortSignal, +): { + cancel: (error?: unknown) => void + cleanup: () => void + } { + let cancelled = false + const cancel = (error: unknown = createStreamAbortError()) => { + if (cancelled) return + cancelled = true + void reader.cancel(error).catch(() => {}) + } + const onAbort = () => cancel(createStreamAbortError()) + + signal?.addEventListener('abort', onAbort, { once: true }) + if (signal?.aborted) { + onAbort() + } + + return { + cancel, + cleanup: () => signal?.removeEventListener('abort', onAbort), + } +} + +function readWithAbort( + reader: ReadableStreamDefaultReader, + signal?: AbortSignal, + cancelReader?: (error?: unknown) => void, +): Promise { + if (!signal) return reader.read() + + return new Promise((resolve, reject) => { + let settled = false + const cleanup = () => { + signal.removeEventListener('abort', onAbort) + } + const finishResolve = (value: StreamReadResult) => { + if (settled) return + settled = true + cleanup() + resolve(value) + } + const finishReject = (error: unknown) => { + if (settled) return + settled = true + cleanup() + reject(error) + } + const cancelAndReject = (error: unknown) => { + if (cancelReader) { + cancelReader(error) + } else { + void reader.cancel(error).catch(() => {}) + } + finishReject(error) + } + const onAbort = () => cancelAndReject(createStreamAbortError()) + + signal.addEventListener('abort', onAbort, { once: true }) + if (signal.aborted) { + onAbort() + return + } + + reader.read().then(finishResolve, finishReject) + }) +} + function isGithubModelsMode(): boolean { return isEnvTruthy(process.env.CLAUDE_CODE_USE_GITHUB) } @@ -1766,33 +1851,26 @@ async function* anthropicSsePassthrough( const readerOrNull = response.body?.getReader() if (!readerOrNull) throw new Error('Response body is not readable') const reader: ReadableStreamDefaultReader = readerOrNull + const readerCanceller = createReaderCanceller(reader, signal) const decoder = new TextDecoder() let buffer = '' - - // Read helper that properly cleans up abort listeners (mirrors codexShim.ts pattern). - type ReadResult = Awaited> - function readWithAbort(): Promise { - if (!signal) return reader.read() - return new Promise((resolve, reject) => { - const onAbort = () => reject(new DOMException('Aborted', 'AbortError')) - signal.addEventListener('abort', onAbort, { once: true }) - reader.read().then( - result => { signal.removeEventListener('abort', onAbort); resolve(result) }, - err => { signal.removeEventListener('abort', onAbort); reject(err) }, - ) - }) - } + let streamComplete = false try { while (true) { - const { done, value } = await readWithAbort() - if (done) break + const { done, value } = await readWithAbort(reader, signal, readerCanceller.cancel) + if (done) { + streamComplete = true + break + } + throwIfStreamAborted(signal) buffer += decoder.decode(value, { stream: true }) const chunks = buffer.split('\n\n') buffer = chunks.pop() ?? '' for (const chunk of chunks) { + throwIfStreamAborted(signal) const lines = chunk.split('\n').map(l => l.trim()).filter(Boolean) if (lines.length === 0) continue @@ -1800,19 +1878,29 @@ async function* anthropicSsePassthrough( if (dataLines.length === 0) continue const rawData = dataLines.map(l => l.slice(6)).join('\n') - if (rawData === '[DONE]') return + if (rawData === '[DONE]') { + streamComplete = true + return + } + let parsed: AnthropicStreamEvent try { - const parsed = JSON.parse(rawData) as AnthropicStreamEvent - if (parsed && typeof parsed === 'object' && 'type' in parsed) { - yield parsed - } + parsed = JSON.parse(rawData) as AnthropicStreamEvent } catch { // skip malformed frames + continue + } + if (parsed && typeof parsed === 'object' && 'type' in parsed) { + throwIfStreamAborted(signal) + yield parsed } } } } finally { + if (!streamComplete || signal?.aborted) { + readerCanceller.cancel(createStreamAbortError()) + } + readerCanceller.cleanup() reader.releaseLock() } } @@ -1828,6 +1916,7 @@ async function* geminiSseToAnthropic( ): AsyncGenerator { const reader: ReadableStreamDefaultReader | undefined = response.body?.getReader() if (!reader) throw new Error('Response body is not readable') + const readerCanceller = createReaderCanceller(reader, signal) const decoder = new TextDecoder() let buffer = '' const messageId = makeMessageId() @@ -1837,18 +1926,7 @@ async function* geminiSseToAnthropic( let hasEmittedCurrentTool = false let usage: Partial | undefined let finishReason: string | undefined - - function readWithAbort(): Promise> { - if (!signal) return reader!.read() as Promise> - return new Promise((resolve, reject) => { - const onAbort = () => reject(new DOMException('Aborted', 'AbortError')) - signal.addEventListener('abort', onAbort, { once: true }) - reader!.read().then( - result => { signal.removeEventListener('abort', onAbort); resolve(result as ReadableStreamReadResult) }, - err => { signal.removeEventListener('abort', onAbort); reject(err) }, - ) - }) - } + let streamComplete = false function mapFinishReason(reason: string | undefined, hasToolUse: boolean): string { if (hasToolUse) return 'tool_use' @@ -1858,14 +1936,19 @@ async function* geminiSseToAnthropic( try { while (true) { - const { done, value } = await readWithAbort() - if (done) break + const { done, value } = await readWithAbort(reader, signal, readerCanceller.cancel) + if (done) { + streamComplete = true + break + } + throwIfStreamAborted(signal) buffer += decoder.decode(value, { stream: true }) const chunks = buffer.split('\n\n') buffer = chunks.pop() ?? '' for (const chunk of chunks) { + throwIfStreamAborted(signal) const lines = chunk.split('\n').map(l => l.trim()).filter(Boolean) const dataLines = lines.filter(l => l.startsWith('data: ')) if (dataLines.length === 0) continue @@ -1873,14 +1956,18 @@ async function* geminiSseToAnthropic( const rawData = dataLines.map(l => l.slice(6)).join('\n') if (rawData === '[DONE]') { if (hasEmittedTextStart || hasEmittedCurrentTool) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } } + throwIfStreamAborted(signal) yield { type: 'message_delta', delta: { stop_reason: mapFinishReason(finishReason, hasEmittedCurrentTool) }, usage: usage ?? {}, } + throwIfStreamAborted(signal) yield { type: 'message_stop' } + streamComplete = true return } @@ -1892,6 +1979,7 @@ async function* geminiSseToAnthropic( } if (!hasEmittedStart) { + throwIfStreamAborted(signal) yield { type: 'message_start', message: { @@ -1928,16 +2016,19 @@ async function* geminiSseToAnthropic( if (!content || !content.parts) continue for (const part of content.parts) { + throwIfStreamAborted(signal) const text = part.text as string | undefined const fc = part.functionCall as { name?: string; args?: unknown } | undefined if (text) { if (hasEmittedCurrentTool) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } contentBlockIndex++ hasEmittedCurrentTool = false } if (!hasEmittedTextStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -1945,6 +2036,7 @@ async function* geminiSseToAnthropic( } hasEmittedTextStart = true } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -1952,11 +2044,13 @@ async function* geminiSseToAnthropic( } } else if (fc?.name) { if (hasEmittedTextStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } contentBlockIndex++ hasEmittedTextStart = false } const toolId = `toolu_${crypto.randomUUID().replace(/-/g, '').slice(0, 24)}` + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -1968,6 +2062,7 @@ async function* geminiSseToAnthropic( }, } hasEmittedCurrentTool = true + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -1982,15 +2077,23 @@ async function* geminiSseToAnthropic( } if (hasEmittedTextStart || hasEmittedCurrentTool) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } } + throwIfStreamAborted(signal) yield { type: 'message_delta', delta: { stop_reason: mapFinishReason(finishReason, hasEmittedCurrentTool) }, usage: usage ?? {}, } + throwIfStreamAborted(signal) yield { type: 'message_stop' } + streamComplete = true } finally { + if (!streamComplete || signal?.aborted) { + readerCanceller.cancel(createStreamAbortError()) + } + readerCanceller.cleanup() reader.releaseLock() } } @@ -2037,34 +2140,16 @@ async function* openaiStreamToAnthropic( let xmlToolCallText: string | null = null let xmlHoldback = '' - // Emit message_start - yield { - type: 'message_start', - message: { - id: messageId, - type: 'message', - role: 'assistant', - content: [], - model, - stop_reason: null, - stop_sequence: null, - usage: { - input_tokens: 0, - output_tokens: 0, - cache_creation_input_tokens: 0, - cache_read_input_tokens: 0, - }, - }, - } - const readerOrNull = response.body?.getReader() if (!readerOrNull) throw new Error('Response body is not readable') const reader: ReadableStreamDefaultReader = readerOrNull + const readerCanceller = createReaderCanceller(reader, signal) const decoder = new TextDecoder() let buffer = '' const STREAM_IDLE_TIMEOUT_MS = 120_000 // 2 minutes without data = connection likely dead let lastDataTime = Date.now() + let streamComplete = false /** * Read from the stream with an idle timeout. If no data arrives within @@ -2077,35 +2162,49 @@ async function* openaiStreamToAnthropic( type ReadResult = Awaited> async function readWithTimeout(): Promise { return new Promise((resolve, reject) => { - const timeoutId = setTimeout(() => { + let settled = false + let timeoutId: ReturnType | undefined = setTimeout(() => { const elapsed = Math.round((Date.now() - lastDataTime) / 1000) - reject(new Error( + cancelAndReject(new Error( `OpenAI/Gemini SSE stream idle for ${elapsed}s (limit: ${STREAM_IDLE_TIMEOUT_MS / 1000}s). Connection likely dropped.`, )) }, STREAM_IDLE_TIMEOUT_MS) - // If the caller aborts, clear the timer so the AbortError surfaces - // cleanly instead of being masked by a spurious idle timeout. - let abortCleanup: (() => void) | undefined - if (signal) { - abortCleanup = () => { + const cleanup = () => { + if (timeoutId !== undefined) { clearTimeout(timeoutId) + timeoutId = undefined } - signal.addEventListener('abort', abortCleanup, { once: true }) + signal?.removeEventListener('abort', onAbort) + } + const finishResolve = (value: ReadResult) => { + if (settled) return + settled = true + cleanup() + if (value.value) lastDataTime = Date.now() + resolve(value) + } + const finishReject = (error: unknown) => { + if (settled) return + settled = true + cleanup() + reject(error) + } + const cancelAndReject = (error: unknown) => { + readerCanceller.cancel(error) + finishReject(error) + } + const onAbort = () => cancelAndReject(createStreamAbortError()) + + signal?.addEventListener('abort', onAbort, { once: true }) + if (signal?.aborted) { + onAbort() + return } reader.read().then( - result => { - clearTimeout(timeoutId) - if (signal && abortCleanup) signal.removeEventListener('abort', abortCleanup) - if (result.value) lastDataTime = Date.now() - resolve(result) - }, - err => { - clearTimeout(timeoutId) - if (signal && abortCleanup) signal.removeEventListener('abort', abortCleanup) - reject(err) - }, + result => finishResolve(result), + err => finishReject(err), ) }) } @@ -2115,6 +2214,7 @@ async function* openaiStreamToAnthropic( const tail = thinkFilter.flush() if (tail) { + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2122,6 +2222,7 @@ async function* openaiStreamToAnthropic( } } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex, @@ -2133,6 +2234,7 @@ async function* openaiStreamToAnthropic( const emitTextDelta = async function* (text: string) { if (!text) return if (!hasEmittedContentStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -2143,6 +2245,7 @@ async function* openaiStreamToAnthropic( const visible = thinkFilter.feed(text) if (visible) { + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2156,6 +2259,7 @@ async function* openaiStreamToAnthropic( toolCalls: ParsedRawToolCall[], ) { if (hasEmittedThinkingStart && !hasClosedThinking) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } contentBlockIndex++ hasClosedThinking = true @@ -2165,6 +2269,7 @@ async function* openaiStreamToAnthropic( } for (const toolCall of toolCalls) { + throwIfStreamAborted(signal) const toolBlockIndex = contentBlockIndex yield { type: 'content_block_start', @@ -2177,6 +2282,7 @@ async function* openaiStreamToAnthropic( }, } contentBlockIndex++ + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: toolBlockIndex, @@ -2185,21 +2291,49 @@ async function* openaiStreamToAnthropic( partial_json: toolCall.argumentsJson, }, } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: toolBlockIndex } processStreamChunk(streamState, toolCall.argumentsJson) } } try { + throwIfStreamAborted(signal) + + // Emit message_start + yield { + type: 'message_start', + message: { + id: messageId, + type: 'message', + role: 'assistant', + content: [], + model, + stop_reason: null, + stop_sequence: null, + usage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }, + } + while (true) { const { done, value } = await readWithTimeout() - if (done) break + if (done) { + streamComplete = true + break + } + throwIfStreamAborted(signal) buffer += decoder.decode(value, { stream: true }) const lines = buffer.split('\n') buffer = lines.pop() ?? '' for (const line of lines) { + throwIfStreamAborted(signal) const trimmed = line.trim() if (!trimmed || trimmed === 'data: [DONE]') continue if (!trimmed.startsWith('data: ')) continue @@ -2240,6 +2374,7 @@ async function* openaiStreamToAnthropic( const chunkUsage = convertChunkUsage(chunk.usage) for (const choice of chunk.choices ?? []) { + throwIfStreamAborted(signal) const delta = choice.delta // Reasoning models (e.g. GLM-5, DeepSeek) may stream chain-of-thought @@ -2247,6 +2382,7 @@ async function* openaiStreamToAnthropic( // Emit reasoning as a thinking block and content as a text block. if (delta.reasoning_content != null && delta.reasoning_content !== '') { if (!hasEmittedThinkingStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -2254,6 +2390,7 @@ async function* openaiStreamToAnthropic( } hasEmittedThinkingStart = true } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2266,6 +2403,7 @@ async function* openaiStreamToAnthropic( if (delta.content != null && delta.content !== '') { // Close thinking block if transitioning from reasoning to content if (hasEmittedThinkingStart && !hasClosedThinking) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } contentBlockIndex++ hasClosedThinking = true @@ -2345,6 +2483,7 @@ async function* openaiStreamToAnthropic( if (tc.id && tc.function?.name) { // New tool call starting — close any open thinking block first if (hasEmittedThinkingStart && !hasClosedThinking) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } contentBlockIndex++ hasClosedThinking = true @@ -2355,6 +2494,7 @@ async function* openaiStreamToAnthropic( // instead of emitting during the streaming phase). if (isOllamaStream && ollamaTextBuffer) { if (!hasEmittedContentStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -2362,6 +2502,7 @@ async function* openaiStreamToAnthropic( } hasEmittedContentStart = true } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2393,6 +2534,7 @@ async function* openaiStreamToAnthropic( normalizeAtStop, }) + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: toolBlockIndex, @@ -2409,6 +2551,7 @@ async function* openaiStreamToAnthropic( // Emit any initial arguments if (tc.function.arguments && !normalizeAtStop) { + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: toolBlockIndex, @@ -2430,6 +2573,7 @@ async function* openaiStreamToAnthropic( continue } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: active.index, @@ -2450,6 +2594,7 @@ async function* openaiStreamToAnthropic( // Close any open thinking block that wasn't closed by content transition if (hasEmittedThinkingStart && !hasClosedThinking) { + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: contentBlockIndex } contentBlockIndex++ hasClosedThinking = true @@ -2478,6 +2623,7 @@ async function* openaiStreamToAnthropic( if (hasEmittedContentStart) { // Text block was already open — emit stripped prose then close it. if (strippedVisible) { + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2488,12 +2634,14 @@ async function* openaiStreamToAnthropic( } else if (strippedVisible) { // Text was buffered (Ollama path, hasEmittedContentStart === false). // Open a text block, emit the visible prose before the tool call, close it. + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, content_block: { type: 'text', text: '' }, } hasEmittedContentStart = true + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2502,6 +2650,7 @@ async function* openaiStreamToAnthropic( yield* closeActiveContentBlock() } for (const tc of textToolCalls) { + throwIfStreamAborted(signal) const toolBlockIndex = contentBlockIndex yield { type: 'content_block_start', @@ -2509,11 +2658,13 @@ async function* openaiStreamToAnthropic( content_block: { type: 'tool_use', id: tc.id, name: tc.name, input: {} }, } contentBlockIndex++ + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: toolBlockIndex, delta: { type: 'input_json_delta', partial_json: JSON.stringify(tc.arguments) }, } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: toolBlockIndex } } // Only remap finish_reason to 'tool_calls' for the normal stop case; @@ -2526,6 +2677,7 @@ async function* openaiStreamToAnthropic( // Open a text block first if one is not already open (guards the edge case // where hasEmittedContentStart is false but the buffer has content). if (!hasEmittedContentStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -2533,6 +2685,7 @@ async function* openaiStreamToAnthropic( } hasEmittedContentStart = true } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2563,6 +2716,7 @@ async function* openaiStreamToAnthropic( xmlClosedContentBlock = true } for (const tc of calls) { + throwIfStreamAborted(signal) const toolBlockIndex = contentBlockIndex yield { type: 'content_block_start', @@ -2570,11 +2724,13 @@ async function* openaiStreamToAnthropic( content_block: { type: 'tool_use', id: tc.id, name: tc.name, input: {} }, } contentBlockIndex++ + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: toolBlockIndex, delta: { type: 'input_json_delta', partial_json: JSON.stringify(tc.arguments) }, } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: toolBlockIndex } } if (originalFinishReason === 'stop') { @@ -2630,6 +2786,7 @@ async function* openaiStreamToAnthropic( } } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: tc.index, @@ -2638,6 +2795,7 @@ async function* openaiStreamToAnthropic( partial_json: partialJson, }, } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: tc.index } continue } @@ -2659,6 +2817,7 @@ async function* openaiStreamToAnthropic( } if (suffixToAdd) { + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: tc.index, @@ -2669,6 +2828,7 @@ async function* openaiStreamToAnthropic( } } + throwIfStreamAborted(signal) yield { type: 'content_block_stop', index: tc.index } } @@ -2682,6 +2842,7 @@ async function* openaiStreamToAnthropic( // Gemini/Azure content safety filter blocked the response. // Emit a visible text block so the user knows why output was truncated. if (!hasEmittedContentStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -2689,6 +2850,7 @@ async function* openaiStreamToAnthropic( } hasEmittedContentStart = true } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2700,6 +2862,7 @@ async function* openaiStreamToAnthropic( // detecting a stalled stream. Either way, the user should know // the answer they're seeing isn't complete. if (!hasEmittedContentStart) { + throwIfStreamAborted(signal) yield { type: 'content_block_start', index: contentBlockIndex, @@ -2707,6 +2870,7 @@ async function* openaiStreamToAnthropic( } hasEmittedContentStart = true } + throwIfStreamAborted(signal) yield { type: 'content_block_delta', index: contentBlockIndex, @@ -2715,6 +2879,7 @@ async function* openaiStreamToAnthropic( } lastStopReason = stopReason + throwIfStreamAborted(signal) yield { type: 'message_delta', delta: { stop_reason: stopReason, stop_sequence: null }, @@ -2732,6 +2897,7 @@ async function* openaiStreamToAnthropic( (chunk.choices?.length ?? 0) === 0 && lastStopReason !== null ) { + throwIfStreamAborted(signal) yield { type: 'message_delta', delta: { stop_reason: lastStopReason, stop_sequence: null }, @@ -2742,6 +2908,10 @@ async function* openaiStreamToAnthropic( } } } finally { + if (!streamComplete || signal?.aborted) { + readerCanceller.cancel(createStreamAbortError()) + } + readerCanceller.cleanup() reader.releaseLock() } @@ -2759,6 +2929,7 @@ async function* openaiStreamToAnthropic( ) } + throwIfStreamAborted(signal) yield { type: 'message_stop' } } @@ -2767,16 +2938,84 @@ async function* openaiStreamToAnthropic( // --------------------------------------------------------------------------- class OpenAIShimStream { - private generator: AsyncGenerator + private makeGenerator: (signal: AbortSignal) => AsyncGenerator + private parentSignal?: AbortSignal + private generator?: AsyncGenerator + private cleanupCombinedSignal?: () => void + private cleanupPreIterationAbort?: () => void // The controller property is checked by claude.ts to distinguish streams from error messages controller = new AbortController() - constructor(generator: AsyncGenerator) { - this.generator = generator + constructor( + makeGenerator: (signal: AbortSignal) => AsyncGenerator, + parentSignal?: AbortSignal, + cancelBeforeIteration?: () => void, + ) { + this.makeGenerator = makeGenerator + this.parentSignal = parentSignal + + if (cancelBeforeIteration) { + let cleaned = false + let cancelled = false + let onAbort: () => void = () => {} + const cleanup = () => { + if (cleaned) return + cleaned = true + this.controller.signal.removeEventListener('abort', onAbort) + parentSignal?.removeEventListener('abort', onAbort) + } + onAbort = () => { + if (!this.generator && !cancelled) { + cancelled = true + cancelBeforeIteration() + } + cleanup() + } + + this.controller.signal.addEventListener('abort', onAbort, { once: true }) + parentSignal?.addEventListener('abort', onAbort, { once: true }) + this.cleanupPreIterationAbort = cleanup + + if (this.controller.signal.aborted || parentSignal?.aborted) { + onAbort() + } + } + } + + private getGenerator(): AsyncGenerator { + if (this.generator) { + return this.generator + } + + this.cleanupPreIterationAbort?.() + this.cleanupPreIterationAbort = undefined + + const combined = createCombinedAbortSignal(this.parentSignal, { + signalB: this.controller.signal, + }) + this.cleanupCombinedSignal = combined.cleanup + this.generator = this.makeGenerator(combined.signal) + return this.generator } async *[Symbol.asyncIterator]() { - yield* this.generator + const generator = this.getGenerator() + let completed = false + try { + yield* generator + completed = true + } finally { + if (!completed && !this.controller.signal.aborted) { + this.controller.abort() + } + this.cleanupCombinedSignal?.() + this.cleanupCombinedSignal = undefined + this.cleanupPreIterationAbort?.() + this.cleanupPreIterationAbort = undefined + if (!completed) { + void generator.return?.(undefined).catch(() => {}) + } + } } } @@ -2826,18 +3065,24 @@ class OpenAIShimMessages { const isResponsesStream = response.url?.includes('/responses') const isMessagesStream = response.url?.includes('/messages') const isGeminiStream = response.url?.includes('/models/gemini-') + const cancelBeforeIteration = () => { + void response.body?.cancel(createStreamAbortError()).catch(() => {}) + } return new OpenAIShimStream( - ( - request.transport === 'codex_responses' || - request.transport === 'responses' || - isResponsesStream - ) - ? codexStreamToAnthropic(response, request.resolvedModel, options?.signal) - : isMessagesStream - ? anthropicSsePassthrough(response, request.resolvedModel, options?.signal) - : isGeminiStream - ? geminiSseToAnthropic(response, request.resolvedModel, options?.signal) - : openaiStreamToAnthropic(response, request.resolvedModel, options?.signal, isLikelyOllamaEndpoint(request.baseUrl)), + streamSignal => + ( + request.transport === 'codex_responses' || + request.transport === 'responses' || + isResponsesStream + ) + ? codexStreamToAnthropic(response, request.resolvedModel, streamSignal) + : isMessagesStream + ? anthropicSsePassthrough(response, request.resolvedModel, streamSignal) + : isGeminiStream + ? geminiSseToAnthropic(response, request.resolvedModel, streamSignal) + : openaiStreamToAnthropic(response, request.resolvedModel, streamSignal, isLikelyOllamaEndpoint(request.baseUrl)), + options?.signal, + cancelBeforeIteration, ) }