diff --git a/.changeset/retry-terminated-stream.md b/.changeset/retry-terminated-stream.md new file mode 100644 index 000000000..342714e96 --- /dev/null +++ b/.changeset/retry-terminated-stream.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/kosong": patch +"@moonshot-ai/kimi-code": patch +--- + +Automatically retry when a model response stream is dropped mid-flight (a `terminated` error) instead of failing the turn. diff --git a/packages/agent-core/test/loop/retry.test.ts b/packages/agent-core/test/loop/retry.test.ts new file mode 100644 index 000000000..47442cf12 --- /dev/null +++ b/packages/agent-core/test/loop/retry.test.ts @@ -0,0 +1,70 @@ +import { APIConnectionError, emptyUsage, isRetryableGenerateError } from '@moonshot-ai/kosong'; +import { describe, expect, it } from 'vitest'; + +import type { LLM, LLMChatParams, LLMChatResponse } from '#/loop/llm'; +import { chatWithRetry } from '#/loop/retry'; + +function okResponse(): LLMChatResponse { + return { toolCalls: [], usage: emptyUsage() }; +} + +function makeInput( + llm: LLM, + signal: AbortSignal, +): Parameters[0] { + return { + llm, + params: { messages: [], tools: [], signal }, + dispatchEvent: async () => {}, + turnId: 't', + currentStep: 1, + stepUuid: 'u', + }; +} + +describe('chatWithRetry: terminated stream drops', () => { + it('retries an APIConnectionError("terminated") and succeeds on a later attempt', async () => { + // A mid-stream `terminated` is classified as a retryable APIConnectionError, + // so an intermittent connection drop should be recovered transparently. + let calls = 0; + const llm: LLM = { + systemPrompt: '', + modelName: 'mock', + isRetryableError: (e) => isRetryableGenerateError(e), + async chat(_params: LLMChatParams): Promise { + calls += 1; + if (calls === 1) throw new APIConnectionError('terminated'); + return okResponse(); + }, + }; + + const response = await chatWithRetry(makeInput(llm, new AbortController().signal)); + + expect(calls).toBe(2); + expect(response).toEqual(okResponse()); + }); + + it('does NOT retry when the signal is aborted (user ESC), surfacing a clean AbortError', async () => { + // Even though `terminated` is retryable, a user-aborted request must never + // be retried: the abort signal is checked before any retry, so it surfaces + // as an AbortError rather than a provider error. + let calls = 0; + const ac = new AbortController(); + ac.abort(); + + const llm: LLM = { + systemPrompt: '', + modelName: 'mock', + isRetryableError: (e) => isRetryableGenerateError(e), + async chat(_params: LLMChatParams): Promise { + calls += 1; + throw new APIConnectionError('terminated'); + }, + }; + + await expect(chatWithRetry(makeInput(llm, ac.signal))).rejects.toMatchObject({ + name: 'AbortError', + }); + expect(calls).toBe(1); + }); +}); diff --git a/packages/kosong/src/providers/openai-common.ts b/packages/kosong/src/providers/openai-common.ts index 120490c24..10fb42726 100644 --- a/packages/kosong/src/providers/openai-common.ts +++ b/packages/kosong/src/providers/openai-common.ts @@ -84,7 +84,11 @@ export function toolToOpenAI(tool: Tool): OpenAIToolParam { }, }; } -const NETWORK_RE = /network|connection|connect|disconnect/i; +// `terminated` is the undici signature for an SSE/HTTP body stream that is +// dropped mid-flight (common with Node's native fetch on long reasoning +// streams). It surfaces as a raw `TypeError: terminated`, so it must be +// recognized here as a transport-layer connection failure. +const NETWORK_RE = /network|connection|connect|disconnect|terminated/i; const TIMEOUT_RE = /timed?\s*out|timeout|deadline/i; function classifyBaseApiError(message: string): ChatProviderError { @@ -129,8 +133,13 @@ export function convertOpenAIError(error: unknown): ChatProviderError { if (error instanceof OpenAIError) { return new ChatProviderError(`Error: ${error.message}`); } + // Raw, non-SDK errors (e.g. undici's `TypeError: terminated` raised when a + // streaming response body is dropped mid-flight) never get wrapped by the + // OpenAI SDK during stream iteration. Route them through the same + // transport-layer heuristic so genuine connection failures become + // retryable instead of fatal generic errors. if (error instanceof Error) { - return new ChatProviderError(`Error: ${error.message}`); + return classifyBaseApiError(error.message); } return new ChatProviderError(`Error: ${String(error)}`); } diff --git a/packages/kosong/test/openai-common-errors.test.ts b/packages/kosong/test/openai-common-errors.test.ts index 3d07a52e2..fd9d9c5f2 100644 --- a/packages/kosong/test/openai-common-errors.test.ts +++ b/packages/kosong/test/openai-common-errors.test.ts @@ -4,6 +4,7 @@ import { APIStatusError, APITimeoutError, ChatProviderError, + isRetryableGenerateError, } from '#/errors'; import type { ContentPart } from '#/message'; import { @@ -186,6 +187,58 @@ describe('OpenAI streaming error propagation', () => { }).rejects.toThrow(/Network connection lost/); }); }); +describe('convertOpenAIError: raw transport-layer stream errors', () => { + it('classifies undici TypeError("terminated") as a retryable APIConnectionError', () => { + // Node v24 + undici raises a raw `TypeError: terminated` when an SSE + // response stream is dropped mid-flight. It is NOT an OpenAI SDK error, + // so it falls into the generic Error branch — but it is a transport-layer + // connection failure and must be retryable like any dropped connection. + const err = new TypeError('terminated'); + (err as { cause?: unknown }).cause = new Error('other side closed'); + + const result = convertOpenAIError(err); + + expect(result).toBeInstanceOf(APIConnectionError); + expect(isRetryableGenerateError(result)).toBe(true); + }); + + it('still wraps an unrelated raw Error as a non-retryable ChatProviderError', () => { + const result = convertOpenAIError(new Error('something completely unrelated')); + + expect(result.constructor).toBe(ChatProviderError); + expect(isRetryableGenerateError(result)).toBe(false); + }); +}); +describe('OpenAI streaming: undici terminated mid-stream', () => { + it('a stream that throws TypeError("terminated") rejects with retryable APIConnectionError', async () => { + // Simulates the real-world failure: the SSE stream drops mid-flight and + // undici raises a raw `TypeError: terminated` from inside the for-await + // loop. The provider must surface a retryable APIConnectionError so the + // loop retries instead of failing the turn outright. + async function* terminatedStream(): AsyncGenerator { + throw new TypeError('terminated'); + yield undefined as never; + } + + const msg = new OpenAILegacyStreamedMessage( + terminatedStream() as AsyncIterable, + true, + undefined, + ); + + let caught: unknown; + try { + for await (const _ of msg) { + void _; + } + } catch (error) { + caught = error; + } + + expect(caught).toBeInstanceOf(APIConnectionError); + expect(isRetryableGenerateError(caught)).toBe(true); + }); +}); describe('convertContentPart', () => { it('converts TextPart to OpenAI text content part', () => { expect(convertContentPart({ type: 'text', text: 'hi' })).toEqual({