starred/openclaude
1
0
Fork 0
mirror of https://github.com/Gitlawb/openclaude.git synced 2026-08-24 02:34:15 -05:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai-shim): wire stream controller abort (#1828)

Browse Source 
* fix(openai-shim): wire stream controller abort

* test(openai-shim): guard Ollama abort fixture cleanup
This commit is contained in:
Bogdan
2026-07-01 07:56:50 +08:00
committed by GitHub
parent f6ecee0c00
commit bb61d8430b
4 changed files with 1687 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/api/codexShim.test.ts
+163
View File
@@ -70,6 +70,64 @@ async function collectStreamEventTypes(responseText: string): Promise<string[]>
return events
}
async function waitForPromise<T>(
promise: Promise<T>,
timeoutMs: number,
message: string,
): Promise<T> {
let timer: ReturnType<typeof setTimeout> | undefined
try {
return await Promise.race([
promise,
new Promise<T>((_, 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<Uint8Array> | undefined
let closed = false
const response = new Response(
new ReadableStream<Uint8Array>({
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 <think> tag block from Codex SSE text stream', async () => {
const responseText = [
'event: response.output_item.added',
src/services/api/codexShim.ts
+327 -225
View File
@@ -81,6 +81,42 @@ type CodexSseEvent = {
data: Record<string, any>
}
function createStreamAbortError(): DOMException {
return new DOMException('Aborted', 'AbortError')
}
function throwIfStreamAborted(signal?: AbortSignal): void {
if (signal?.aborted) {
throw createStreamAbortError()
}
}
function createReaderCanceller(
reader: ReadableStreamDefaultReader<Uint8Array>,
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<string, unknown>): 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<CodexSseEvent> {
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<Bun.ReadableStreamDefaultReadResult<Uint8Array<ArrayBuffer>>> {
return new Promise((resolve, reject) => {
const timeoutId = setTimeout(() => {
let settled = false
let timeoutId: ReturnType<typeof setTimeout> | 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<Uint8Array<ArrayBuffer>>,
) => {
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<string, any>
try {
const parsed = JSON.parse(rawData)
if (!parsed || typeof parsed !== 'object') continue
data = parsed as Record<string, any>
} 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<string, any>
try {
const parsed = JSON.parse(rawData)
if (!parsed || typeof parsed !== 'object') continue
data = parsed as Record<string, any>
} 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<string, any> | 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<string, unknown> | 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<string, unknown> | undefined,
),
}
yield { type: 'message_stop' }
}
export function convertCodexResponseToAnthropicMessage(
src/services/api/openaiShim.test.ts
+862
View File
@@ -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<Uint8Array> | undefined
let closed = false
const response = new Response(
new ReadableStream<Uint8Array>({
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<Record<string, unknown>> & {
controller: AbortController
}
type StreamDrainOutcome =
| { status: 'completed'; events: Array<Record<string, unknown>> }
| {
status: 'rejected'
events: Array<Record<string, unknown>>
error: unknown
}
async function waitForPromise<T>(
promise: Promise<T>,
timeoutMs: number,
message: string,
): Promise<T> {
let timer: ReturnType<typeof setTimeout> | undefined
try {
return await Promise.race([
promise,
new Promise<T>((_, 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<StreamDrainOutcome> {
const events: Array<Record<string, unknown>> = []
let resolveReady!: () => void
const ready = new Promise<void>(resolve => {
resolveReady = resolve
})
const drain = (async (): Promise<StreamDrainOutcome> => {
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<IteratorResult<Record<string, unknown>>> {
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<Record<string, unknown>>
}
async function expectBufferedAbortRejectsNext({
expectedText,
label,
stream,
}: {
expectedText?: string
label: string
stream: ShimStream
}): Promise<void> {
const iterator = stream[Symbol.asyncIterator]()
try {
let firstDelta: Record<string, unknown> | 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<string, unknown>,
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'
src/services/api/openaiShim.ts
+335 -90
View File
@@ -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<ReadableStreamDefaultReader<Uint8Array>['read']>
>
function createReaderCanceller(
reader: ReadableStreamDefaultReader<Uint8Array>,
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<Uint8Array>,
signal?: AbortSignal,
cancelReader?: (error?: unknown) => void,
): Promise<StreamReadResult> {
if (!signal) return reader.read()
return new Promise<StreamReadResult>((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<Uint8Array> = 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<ReturnType<typeof reader.read>>
function readWithAbort(): Promise<ReadResult> {
if (!signal) return reader.read()
return new Promise<ReadResult>((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<AnthropicStreamEvent> {
const reader: ReadableStreamDefaultReader<Uint8Array> | 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<AnthropicUsage> | undefined
let finishReason: string | undefined
function readWithAbort(): Promise<ReadableStreamReadResult<Uint8Array>> {
if (!signal) return reader!.read() as Promise<ReadableStreamReadResult<Uint8Array>>
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<Uint8Array>) },
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<Uint8Array> = 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<ReturnType<typeof reader.read>>
async function readWithTimeout(): Promise<ReadResult> {
return new Promise<ReadResult>((resolve, reject) => {
const timeoutId = setTimeout(() => {
let settled = false
let timeoutId: ReturnType<typeof setTimeout> | 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<AnthropicStreamEvent>
private makeGenerator: (signal: AbortSignal) => AsyncGenerator<AnthropicStreamEvent>
private parentSignal?: AbortSignal
private generator?: AsyncGenerator<AnthropicStreamEvent>
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<AnthropicStreamEvent>) {
this.generator = generator
constructor(
makeGenerator: (signal: AbortSignal) => AsyncGenerator<AnthropicStreamEvent>,
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<AnthropicStreamEvent> {
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,
)
}