From d0f99e88f8ac8d666a3a20f9de43e54f8b0fc07c Mon Sep 17 00:00:00 2001 From: jianYanZhiX7 Date: Thu, 17 Sep 2026 18:29:54 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20OpenAI=20=E6=B5=81?= =?UTF-8?q?=E9=9D=99=E9=BB=98=E6=88=AA=E6=96=AD=E4=B8=8E=E5=81=9C=E6=BB=9E?= =?UTF-8?q?=E6=8C=82=E8=B5=B7=EF=BC=8C=E8=A1=A5=E5=BC=BA=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E8=AF=AD=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 静默截断：迭代结束无 finish_reason 时抛 OpenAIStreamIncompleteError， 不再把半截消息当完整回复（stop_reason 为 null） - 停滞挂起：新增流空闲看门狗（OPENAI_STREAM_IDLE_TIMEOUT_MS，默认 90s）， 每次尝试持有独立 AbortController，空闲超时单独计数（上限 2 次） - 重试框架：streamRetry 前缀比对续传；续传分歧时丢弃前缀重开请求 （MAX_RESUME_RESTARTS=1）；零事件截断豁免 hasProgress 守卫单独重试 - 用户中断（Ctrl+C）静默退出，与 Anthropic 路径对齐 - ChatGPT Responses 路径补齐同类截断检测 - OpenAI 请求 maxRetries 可通过 OPENAI_MAX_RETRIES 配置（默认 10） --- packages/@ant/model-provider/src/index.ts | 4 + .../__tests__/openaiStreamAdapter.test.ts | 343 ++++++++ .../src/shared/openaiStreamAdapter.ts | 74 +- .../src/shared/openaiStreamTermination.ts | 11 + .../api/openai/__tests__/client.test.ts | 57 ++ .../__tests__/queryModelOpenAI.isolated.ts | 1 + .../__tests__/streamIdleTimeout.test.ts | 131 +++ .../api/openai/__tests__/streamRetry.test.ts | 772 ++++++++++++++++++ src/services/api/openai/client.ts | 13 +- src/services/api/openai/index.ts | 128 +-- src/services/api/openai/responsesAdapter.ts | 17 +- src/services/api/openai/streamIdleTimeout.ts | 89 ++ src/services/api/openai/streamRetry.ts | 407 +++++++++ .../__tests__/sideQuery.chatgptAuth.test.ts | 96 ++- 14 files changed, 2033 insertions(+), 110 deletions(-) create mode 100644 packages/@ant/model-provider/src/shared/openaiStreamTermination.ts create mode 100644 src/services/api/openai/__tests__/client.test.ts create mode 100644 src/services/api/openai/__tests__/streamIdleTimeout.test.ts create mode 100644 src/services/api/openai/__tests__/streamRetry.test.ts create mode 100644 src/services/api/openai/streamIdleTimeout.ts create mode 100644 src/services/api/openai/streamRetry.ts diff --git a/packages/@ant/model-provider/src/index.ts b/packages/@ant/model-provider/src/index.ts index 823edb3812..262c089e16 100644 --- a/packages/@ant/model-provider/src/index.ts +++ b/packages/@ant/model-provider/src/index.ts @@ -67,6 +67,10 @@ export { anthropicToolChoiceToOpenAI, } from './shared/openaiConvertTools.js' export { adaptOpenAIStreamToAnthropic } from './shared/openaiStreamAdapter.js' +export { + allowsIncompleteOpenAIStream, + OpenAIStreamIncompleteError, +} from './shared/openaiStreamTermination.js' export { normalizeOpenAIUsage, type AnthropicUsage, diff --git a/packages/@ant/model-provider/src/shared/__tests__/openaiStreamAdapter.test.ts b/packages/@ant/model-provider/src/shared/__tests__/openaiStreamAdapter.test.ts index 9aba0fb6c2..af03abe48b 100644 --- a/packages/@ant/model-provider/src/shared/__tests__/openaiStreamAdapter.test.ts +++ b/packages/@ant/model-provider/src/shared/__tests__/openaiStreamAdapter.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from 'bun:test' import type { ChatCompletionChunk } from 'openai/resources/chat/completions/completions.mjs' import { adaptOpenAIStreamToAnthropic } from '../openaiStreamAdapter.js' +import { OpenAIStreamIncompleteError } from '../openaiStreamTermination.js' /** Helper to create a mock async iterable from chunk array */ function mockStream( @@ -912,3 +913,345 @@ describe('prompt caching support', () => { expect(msgDelta.usage.output_tokens).toBe(30) }) }) + +describe('qw reasoning model edge cases', () => { + test('finish_reason in chunk with empty reasoning_content after text has started', async () => { + // Bug fix: qwen3.7-max sends chunks with reasoning_content: "" together + // with finish_reason. The old `if (textBlockOpen) continue` skipped the + // entire chunk, causing finish_reason to be lost and message_stop never emitted. + const events = await collectEvents([ + // reasoning only + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'Let me think...' }, + finish_reason: null, + }, + ], + }), + // transition: content starts (thinking closes, text opens) + makeChunk({ + choices: [ + { index: 0, delta: { content: 'The answer' }, finish_reason: null }, + ], + }), + // MORE text (typical multi-chunk response) + makeChunk({ + choices: [ + { index: 0, delta: { content: ' is 42.' }, finish_reason: null }, + ], + }), + // BUG SCENARIO: empty reasoning + finish_reason in same chunk after text started + makeChunk({ + choices: [ + { index: 0, delta: { reasoning_content: '' }, finish_reason: 'stop' }, + ], + }), + ]) + + // message_delta + message_stop must be emitted + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta).toBeDefined() + expect(msgDelta.delta.stop_reason).toBe('end_turn') + + const msgStop = events.find(e => e.type === 'message_stop') as any + expect(msgStop).toBeDefined() + + // Text should be one continuous block (not fragmented) + const textDeltas = events.filter( + e => e.type === 'content_block_delta' && e.delta.type === 'text_delta', + ) as any[] + expect(textDeltas.length).toBe(2) + expect(textDeltas[0].delta.text).toBe('The answer') + expect(textDeltas[1].delta.text).toBe(' is 42.') + }) + + test('suppresses mid-text reasoning_content but still processes finish_reason', async () => { + // qwen3.7-max interleaves reasoning_content within content stream. + // Mid-text reasoning must be suppressed (not open new thinking blocks) + // but finish_reason must still be captured. + const events = await collectEvents([ + // initial reasoning + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'Initial thought.' }, + finish_reason: null, + }, + ], + }), + // text starts + makeChunk({ + choices: [ + { + index: 0, + delta: { content: 'Here is part 1.' }, + finish_reason: null, + }, + ], + }), + // INTERLEAVED: reasoning arrives after text has started — must be suppressed + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'Mid-text thought.' }, + finish_reason: null, + }, + ], + }), + // more text + makeChunk({ + choices: [ + { + index: 0, + delta: { content: ' Here is part 2.' }, + finish_reason: null, + }, + ], + }), + // finish (may also carry empty reasoning) + makeChunk({ + choices: [ + { index: 0, delta: { reasoning_content: '' }, finish_reason: 'stop' }, + ], + }), + ]) + + // Only ONE thinking block should be opened (the initial one) + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] + const thinkingStarts = blockStarts.filter( + b => b.content_block.type === 'thinking', + ) + const textStarts = blockStarts.filter(b => b.content_block.type === 'text') + expect(thinkingStarts.length).toBe(1) // only the initial thinking block + expect(textStarts.length).toBe(1) // only one text block (not fragmented) + + // Text deltas should NOT include the mid-text reasoning + const textDeltas = events.filter( + e => e.type === 'content_block_delta' && e.delta.type === 'text_delta', + ) as any[] + const fullText = textDeltas.map((d: any) => d.delta.text).join('') + expect(fullText).toBe('Here is part 1. Here is part 2.') + expect(fullText).not.toContain('Mid-text thought') + + // Must still emit message_delta + message_stop + expect(events.find(e => e.type === 'message_delta')).toBeDefined() + expect(events.find(e => e.type === 'message_stop')).toBeDefined() + }) + + test('handles reasoning+content+finish all in one chunk', async () => { + // Some APIs might pack all three into a single chunk. + const events = await collectEvents([ + makeChunk({ + choices: [ + { + index: 0, + delta: { + reasoning_content: 'Let me answer.', + content: 'Hello world', + }, + finish_reason: 'stop', + }, + ], + }), + ]) + + // Thinking block opened, then closed; text block opened, then closed + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] + expect(blockStarts.length).toBe(2) + expect(blockStarts[0].content_block.type).toBe('thinking') + expect(blockStarts[1].content_block.type).toBe('text') + + // Both blocks stopped + const blockStops = events.filter( + e => e.type === 'content_block_stop', + ) as any[] + // finish_reason handler closes thinking (if open), text (if open), and safety cleanup + // thinking was closed when text started, so at finish time only text is open + // But wait — finish_reason closes thinking + text if they're open. + // In this case: thinking was closed when text started, text is open at finish. + // Then safety cleanup runs (nothing left open). + // So we should see: thinking_stop(from text handler) + text_stop(from finish handler) + // + safety cleanup(none). Total: 2 content_block_stop events. + expect(blockStops.length).toBeGreaterThanOrEqual(2) + + // message_delta + message_stop must be present + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.delta.stop_reason).toBe('end_turn') + expect(events.find(e => e.type === 'message_stop')).toBeDefined() + }) + + test('text block stays continuous through mid-stream reasoning', async () => { + // Verify the text block index doesn't change when mid-text reasoning is suppressed. + const events = await collectEvents([ + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'thinking...' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [ + { index: 0, delta: { content: 'Part A' }, finish_reason: null }, + ], + }), + // Mid-text reasoning (suppressed) + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'more thinking...' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [ + { index: 0, delta: { content: ' Part B' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + ]) + + // All text deltas should share the same index (text block index = 1) + const textDeltas = events.filter( + e => e.type === 'content_block_delta' && e.delta.type === 'text_delta', + ) as any[] + expect(textDeltas.length).toBe(2) + expect(textDeltas[0].index).toBe(1) // text block index + expect(textDeltas[1].index).toBe(1) // same index, not a new block + }) + + test('handles multiple text→reasoning→text oscillations', async () => { + // Stress test: multiple rounds of text/reasoning interleaving. + const events = await collectEvents([ + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'Think 1.' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [ + { index: 0, delta: { content: 'Text 1.' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'Think 2.' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [ + { index: 0, delta: { content: 'Text 2.' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'Think 3.' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [ + { index: 0, delta: { content: 'Text 3.' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [ + { index: 0, delta: { reasoning_content: '' }, finish_reason: 'stop' }, + ], + }), + ]) + + // Only 1 thinking block (the initial one) + 1 text block (not fragmented) + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] + const thinkingStarts = blockStarts.filter( + b => b.content_block.type === 'thinking', + ) + const textStarts = blockStarts.filter(b => b.content_block.type === 'text') + expect(thinkingStarts.length).toBe(1) + expect(textStarts.length).toBe(1) + + // Text accumulates correctly without mid-text reasoning contamination + const textDeltas = events.filter( + e => e.type === 'content_block_delta' && e.delta.type === 'text_delta', + ) as any[] + const fullText = textDeltas.map((d: any) => d.delta.text).join('') + expect(fullText).toBe('Text 1.Text 2.Text 3.') + + expect(events.find(e => e.type === 'message_delta')).toBeDefined() + expect(events.find(e => e.type === 'message_stop')).toBeDefined() + }) +}) + +describe('stream termination', () => { + test('throws when the stream ends without a finish_reason', async () => { + const chunks = [ + makeChunk({ + choices: [ + { + index: 0, + delta: { role: 'assistant', content: 'partial' }, + finish_reason: null, + }, + ], + }), + ] + + await expect(async () => { + await collectEvents(chunks) + }).toThrow(OpenAIStreamIncompleteError) + }) + + test('throws when the stream ends without any chunk', async () => { + await expect(async () => { + await collectEvents([]) + }).toThrow(OpenAIStreamIncompleteError) + }) + + test('tolerates a missing finish_reason when OPENAI_ALLOW_INCOMPLETE_STREAM is set', async () => { + process.env.OPENAI_ALLOW_INCOMPLETE_STREAM = '1' + try { + const events = await collectEvents([ + makeChunk({ + choices: [ + { + index: 0, + delta: { role: 'assistant', content: 'partial' }, + finish_reason: null, + }, + ], + }), + ]) + + expect(events.find(e => e.type === 'message_stop')).toBeUndefined() + } finally { + delete process.env.OPENAI_ALLOW_INCOMPLETE_STREAM + } + }) +}) diff --git a/packages/@ant/model-provider/src/shared/openaiStreamAdapter.ts b/packages/@ant/model-provider/src/shared/openaiStreamAdapter.ts index 1c9f608c03..926e1c65e0 100644 --- a/packages/@ant/model-provider/src/shared/openaiStreamAdapter.ts +++ b/packages/@ant/model-provider/src/shared/openaiStreamAdapter.ts @@ -2,6 +2,10 @@ import type { BetaRawMessageStreamEvent } from '@anthropic-ai/sdk/resources/beta import type { ChatCompletionChunk } from 'openai/resources/chat/completions/completions.mjs' import { randomUUID } from 'crypto' import { normalizeOpenAIUsage } from './openaiUsage.js' +import { + allowsIncompleteOpenAIStream, + OpenAIStreamIncompleteError, +} from './openaiStreamTermination.js' /** * Adapt an OpenAI streaming response into Anthropic BetaRawMessageStreamEvent. @@ -13,6 +17,14 @@ import { normalizeOpenAIUsage } from './openaiUsage.js' * delta.tool_calls → content_block_start(tool_use) + input_json_delta + content_block_stop * finish_reason → message_delta(stop_reason) + message_stop * + * Termination: + * A healthy stream always carries a finish_reason. If the iterator ends + * without one the upstream connection was cut mid-response (the OpenAI SDK + * treats a missing `[DONE]` as a normal end of stream, so this is the only + * place the truncation is observable). Throwing a retryable error there + * prevents a half-written answer from being reported as complete. + * Set OPENAI_ALLOW_INCOMPLETE_STREAM=1 to restore the lenient behavior. + * * Usage field mapping (OpenAI → Anthropic): * prompt_tokens - cached_tokens - cache_write_tokens → input_tokens * completion_tokens → output_tokens @@ -137,31 +149,39 @@ export async function* adaptOpenAIStreamToAnthropic( // requests, otherwise DeepSeek rejects with 400. const reasoningContent = (delta as any).reasoning_content if (reasoningContent != null) { - if (!thinkingBlockOpen) { - currentContentIndex++ - thinkingBlockOpen = true - openBlockIndices.add(currentContentIndex) + // If text output has already started, suppress intermediate + // reasoning blocks — models like qwen3.7-max interleave + // reasoning_content within content, and opening new thinking + // blocks mid-text would fragment the displayed output. + // NOTE: Do NOT `continue` here — the chunk may also carry + // finish_reason or other fields that must be processed. + if (!textBlockOpen) { + if (!thinkingBlockOpen) { + currentContentIndex++ + thinkingBlockOpen = true + openBlockIndices.add(currentContentIndex) - yield { - type: 'content_block_start', - index: currentContentIndex, - content_block: { - type: 'thinking', - thinking: '', - signature: '', - }, - } as BetaRawMessageStreamEvent - } + yield { + type: 'content_block_start', + index: currentContentIndex, + content_block: { + type: 'thinking', + thinking: '', + signature: '', + }, + } as BetaRawMessageStreamEvent + } - if (reasoningContent !== '') { - yield { - type: 'content_block_delta', - index: currentContentIndex, - delta: { - type: 'thinking_delta', - thinking: reasoningContent, - }, - } as BetaRawMessageStreamEvent + if (reasoningContent !== '') { + yield { + type: 'content_block_delta', + index: currentContentIndex, + delta: { + type: 'thinking_delta', + thinking: reasoningContent, + }, + } as BetaRawMessageStreamEvent + } } } @@ -305,6 +325,14 @@ export async function* adaptOpenAIStreamToAnthropic( } } + if (pendingFinishReason === null && !allowsIncompleteOpenAIStream()) { + throw new OpenAIStreamIncompleteError( + started + ? 'OpenAI stream ended before the response was finished' + : 'OpenAI stream ended without returning any data', + ) + } + // Safety: close any remaining open blocks for (const idx of openBlockIndices) { yield { diff --git a/packages/@ant/model-provider/src/shared/openaiStreamTermination.ts b/packages/@ant/model-provider/src/shared/openaiStreamTermination.ts new file mode 100644 index 0000000000..64f90c18b1 --- /dev/null +++ b/packages/@ant/model-provider/src/shared/openaiStreamTermination.ts @@ -0,0 +1,11 @@ +export class OpenAIStreamIncompleteError extends Error { + constructor(message: string) { + super(message) + this.name = 'OpenAIStreamIncompleteError' + } +} + +export function allowsIncompleteOpenAIStream(): boolean { + const value = process.env.OPENAI_ALLOW_INCOMPLETE_STREAM + return value === '1' || value?.toLowerCase() === 'true' +} diff --git a/src/services/api/openai/__tests__/client.test.ts b/src/services/api/openai/__tests__/client.test.ts new file mode 100644 index 0000000000..94dd8aa440 --- /dev/null +++ b/src/services/api/openai/__tests__/client.test.ts @@ -0,0 +1,57 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' +import { logMock } from '../../../../../tests/mocks/log' +import { debugMock } from '../../../../../tests/mocks/debug' + +mock.module('src/utils/log.ts', logMock) +mock.module('src/utils/debug.ts', debugMock) + +import { clearOpenAIClientCache, getOpenAIClient } from '../client.js' + +const ENV_KEY = 'OPENAI_MAX_RETRIES' +let savedEnv: string | undefined + +beforeEach(() => { + savedEnv = process.env[ENV_KEY] + clearOpenAIClientCache() +}) + +afterEach(() => { + if (savedEnv === undefined) { + delete process.env[ENV_KEY] + } else { + process.env[ENV_KEY] = savedEnv + } + clearOpenAIClientCache() +}) + +describe('getOpenAIClient maxRetries', () => { + test('defaults to 10 when OPENAI_MAX_RETRIES is unset', () => { + delete process.env[ENV_KEY] + expect(getOpenAIClient().maxRetries).toBe(10) + }) + + test('uses OPENAI_MAX_RETRIES when set', () => { + process.env[ENV_KEY] = '5' + expect(getOpenAIClient().maxRetries).toBe(5) + }) + + test('allows OPENAI_MAX_RETRIES=0 to disable retries', () => { + process.env[ENV_KEY] = '0' + expect(getOpenAIClient().maxRetries).toBe(0) + }) + + test('falls back to default for unparseable values', () => { + process.env[ENV_KEY] = 'abc' + expect(getOpenAIClient().maxRetries).toBe(10) + }) + + test('falls back to default for negative values', () => { + process.env[ENV_KEY] = '-1' + expect(getOpenAIClient().maxRetries).toBe(10) + }) + + test('explicit maxRetries option overrides env and default', () => { + delete process.env[ENV_KEY] + expect(getOpenAIClient({ maxRetries: 0 }).maxRetries).toBe(0) + }) +}) diff --git a/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts b/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts index eea7565b14..a2bfedbd6c 100644 --- a/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts +++ b/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts @@ -265,6 +265,7 @@ mock.module('bun:bundle', () => ({ })) mock.module('../client.js', () => ({ + getOpenAIMaxRetries: () => 10, getOpenAIClient: () => ({ chat: { completions: { diff --git a/src/services/api/openai/__tests__/streamIdleTimeout.test.ts b/src/services/api/openai/__tests__/streamIdleTimeout.test.ts new file mode 100644 index 0000000000..81845e942e --- /dev/null +++ b/src/services/api/openai/__tests__/streamIdleTimeout.test.ts @@ -0,0 +1,131 @@ +import { describe, expect, test } from 'bun:test' +import { + getOpenAIStreamIdleTimeoutMs, + OpenAIStreamIdleTimeoutError, + watchStreamIdle, +} from '../streamIdleTimeout.js' + +function delay(ms: number): Promise { + return new Promise(resolve => setTimeout(resolve, ms)) +} + +async function* paced( + items: Array<{ delayMs: number; value: T }>, +): AsyncGenerator { + for (const item of items) { + await delay(item.delayMs) + yield item.value + } +} + +function stalled(): AsyncIterable { + return { + [Symbol.asyncIterator](): AsyncIterator { + return { next: () => new Promise>(() => {}) } + }, + } +} + +async function drain(stream: AsyncIterable): Promise { + const values: T[] = [] + for await (const value of stream) values.push(value) + return values +} + +describe('watchStreamIdle', () => { + test('passes events through while they keep arriving', async () => { + const stream = watchStreamIdle( + paced([ + { delayMs: 40, value: 1 }, + { delayMs: 40, value: 2 }, + { delayMs: 40, value: 3 }, + ]), + { + timeoutMs: 60, + onTimeout: () => { + throw new Error('watchdog fired on a healthy stream') + }, + }, + ) + + expect(await drain(stream)).toEqual([1, 2, 3]) + }) + + test('warns at half the timeout and fails a read that stays quiet', async () => { + const phases: string[] = [] + const stream = watchStreamIdle(stalled(), { + timeoutMs: 60, + onTimeout: () => phases.push('timeout'), + onWarning: () => phases.push('warning'), + }) + + await expect(drain(stream)).rejects.toThrow(OpenAIStreamIdleTimeoutError) + expect(phases).toEqual(['warning', 'timeout']) + }) + + test('reports the timeout in the error message', () => { + expect(new OpenAIStreamIdleTimeoutError(90_000).message).toContain('90s') + }) + + test('yields the source unchanged when the watchdog is disabled', async () => { + const stream = watchStreamIdle(paced([{ delayMs: 30, value: 'a' }]), { + timeoutMs: 0, + onTimeout: () => { + throw new Error('watchdog fired while disabled') + }, + }) + + expect(await drain(stream)).toEqual(['a']) + }) + + test('disarms the timer when the consumer stops early', async () => { + let timeouts = 0 + const stream = watchStreamIdle( + paced([ + { delayMs: 5, value: 'a' }, + { delayMs: 5, value: 'b' }, + ]), + { + timeoutMs: 30, + onTimeout: () => { + timeouts++ + }, + }, + ) + + for await (const value of stream) { + if (value === 'a') break + } + await delay(50) + + expect(timeouts).toBe(0) + }) +}) + +describe('getOpenAIStreamIdleTimeoutMs', () => { + test('defaults to 90s and honors overrides and disables', () => { + const previous = process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS + try { + delete process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS + expect(getOpenAIStreamIdleTimeoutMs()).toBe(90_000) + + process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS = '5000' + expect(getOpenAIStreamIdleTimeoutMs()).toBe(5000) + + process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS = '0' + expect(getOpenAIStreamIdleTimeoutMs()).toBe(0) + + process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS = 'off' + expect(getOpenAIStreamIdleTimeoutMs()).toBe(0) + + process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS = 'nonsense' + expect(getOpenAIStreamIdleTimeoutMs()).toBe(90_000) + } finally { + if (previous === undefined) { + delete process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS + } else { + process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS = previous + } + } + }) +}) diff --git a/src/services/api/openai/__tests__/streamRetry.test.ts b/src/services/api/openai/__tests__/streamRetry.test.ts new file mode 100644 index 0000000000..e58d38fc9f --- /dev/null +++ b/src/services/api/openai/__tests__/streamRetry.test.ts @@ -0,0 +1,772 @@ +import { describe, expect, test } from 'bun:test' +import type { BetaRawMessageStreamEvent } from '@anthropic-ai/sdk/resources/beta/messages/messages.mjs' +import type { ChatCompletionChunk } from 'openai/resources/chat/completions/completions.mjs' +import { + adaptOpenAIStreamToAnthropic, + OpenAIStreamIncompleteError, +} from '@ant/model-provider' +import { + isRetryableOpenAIStreamError, + OpenAIStreamResumeError, + retryOpenAIStream, +} from '../streamRetry.js' +import { OpenAIStreamIdleTimeoutError } from '../streamIdleTimeout.js' + +function messageStart(): BetaRawMessageStreamEvent { + return { + type: 'message_start', + message: { + id: 'msg_test', + type: 'message', + role: 'assistant', + content: [], + model: 'test-model', + stop_reason: null, + stop_sequence: null, + usage: { input_tokens: 1, output_tokens: 0 }, + }, + } as unknown as BetaRawMessageStreamEvent +} + +function blockStart(index = 0): BetaRawMessageStreamEvent { + return { + type: 'content_block_start', + index, + content_block: { type: 'text', text: '' }, + } as BetaRawMessageStreamEvent +} + +function textDelta(text: string, index = 0): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'text_delta', text }, + } as BetaRawMessageStreamEvent +} + +function toolStart(index = 0): BetaRawMessageStreamEvent { + return { + type: 'content_block_start', + index, + content_block: { + type: 'tool_use', + id: 'toolu_test', + name: 'Bash', + input: {}, + }, + } as BetaRawMessageStreamEvent +} + +function inputDelta(json: string, index = 0): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'input_json_delta', partial_json: json }, + } as BetaRawMessageStreamEvent +} + +function blockStop(index = 0): BetaRawMessageStreamEvent { + return { type: 'content_block_stop', index } as BetaRawMessageStreamEvent +} + +function messageStop(): BetaRawMessageStreamEvent { + return { type: 'message_stop' } as BetaRawMessageStreamEvent +} + +function collectText(events: BetaRawMessageStreamEvent[]): string { + return events + .filter( + event => + event.type === 'content_block_delta' && + event.delta.type === 'text_delta', + ) + .map(event => + event.type === 'content_block_delta' && event.delta.type === 'text_delta' + ? event.delta.text + : '', + ) + .join('') +} + +async function* truncatedStream() { + yield messageStart() + yield blockStart() + yield textDelta('Hello ') +} + +async function* interruptedStream() { + yield messageStart() + yield blockStart() + yield textDelta('Hello ') + throw new Error('terminated') +} + +async function* resumedStream(text: string) { + yield messageStart() + yield blockStart() + yield textDelta('Hel') + yield textDelta(text) + yield blockStop() + yield messageStop() +} + +async function* completedStream() { + yield messageStart() + yield blockStart() + yield textDelta('Hello ') + yield textDelta('world') + yield blockStop() + yield messageStop() +} + +async function* emptyStream() {} + +/** + * Mirrors how the SDK behaves when the caller aborts: the pending read + * resolves and the iterator ends cleanly instead of throwing. + */ +async function* stallingStream( + signal: AbortSignal, + emitFirst: boolean, +): AsyncGenerator { + if (emitFirst) { + yield messageStart() + yield blockStart() + } + await new Promise(resolve => { + if (signal.aborted) { + resolve() + return + } + signal.addEventListener('abort', () => resolve(), { once: true }) + }) +} + +describe('retryOpenAIStream', () => { + test('retries an interrupted stream and emits only the new suffix', async () => { + let calls = 0 + let waits = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + + for await (const item of retryOpenAIStream( + async () => { + calls++ + return calls === 1 ? interruptedStream() : resumedStream('lo world') + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => { + waits++ + }, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + const text = outputEvents + .filter( + event => + event.type === 'content_block_delta' && + event.delta.type === 'text_delta', + ) + .map(event => + event.type === 'content_block_delta' && + event.delta.type === 'text_delta' + ? event.delta.text + : '', + ) + .join('') + + expect(calls).toBe(2) + expect(waits).toBe(1) + expect(text).toBe('Hello world') + expect( + outputEvents.filter(event => event.type === 'message_start'), + ).toHaveLength(1) + expect( + outputEvents.filter(event => event.type === 'content_block_start'), + ).toHaveLength(1) + }) + + test('restarts from scratch when the retried stream diverges', async () => { + let calls = 0 + const attempts = new Map() + + for await (const item of retryOpenAIStream( + async () => { + calls++ + return calls === 1 ? interruptedStream() : resumedStream('p') + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => {}, + }, + )) { + const current = attempts.get(item.attempt) ?? { text: '', stopped: false } + attempts.set(item.attempt, current) + if (item.outputEvent) { + current.text += collectText([item.outputEvent]) + if (item.outputEvent.type === 'message_stop') current.stopped = true + } + } + + expect(calls).toBe(3) + expect([...attempts].map(([n, a]) => [n, a.text, a.stopped])).toEqual([ + [1, 'Hello ', false], + [2, '', false], + [3, 'Help', true], + ]) + }) + + test('reports divergence instead of restarting when no retry is left', async () => { + let calls = 0 + const stream = retryOpenAIStream( + async () => { + calls++ + return calls === 1 ? interruptedStream() : resumedStream('p') + }, + { + maxRetries: 1, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow(OpenAIStreamResumeError) + expect(calls).toBe(2) + }) + + test('stops restarting once the restart budget is spent', async () => { + let calls = 0 + async function* emitThenFail(text: string) { + yield messageStart() + yield blockStart() + yield textDelta(text) + throw new Error('terminated') + } + const next = [ + () => emitThenFail('Hello '), + () => resumedStream('p'), + () => emitThenFail('Help'), + () => resumedStream('q'), + ] + + const stream = retryOpenAIStream( + async () => { + calls++ + return next[Math.min(calls, next.length) - 1]() + }, + { + maxRetries: 5, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow(OpenAIStreamResumeError) + expect(calls).toBe(4) + }) + + test('resumes partial tool input without duplicating JSON', async () => { + let calls = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + async function* firstAttempt() { + yield messageStart() + yield toolStart() + yield inputDelta('{"command":"ec') + throw new Error('terminated') + } + async function* secondAttempt() { + yield messageStart() + yield toolStart() + yield inputDelta('{"command":') + yield inputDelta('"echo hi"}') + yield blockStop() + yield messageStop() + } + + for await (const item of retryOpenAIStream( + async () => { + calls++ + return calls === 1 ? firstAttempt() : secondAttempt() + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => {}, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + const json = outputEvents + .filter( + event => + event.type === 'content_block_delta' && + event.delta.type === 'input_json_delta', + ) + .map(event => + event.type === 'content_block_delta' && + event.delta.type === 'input_json_delta' + ? event.delta.partial_json + : '', + ) + .join('') + + expect(json).toBe('{"command":"echo hi"}') + expect( + outputEvents.filter(event => event.type === 'content_block_start'), + ).toHaveLength(1) + }) + + test('does not retry failures before the stream emits progress', async () => { + let calls = 0 + const stream = retryOpenAIStream( + async () => { + calls++ + throw new Error('terminated') + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow('terminated') + expect(calls).toBe(1) + }) + + test('retries a truncated stream that emitted nothing', async () => { + let calls = 0 + let waits = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + + for await (const item of retryOpenAIStream( + async () => { + calls++ + return calls === 1 ? emptyStream() : completedStream() + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => { + waits++ + }, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + expect(calls).toBe(2) + expect(waits).toBe(1) + expect(collectText(outputEvents)).toBe('Hello world') + expect( + outputEvents.filter(event => event.type === 'message_stop'), + ).toHaveLength(1) + }) + + test('gives up once the empty stream budget is spent', async () => { + let calls = 0 + + const stream = retryOpenAIStream( + async () => { + calls++ + return emptyStream() + }, + { + maxRetries: 10, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow(OpenAIStreamIncompleteError) + expect(calls).toBe(3) + }) + + test('honors the configured retry count after progress', async () => { + let calls = 0 + const stream = retryOpenAIStream( + async () => { + calls++ + return interruptedStream() + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow('terminated') + expect(calls).toBe(4) + }) + + test('retries a stream that ends without a terminal event', async () => { + let calls = 0 + let waits = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + + for await (const item of retryOpenAIStream( + async () => { + calls++ + return calls === 1 ? truncatedStream() : completedStream() + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => { + waits++ + }, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + expect(calls).toBe(2) + expect(waits).toBe(1) + expect(collectText(outputEvents)).toBe('Hello world') + expect( + outputEvents.filter(event => event.type === 'message_stop'), + ).toHaveLength(1) + }) + + test('gives up after the configured retries when no terminal event arrives', async () => { + let calls = 0 + const stream = retryOpenAIStream( + async () => { + calls++ + return truncatedStream() + }, + { + maxRetries: 2, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow(OpenAIStreamIncompleteError) + expect(calls).toBe(3) + }) + + test('returns a tolerated incomplete stream when OPENAI_ALLOW_INCOMPLETE_STREAM is set', async () => { + process.env.OPENAI_ALLOW_INCOMPLETE_STREAM = '1' + try { + const outputEvents: BetaRawMessageStreamEvent[] = [] + for await (const item of retryOpenAIStream( + async () => truncatedStream(), + { + maxRetries: 2, + signal: new AbortController().signal, + wait: async () => {}, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + expect(collectText(outputEvents)).toBe('Hello ') + expect( + outputEvents.filter(event => event.type === 'message_stop'), + ).toHaveLength(0) + } finally { + delete process.env.OPENAI_ALLOW_INCOMPLETE_STREAM + } + }) + + test('retries a stream that stalls without closing the connection', async () => { + let calls = 0 + let waits = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + + for await (const item of retryOpenAIStream( + async (signal: AbortSignal) => { + calls++ + return calls === 1 + ? stallingStream(signal, true) + : resumedStream('lo world') + }, + { + maxRetries: 3, + idleTimeoutMs: 20, + signal: new AbortController().signal, + wait: async () => { + waits++ + }, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + expect(calls).toBe(2) + expect(waits).toBe(1) + expect(collectText(outputEvents)).toBe('Hello world') + expect( + outputEvents.filter(event => event.type === 'message_stop'), + ).toHaveLength(1) + }) + + test('retries a stall that happens before the first event', async () => { + let calls = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + + for await (const item of retryOpenAIStream( + async (signal: AbortSignal) => { + calls++ + return calls === 1 + ? stallingStream(signal, false) + : resumedStream('lo world') + }, + { + maxRetries: 3, + idleTimeoutMs: 20, + signal: new AbortController().signal, + wait: async () => {}, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + expect(calls).toBe(2) + expect(collectText(outputEvents)).toBe('Hello world') + }) + + test('stops once the idle timeout budget is spent', async () => { + let calls = 0 + const stream = retryOpenAIStream( + async (signal: AbortSignal) => { + calls++ + return stallingStream(signal, true) + }, + { + maxRetries: 10, + idleTimeoutMs: 20, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow(OpenAIStreamIdleTimeoutError) + expect(calls).toBe(3) + }) + + test('ends quietly when the caller aborts a stalled stream', async () => { + const controller = new AbortController() + const outputEvents: BetaRawMessageStreamEvent[] = [] + let calls = 0 + + const consume = async () => { + for await (const item of retryOpenAIStream( + async (signal: AbortSignal) => { + calls++ + return stallingStream(signal, true) + }, + { + maxRetries: 3, + idleTimeoutMs: 5_000, + signal: controller.signal, + wait: async () => {}, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + if (item.event.type === 'content_block_start') controller.abort() + } + } + + await expect(consume()).resolves.toBeUndefined() + expect(calls).toBe(1) + expect( + outputEvents.filter(event => event.type === 'message_stop'), + ).toHaveLength(0) + }) +}) + +describe('isRetryableOpenAIStreamError', () => { + test('recognizes terminated and nested socket errors', () => { + expect(isRetryableOpenAIStreamError(new Error('terminated'))).toBe(true) + expect( + isRetryableOpenAIStreamError( + new Error('stream failed', { + cause: Object.assign(new Error('other side closed'), { + code: 'UND_ERR_SOCKET', + }), + }), + ), + ).toBe(true) + }) + + test('retries streams that ended without a terminal event', () => { + expect( + isRetryableOpenAIStreamError( + new OpenAIStreamIncompleteError('ended without finish_reason'), + ), + ).toBe(true) + expect( + isRetryableOpenAIStreamError( + new Error('wrapper', { + cause: new OpenAIStreamIncompleteError('ended without finish_reason'), + }), + ), + ).toBe(true) + }) + + test('retries streams that stalled', () => { + expect( + isRetryableOpenAIStreamError(new OpenAIStreamIdleTimeoutError(90_000)), + ).toBe(true) + expect( + isRetryableOpenAIStreamError( + new Error('wrapper', { + cause: new OpenAIStreamIdleTimeoutError(90_000), + }), + ), + ).toBe(true) + }) + + test('does not retry user aborts or ordinary API errors', () => { + expect( + isRetryableOpenAIStreamError( + Object.assign(new Error('Request was aborted'), { + name: 'APIUserAbortError', + }), + ), + ).toBe(false) + expect( + isRetryableOpenAIStreamError( + Object.assign(new Error('bad request'), { status: 400 }), + ), + ).toBe(false) + expect( + isRetryableOpenAIStreamError( + Object.assign(new Error('terminated'), { status: 400 }), + ), + ).toBe(false) + }) +}) + +describe('truncated upstream response', () => { + function chunks( + items: Array<{ content?: string; finishReason?: string | null }>, + ) { + return items.map( + item => + ({ + id: 'chatcmpl-test', + object: 'chat.completion.chunk', + created: 0, + model: 'gpt-4o', + choices: [ + { + index: 0, + delta: + item.content === undefined ? {} : { content: item.content }, + finish_reason: item.finishReason ?? null, + }, + ], + }) as unknown as ChatCompletionChunk, + ) + } + + function chunkStream( + items: ChatCompletionChunk[], + ): AsyncIterable { + return { + [Symbol.asyncIterator](): AsyncIterator { + let index = 0 + return { + async next(): Promise> { + if (index >= items.length) return { done: true, value: undefined } + return { done: false, value: items[index++] } + }, + } + }, + } + } + + const firstAttempt = chunks([{ content: 'Hello ' }, { content: 'wor' }]) + const secondAttempt = chunks([ + { content: 'Hello ' }, + { content: 'wor' }, + { content: 'ld' }, + { finishReason: 'stop' }, + ]) + + test('retries the cut connection and finishes the answer', async () => { + let attempts = 0 + let waits = 0 + const outputEvents: BetaRawMessageStreamEvent[] = [] + + for await (const item of retryOpenAIStream( + async () => { + attempts++ + return adaptOpenAIStreamToAnthropic( + chunkStream(attempts === 1 ? firstAttempt : secondAttempt), + 'gpt-4o', + ) + }, + { + maxRetries: 3, + signal: new AbortController().signal, + wait: async () => { + waits++ + }, + }, + )) { + if (item.outputEvent) outputEvents.push(item.outputEvent) + } + + expect(attempts).toBe(2) + expect(waits).toBe(1) + expect(collectText(outputEvents)).toBe('Hello world') + expect( + outputEvents.filter(event => event.type === 'message_stop'), + ).toHaveLength(1) + expect( + outputEvents.filter(event => event.type === 'content_block_start'), + ).toHaveLength(1) + }) + + test('reports a terminal error when every attempt is cut', async () => { + let attempts = 0 + const stream = retryOpenAIStream( + async () => { + attempts++ + return adaptOpenAIStreamToAnthropic(chunkStream(firstAttempt), 'gpt-4o') + }, + { + maxRetries: 2, + signal: new AbortController().signal, + wait: async () => {}, + }, + ) + + await expect(async () => { + for await (const _item of stream) { + } + }).toThrow(OpenAIStreamIncompleteError) + expect(attempts).toBe(3) + }) +}) diff --git a/src/services/api/openai/client.ts b/src/services/api/openai/client.ts index 5ee37cd414..47d5ced207 100644 --- a/src/services/api/openai/client.ts +++ b/src/services/api/openai/client.ts @@ -10,10 +10,21 @@ import { getProxyFetchOptions } from 'src/utils/proxy.js' * OPENAI_BASE_URL: Recommended. Base URL for the endpoint (e.g. http://localhost:11434/v1). * OPENAI_ORG_ID: Optional. Organization ID. * OPENAI_PROJECT_ID: Optional. Project ID. + * OPENAI_MAX_RETRIES: Optional. Retries for retryable failures (429, 408, 409, + * 5xx, connection errors). Defaults to 10. */ let cachedClient: OpenAI | null = null +const DEFAULT_MAX_RETRIES = 10 + +export function getOpenAIMaxRetries(override?: number): number { + if (override !== undefined) return override + const raw = process.env.OPENAI_MAX_RETRIES + const parsed = raw ? Number.parseInt(raw, 10) : Number.NaN + return Number.isFinite(parsed) && parsed >= 0 ? parsed : DEFAULT_MAX_RETRIES +} + /** * Wrap a fetch so that every response's rate-limit headers are fed into the * provider usage store. Errors in parsing must never break the request. @@ -52,7 +63,7 @@ export function getOpenAIClient(options?: { const client = new OpenAI({ apiKey, ...(baseURL && { baseURL }), - maxRetries: options?.maxRetries ?? 0, + maxRetries: getOpenAIMaxRetries(options?.maxRetries), timeout: parseInt(process.env.API_TIMEOUT_MS || String(600 * 1000), 10), dangerouslyAllowBrowser: true, ...(process.env.OPENAI_ORG_ID && { diff --git a/src/services/api/openai/index.ts b/src/services/api/openai/index.ts index 0aaaa763ec..03ff4c94d3 100644 --- a/src/services/api/openai/index.ts +++ b/src/services/api/openai/index.ts @@ -14,7 +14,7 @@ import type { import type { AgentId } from '../../../types/ids.js' import type { Tools } from '../../../Tool.js' import { getSessionId } from '../../../bootstrap/state.js' -import { getOpenAIClient } from './client.js' +import { getOpenAIClient, getOpenAIMaxRetries } from './client.js' import { formatOpenAIPromptCacheKey, getOfficialOpenAIPromptCacheKey, @@ -77,6 +77,7 @@ import { isDeferredTool, SEARCH_EXTRA_TOOLS_TOOL_NAME, } from '@claude-code-best/builtin-tools/tools/SearchExtraToolsTool/prompt.js' +import { retryOpenAIStream } from './streamRetry.js' function convertToResponsesReasoningEffort( effortValue: unknown, @@ -367,52 +368,70 @@ export async function* queryModelOpenAI( `[OpenAI] Calling model=${openaiModel}, messages=${openaiMessages.length}, tools=${openaiTools.length}, thinking=${enableThinking}${promptCacheKey ? `, prompt_cache_key=${promptCacheKey}` : ''}`, ) - // 11. Call OpenAI API with streaming. ChatGPT subscription auth uses the - // Codex Responses backend; API-key/OpenAI-compatible auth keeps the - // existing Chat Completions adapter. - const adaptedStream = useChatGPTResponses - ? adaptResponsesStreamToAnthropic( - await createChatGPTResponsesStream({ - request: buildResponsesRequest({ - model: openaiModel, - messages: openaiMessages, - tools: openaiTools, - toolChoice: openaiToolChoice, - reasoningEffort, - promptCacheKey: sessionPromptCacheKey, + const createAdaptedStream = async (attemptSignal: AbortSignal) => + useChatGPTResponses + ? adaptResponsesStreamToAnthropic( + await createChatGPTResponsesStream({ + request: buildResponsesRequest({ + model: openaiModel, + messages: openaiMessages, + tools: openaiTools, + toolChoice: openaiToolChoice, + reasoningEffort, + promptCacheKey: sessionPromptCacheKey, + }), + signal: attemptSignal, + fetchOverride: options.fetchOverride as unknown as typeof fetch, }), - signal, - fetchOverride: options.fetchOverride as unknown as typeof fetch, - }), - openaiModel, + openaiModel, + ) + : adaptOpenAIStreamToAnthropic( + await getOpenAIClient({ + fetchOverride: options.fetchOverride as unknown as typeof fetch, + source: options.querySource, + }).chat.completions.create( + buildOpenAIRequestBody({ + model: openaiModel, + messages: openaiMessages, + tools: openaiTools, + toolChoice: openaiToolChoice, + enableThinking, + maxTokens, + temperatureOverride: options.temperatureOverride, + promptCacheKey, + }), + { signal: attemptSignal }, + ), + openaiModel, + { includeCacheWriteTokens: useOfficialOpenAICache }, + ) + + const adaptedStream = retryOpenAIStream(createAdaptedStream, { + maxRetries: getOpenAIMaxRetries(), + signal, + onRetry: (error, attempt, delayMs) => { + const message = error instanceof Error ? error.message : String(error) + logForDebugging( + `[OpenAI] Stream interrupted (attempt ${attempt}), retrying in ${Math.round(delayMs)}ms: ${message}`, + { level: 'error' }, ) - : adaptOpenAIStreamToAnthropic( - await getOpenAIClient({ - maxRetries: 0, - fetchOverride: options.fetchOverride as unknown as typeof fetch, - source: options.querySource, - }).chat.completions.create( - buildOpenAIRequestBody({ - model: openaiModel, - messages: openaiMessages, - tools: openaiTools, - toolChoice: openaiToolChoice, - enableThinking, - maxTokens, - temperatureOverride: options.temperatureOverride, - promptCacheKey, - }), - { signal }, - ), - openaiModel, - { includeCacheWriteTokens: useOfficialOpenAICache }, + }, + onIdle: (phase, ms) => { + const seconds = Math.round(ms / 1000) + logForDebugging( + phase === 'warning' + ? `[OpenAI] Stream idle for ${seconds}s, still waiting` + : `[OpenAI] Stream idle for ${seconds}s, aborting request`, + { level: phase === 'warning' ? 'warn' : 'error' }, ) + }, + }) // 12. Convert OpenAI stream to Anthropic events, then process into // AssistantMessage + StreamEvent (matching the Anthropic path behavior) // Accumulate content blocks and usage, same as the Anthropic path in claude.ts - const contentBlocks: Record> = {} + let contentBlocks: Record> = {} const collectedMessages: AssistantMessage[] = [] let partialMessage: BetaMessage | null = null let stopReason: string | null = null @@ -424,8 +443,21 @@ export async function* queryModelOpenAI( } let ttftMs = 0 const start = Date.now() + let activeAttempt = 0 - for await (const event of adaptedStream) { + for await (const { attempt, event, outputEvent } of adaptedStream) { + if (attempt !== activeAttempt) { + activeAttempt = attempt + contentBlocks = {} + partialMessage = null + stopReason = null + usage = { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + } + } switch (event.type) { case 'message_start': { partialMessage = event.message @@ -526,12 +558,13 @@ export async function* queryModelOpenAI( } } - // Also yield as StreamEvent for real-time display (matching Anthropic path) - yield { - type: 'stream_event', - event, - ...(event.type === 'message_start' ? { ttftMs } : undefined), - } as StreamEvent + if (outputEvent) { + yield { + type: 'stream_event', + event: outputEvent, + ...(outputEvent.type === 'message_start' ? { ttftMs } : undefined), + } as StreamEvent + } } // Record LLM observation in Langfuse (no-op if not configured) @@ -568,6 +601,9 @@ export async function* queryModelOpenAI( } } } catch (error) { + // Don't surface an error message for user aborts, the interruption message + // is handled in query.ts. + if (signal.aborted) return const errorMessage = error instanceof Error ? error.message : String(error) logForDebugging(`[OpenAI] Error: ${errorMessage}`, { level: 'error' }) yield createAssistantAPIErrorMessage({ diff --git a/src/services/api/openai/responsesAdapter.ts b/src/services/api/openai/responsesAdapter.ts index 74b3c739a7..8a35558511 100644 --- a/src/services/api/openai/responsesAdapter.ts +++ b/src/services/api/openai/responsesAdapter.ts @@ -1,6 +1,11 @@ import { randomUUID } from 'crypto' import type { BetaRawMessageStreamEvent } from '@anthropic-ai/sdk/resources/beta/messages/messages.mjs' -import { normalizeOpenAIUsage, type AnthropicUsage } from '@ant/model-provider' +import { + normalizeOpenAIUsage, + type AnthropicUsage, + allowsIncompleteOpenAIStream, + OpenAIStreamIncompleteError, +} from '@ant/model-provider' import { getValidChatGPTAuth } from './chatgptAuth.js' type ResponsesInputItem = Record @@ -285,6 +290,7 @@ export async function* adaptResponsesStreamToAnthropic( let currentContentIndex = -1 let textBlockOpen = false let thinkingBlockOpen = false + let terminalSeen = false const ensureStarted = async function* () { if (started) return @@ -443,6 +449,7 @@ export async function* adaptResponsesStreamToAnthropic( } if (type === 'response.completed' || type === 'response.incomplete') { + terminalSeen = true if (textBlockOpen) { yield { type: 'content_block_stop', @@ -466,6 +473,14 @@ export async function* adaptResponsesStreamToAnthropic( yield { type: 'message_stop' } as BetaRawMessageStreamEvent } } + + if (!terminalSeen && !allowsIncompleteOpenAIStream()) { + throw new OpenAIStreamIncompleteError( + started + ? 'ChatGPT Responses stream ended before the response was finished' + : 'ChatGPT Responses stream ended without returning any data', + ) + } } export async function createChatGPTResponsesStream(params: { diff --git a/src/services/api/openai/streamIdleTimeout.ts b/src/services/api/openai/streamIdleTimeout.ts new file mode 100644 index 0000000000..36ee16646c --- /dev/null +++ b/src/services/api/openai/streamIdleTimeout.ts @@ -0,0 +1,89 @@ +const DEFAULT_IDLE_TIMEOUT_MS = 90_000 + +/** + * Raised when the upstream stops emitting events without closing the + * connection. Retryable: the request is re-issued from scratch. + */ +export class OpenAIStreamIdleTimeoutError extends Error { + constructor(timeoutMs: number) { + super( + `OpenAI stream stalled: no events received for ${Math.round(timeoutMs / 1000)}s`, + ) + this.name = 'OpenAIStreamIdleTimeoutError' + } +} + +export function getOpenAIStreamIdleTimeoutMs(): number { + const raw = process.env.OPENAI_STREAM_IDLE_TIMEOUT_MS?.trim() + if (!raw) return DEFAULT_IDLE_TIMEOUT_MS + if (raw === '0' || raw.toLowerCase() === 'off') return 0 + const parsed = Number.parseInt(raw, 10) + return Number.isFinite(parsed) && parsed > 0 + ? parsed + : DEFAULT_IDLE_TIMEOUT_MS +} + +export interface StreamIdleWatchdogOptions { + timeoutMs: number + /** Must release the source — the pending read stays blocked otherwise. */ + onTimeout: (timeoutMs: number) => void + onWarning?: (warningMs: number) => void +} + +/** + * Fails the read when the source goes quiet for `timeoutMs`. The SDK's request + * timeout only covers the initial fetch, so an upstream that answers with + * headers and then stalls would otherwise block forever. + */ +export async function* watchStreamIdle( + source: AsyncIterable, + options: StreamIdleWatchdogOptions, +): AsyncGenerator { + const { timeoutMs } = options + if (timeoutMs <= 0) { + yield* source + return + } + + const warningMs = Math.floor(timeoutMs / 2) + const iterator = source[Symbol.asyncIterator]() + let warningTimer: ReturnType | null = null + let idleTimer: ReturnType | null = null + + const clearTimers = (): void => { + if (warningTimer !== null) { + clearTimeout(warningTimer) + warningTimer = null + } + if (idleTimer !== null) { + clearTimeout(idleTimer) + idleTimer = null + } + } + + try { + while (true) { + clearTimers() + const stalled = new Promise((_, reject) => { + if (options.onWarning) { + warningTimer = setTimeout( + () => options.onWarning?.(warningMs), + warningMs, + ) + } + idleTimer = setTimeout(() => { + options.onTimeout(timeoutMs) + reject(new OpenAIStreamIdleTimeoutError(timeoutMs)) + }, timeoutMs) + }) + + const next = iterator.next() + next.catch(() => {}) + const result = await Promise.race([next, stalled]) + if (result.done) return + yield result.value + } + } finally { + clearTimers() + } +} diff --git a/src/services/api/openai/streamRetry.ts b/src/services/api/openai/streamRetry.ts new file mode 100644 index 0000000000..3962e2f3b3 --- /dev/null +++ b/src/services/api/openai/streamRetry.ts @@ -0,0 +1,407 @@ +import type { BetaRawMessageStreamEvent } from '@anthropic-ai/sdk/resources/beta/messages/messages.mjs' +import { + allowsIncompleteOpenAIStream, + OpenAIStreamIncompleteError, +} from '@ant/model-provider' +import { sleep } from '../../../utils/sleep.js' +import { + getOpenAIStreamIdleTimeoutMs, + OpenAIStreamIdleTimeoutError, + watchStreamIdle, +} from './streamIdleTimeout.js' + +type RetryableError = Error & { + cause?: unknown + code?: string + status?: number +} + +type StreamFactory = ( + signal: AbortSignal, +) => Promise> + +export interface ResumableOpenAIStreamEvent { + attempt: number + event: BetaRawMessageStreamEvent + outputEvent?: BetaRawMessageStreamEvent +} + +export interface OpenAIStreamRetryOptions { + maxRetries: number + signal: AbortSignal + onRetry?: (error: unknown, attempt: number, delayMs: number) => void + onIdle?: (phase: 'warning' | 'timeout', ms: number) => void + idleTimeoutMs?: number + wait?: (delayMs: number, signal: AbortSignal) => Promise +} + +/** + * A stall is rarely transient in the same way a dropped socket is, so idle + * timeouts get a smaller budget than the provider's retry count. + */ +const MAX_IDLE_TIMEOUT_RETRIES = 2 + +/** + * A stream that ends with no events at all is safe to re-issue: nothing was + * emitted to the caller, so no prefix has to be reconciled. + */ +const MAX_EMPTY_STREAM_RETRIES = 2 + +/** + * Resuming assumes the re-issued request regenerates the emitted prefix + * verbatim. When it does not, the prefix is abandoned and the request is + * re-issued from scratch — at the cost of re-rendering what the caller saw. + */ +const MAX_RESUME_RESTARTS = 1 + +export class OpenAIStreamResumeError extends Error { + constructor(message: string) { + super(message) + this.name = 'OpenAIStreamResumeError' + } +} + +class OpenAIStreamResumeState { + private messageStarted = false + private readonly blockDescriptors = new Map() + private readonly stoppedBlocks = new Set() + private readonly emittedValues = new Map() + private readonly attemptStartedBlocks = new Set() + private readonly attemptValues = new Map() + + startAttempt(): void { + this.attemptStartedBlocks.clear() + this.attemptValues.clear() + } + + reset(): void { + this.messageStarted = false + this.blockDescriptors.clear() + this.stoppedBlocks.clear() + this.emittedValues.clear() + this.attemptStartedBlocks.clear() + this.attemptValues.clear() + } + + filter( + event: BetaRawMessageStreamEvent, + ): BetaRawMessageStreamEvent | undefined { + switch (event.type) { + case 'message_start': { + if (this.messageStarted) return undefined + this.messageStarted = true + return event + } + case 'content_block_start': { + const descriptor = JSON.stringify({ + type: event.content_block.type, + ...('name' in event.content_block + ? { name: event.content_block.name } + : {}), + }) + const previous = this.blockDescriptors.get(event.index) + this.attemptStartedBlocks.add(event.index) + if (previous === undefined) { + this.blockDescriptors.set(event.index, descriptor) + return event + } + if (previous !== descriptor) { + throw new OpenAIStreamResumeError( + `OpenAI stream changed content block ${event.index} while resuming`, + ) + } + return undefined + } + case 'content_block_delta': + return this.filterDelta(event) + case 'content_block_stop': { + this.assertBlockCaughtUp(event.index) + if (this.stoppedBlocks.has(event.index)) return undefined + this.stoppedBlocks.add(event.index) + return event + } + case 'message_stop': + this.assertAttemptCaughtUp() + return event + default: + return event + } + } + + private filterDelta( + event: Extract, + ): BetaRawMessageStreamEvent | undefined { + const value = getDeltaValue(event) + if (value === undefined) return event + + const key = `${event.index}:${event.delta.type}` + if (event.delta.type === 'signature_delta') { + const previous = this.emittedValues.get(key) + this.attemptValues.set(key, value) + if (previous === undefined) { + this.emittedValues.set(key, value) + return event + } + if (previous !== value) { + throw new OpenAIStreamResumeError( + `OpenAI stream changed signature for content block ${event.index} while resuming`, + ) + } + return undefined + } + + const attemptValue = (this.attemptValues.get(key) ?? '') + value + const emittedValue = this.emittedValues.get(key) ?? '' + this.attemptValues.set(key, attemptValue) + + if (emittedValue.startsWith(attemptValue)) return undefined + if (!attemptValue.startsWith(emittedValue)) { + throw new OpenAIStreamResumeError( + `OpenAI stream content diverged at content block ${event.index} while resuming`, + ) + } + + const suffix = attemptValue.slice(emittedValue.length) + this.emittedValues.set(key, attemptValue) + return suffix.length > 0 ? withDeltaValue(event, suffix) : undefined + } + + private assertBlockCaughtUp(index: number): void { + for (const [key, emittedValue] of this.emittedValues) { + if (!key.startsWith(`${index}:`)) continue + if (this.attemptValues.get(key) !== emittedValue) { + throw new OpenAIStreamResumeError( + `OpenAI stream ended content block ${index} before reaching the previous output`, + ) + } + } + } + + private assertAttemptCaughtUp(): void { + for (const index of this.blockDescriptors.keys()) { + if (!this.attemptStartedBlocks.has(index)) { + throw new OpenAIStreamResumeError( + `OpenAI stream ended before replaying content block ${index}`, + ) + } + this.assertBlockCaughtUp(index) + } + } +} + +function getDeltaValue( + event: Extract, +): string | undefined { + switch (event.delta.type) { + case 'text_delta': + return event.delta.text + case 'input_json_delta': + return event.delta.partial_json + case 'thinking_delta': + return event.delta.thinking + case 'signature_delta': + return event.delta.signature + default: + return undefined + } +} + +function withDeltaValue( + event: Extract, + value: string, +): BetaRawMessageStreamEvent { + switch (event.delta.type) { + case 'text_delta': + return { ...event, delta: { ...event.delta, text: value } } + case 'input_json_delta': + return { ...event, delta: { ...event.delta, partial_json: value } } + case 'thinking_delta': + return { ...event, delta: { ...event.delta, thinking: value } } + default: + return event + } +} + +function getErrorChain(error: unknown): RetryableError[] { + const chain: RetryableError[] = [] + let current = error + const seen = new Set() + while (current instanceof Error && !seen.has(current)) { + seen.add(current) + chain.push(current as RetryableError) + current = (current as RetryableError).cause + } + return chain +} + +export function isRetryableOpenAIStreamError(error: unknown): boolean { + const chain = getErrorChain(error) + if ( + chain.some( + item => item.name === 'AbortError' || item.name === 'APIUserAbortError', + ) + ) { + return false + } + + if ( + chain.some( + item => + item instanceof OpenAIStreamIncompleteError || + item instanceof OpenAIStreamIdleTimeoutError, + ) + ) { + return true + } + + return chain.some(item => { + if (item.status !== undefined) { + return ( + item.status === 408 || + item.status === 409 || + item.status === 429 || + item.status >= 500 + ) + } + if ( + item.name === 'APIConnectionError' || + item.name === 'APIConnectionTimeoutError' + ) { + return true + } + const code = item.code?.toUpperCase() + if ( + code === 'ECONNRESET' || + code === 'EPIPE' || + code === 'ETIMEDOUT' || + code === 'UND_ERR_SOCKET' + ) { + return true + } + return /\bterminated\b|fetch failed|socket connection was closed|other side closed/i.test( + item.message, + ) + }) +} + +export function getOpenAIStreamRetryDelay(attempt: number): number { + const baseDelay = Math.min(500 * 2 ** (attempt - 1), 8000) + return baseDelay + Math.random() * 0.25 * baseDelay +} + +export async function* retryOpenAIStream( + createStream: StreamFactory, + options: OpenAIStreamRetryOptions, +): AsyncGenerator { + const resumeState = new OpenAIStreamResumeState() + const idleTimeoutMs = options.idleTimeoutMs ?? getOpenAIStreamIdleTimeoutMs() + let hasProgress = false + let idleTimeouts = 0 + let emptyStreams = 0 + let restarts = 0 + + for (let attempt = 1; attempt <= options.maxRetries + 1; attempt++) { + if (options.signal.aborted) throw new Error('Request was aborted') + resumeState.startAttempt() + let completed = false + let restart = false + + // Each attempt owns a controller so the idle watchdog can kill a stalled + // request without aborting the caller's signal, which means "user + // interrupted the whole turn". + const attemptController = new AbortController() + const abortAttempt = (): void => attemptController.abort() + options.signal.addEventListener('abort', abortAttempt, { once: true }) + + try { + const source = await createStream(attemptController.signal) + const stream = watchStreamIdle(source, { + timeoutMs: idleTimeoutMs, + onTimeout: () => { + abortAttempt() + options.onIdle?.('timeout', idleTimeoutMs) + }, + ...(options.onIdle && { + onWarning: (warningMs: number) => + options.onIdle?.('warning', warningMs), + }), + }) + + for await (const event of stream) { + hasProgress = true + let outputEvent: BetaRawMessageStreamEvent | undefined + try { + outputEvent = resumeState.filter(event) + } catch (error) { + if ( + !(error instanceof OpenAIStreamResumeError) || + restarts >= MAX_RESUME_RESTARTS || + attempt > options.maxRetries + ) { + throw error + } + // The re-issued request regenerated a different prefix, so the + // resume is unsound: discard what was emitted and start over. + restarts++ + resumeState.reset() + restart = true + options.onRetry?.(error, attempt, 0) + abortAttempt() + break + } + if (event.type === 'message_stop') completed = true + yield { attempt, event, ...(outputEvent && { outputEvent }) } + } + + if (restart) continue + if (!completed) { + // The SDK swallows abort errors and ends the iterator cleanly, so an + // aborted read is indistinguishable from a truncated stream here. + if (options.signal.aborted) return + if (!allowsIncompleteOpenAIStream()) { + throw new OpenAIStreamIncompleteError( + 'OpenAI stream ended without a terminal event', + ) + } + } + return + } catch (error) { + if (completed) return + if (options.signal.aborted) throw error + const idleTimeout = error instanceof OpenAIStreamIdleTimeoutError + if (idleTimeout) idleTimeouts++ + // A stream that produced nothing can be re-issued even though the SDK + // reports its truncation as a normal end, so it bypasses the progress + // gate for the same reason a stall does. + const emptyStream = + error instanceof OpenAIStreamIncompleteError && !hasProgress + if (emptyStream) emptyStreams++ + // A stall can hit before the first event, so idle timeouts bypass the + // progress gate that otherwise defers to the SDK's own retries. + const hasBudget = idleTimeout + ? idleTimeouts <= MAX_IDLE_TIMEOUT_RETRIES + : emptyStream + ? emptyStreams <= MAX_EMPTY_STREAM_RETRIES + : hasProgress + if ( + !hasBudget || + attempt > options.maxRetries || + !isRetryableOpenAIStreamError(error) + ) { + throw error + } + const delayMs = getOpenAIStreamRetryDelay(attempt) + options.onRetry?.(error, attempt, delayMs) + await (options.wait ?? waitForRetry)(delayMs, options.signal) + } finally { + options.signal.removeEventListener('abort', abortAttempt) + } + } +} + +async function waitForRetry( + delayMs: number, + signal: AbortSignal, +): Promise { + await sleep(delayMs, signal, { throwOnAbort: true }) +} diff --git a/src/utils/__tests__/sideQuery.chatgptAuth.test.ts b/src/utils/__tests__/sideQuery.chatgptAuth.test.ts index c39c9bb4f7..0123d0543e 100644 --- a/src/utils/__tests__/sideQuery.chatgptAuth.test.ts +++ b/src/utils/__tests__/sideQuery.chatgptAuth.test.ts @@ -9,7 +9,7 @@ * ChatGPT Responses + OAuth path used by the main loop. * * Avoid mocking getAPIProvider (process-global pollution). Select OpenAI via - * CLAUDE_CODE_USE_OPENAI env. Mock only client + ChatGPT token surface. + * CLAUDE_CODE_USE_OPENAI env. Mock only the OpenAI SDK + ChatGPT token surface. */ import { afterEach, beforeEach, describe, expect, mock, test } from 'bun:test' import { logMock } from '../../../tests/mocks/log' @@ -26,49 +26,66 @@ mock.module('src/services/analytics/index.js', () => ({ _resetForTesting: () => {}, })) -let getOpenAIClientCallCount = 0 +let openaiClientConstructCount = 0 let chatCompletionsCreateCount = 0 let lastChatCompletionsArgs: Record | null = null let chatCompletionsUsage: Record = {} -mock.module('src/services/api/openai/client.js', () => ({ - getOpenAIClient: () => { - getOpenAIClientCallCount++ - return { - chat: { - completions: { - create: async (args: Record) => { - chatCompletionsCreateCount++ - lastChatCompletionsArgs = args - return { - id: 'chatcmpl_test', - choices: [ - { - finish_reason: 'tool_calls', - message: { - content: null, - tool_calls: [ - { - type: 'function', - id: 'call_api_key', - function: { - name: 'classify_result', - arguments: JSON.stringify({ shouldBlock: false }), - }, +// Mock the `openai` SDK (third-party network lib) instead of the local +// client wrapper, so the real client module stays intact for other test +// files that import it (mock.module is process-global). +mock.module('openai', () => { + class MockOpenAI { + apiKey: string + baseURL?: string + maxRetries: number + + constructor(opts: { + apiKey: string + baseURL?: string + maxRetries?: number + }) { + openaiClientConstructCount++ + this.apiKey = opts.apiKey + this.baseURL = opts.baseURL + this.maxRetries = opts.maxRetries ?? 0 + } + + chat = { + completions: { + create: async (args: Record) => { + chatCompletionsCreateCount++ + lastChatCompletionsArgs = args + return { + id: 'chatcmpl_test', + choices: [ + { + finish_reason: 'tool_calls', + message: { + content: null, + tool_calls: [ + { + type: 'function', + id: 'call_api_key', + function: { + name: 'classify_result', + arguments: JSON.stringify({ shouldBlock: false }), }, - ], - }, + }, + ], }, - ], - usage: chatCompletionsUsage, - } - }, + }, + ], + usage: chatCompletionsUsage, + } }, }, } - }, - clearOpenAIClientCache: () => {}, -})) + } + return { default: MockOpenAI } +}) + +import { clearOpenAIClientCache } from '../../services/api/openai/client.js' // Keep isChatGPTAuthEnabled env-driven (same as production) so other suite // files are not forced into ChatGPT mode. @@ -157,8 +174,9 @@ beforeEach(() => { for (const key of ENV_KEYS) { savedEnv[key] = process.env[key] } - getOpenAIClientCallCount = 0 + openaiClientConstructCount = 0 chatCompletionsCreateCount = 0 + clearOpenAIClientCache() lastChatCompletionsArgs = null chatCompletionsUsage = { prompt_tokens: 3, completion_tokens: 2 } capturedFetch = null @@ -234,7 +252,7 @@ describe('sideQuery OpenAI ChatGPT OAuth path', () => { max_tokens: 256, }) - expect(getOpenAIClientCallCount).toBe(0) + expect(openaiClientConstructCount).toBe(0) expect(chatCompletionsCreateCount).toBe(0) expect(capturedFetch).not.toBeNull() expect(capturedFetch!.url).toContain( @@ -291,7 +309,7 @@ describe('sideQuery OpenAI ChatGPT OAuth path', () => { tool_choice: { type: 'tool', name: 'classify_result' }, }) - expect(getOpenAIClientCallCount).toBe(1) + expect(openaiClientConstructCount).toBe(1) expect(chatCompletionsCreateCount).toBe(1) expect(capturedFetch).toBeNull() expect(lastChatCompletionsArgs?.model).toBe('gpt-4o') @@ -353,6 +371,6 @@ describe('sideQuery OpenAI ChatGPT OAuth path', () => { tool_choice: { type: 'tool', name: 'classify_result' }, }), ).rejects.toThrow(/ChatGPT Responses API request failed \(401\)/) - expect(getOpenAIClientCallCount).toBe(0) + expect(openaiClientConstructCount).toBe(0) }) })