From 074bb9ba1359dd3ea2a55eff81986f2bb4772793 Mon Sep 17 00:00:00 2001 From: Kai Date: Wed, 1 Jul 2026 22:48:57 +0800 Subject: [PATCH] fix(kosong): retry a dropped provider stream (terminated) on the Anthropic path (#1274) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A raw undici `terminated` error — an SSE/HTTP response body cut mid-flight, common on long streaming responses — fell through convertAnthropicError to a generic base ChatProviderError, which isRetryableGenerateError treats as fatal, so it was never retried. Route raw non-SDK errors through the shared classifyBaseApiError heuristic (already used by the OpenAI path) so a dropped stream is classified as a retryable APIConnectionError and retried instead of failing the turn. --- .changeset/session-recovery-terminated.md | 5 ++ packages/kosong/src/errors.ts | 27 ++++++++++ packages/kosong/src/providers/anthropic.ts | 8 ++- .../kosong/src/providers/openai-common.ts | 17 +------ packages/kosong/test/anthropic-errors.test.ts | 50 +++++++++++++++++++ 5 files changed, 90 insertions(+), 17 deletions(-) create mode 100644 .changeset/session-recovery-terminated.md diff --git a/.changeset/session-recovery-terminated.md b/.changeset/session-recovery-terminated.md new file mode 100644 index 000000000..3f4571d2d --- /dev/null +++ b/.changeset/session-recovery-terminated.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kosong": patch +--- + +Retry a dropped provider stream instead of failing the turn. A raw undici `terminated` error (an SSE/HTTP response body cut mid-flight, common on long streaming responses) is now classified as a retryable `APIConnectionError` on the Anthropic path — matching the OpenAI path, which already recognized it — so a transient stream drop is retried rather than surfaced as a fatal error. diff --git a/packages/kosong/src/errors.ts b/packages/kosong/src/errors.ts index a2a8cdc4a..49bbebaae 100644 --- a/packages/kosong/src/errors.ts +++ b/packages/kosong/src/errors.ts @@ -98,6 +98,33 @@ export function isRetryableGenerateError(error: unknown): boolean { return error instanceof APIStatusError && [429, 500, 502, 503, 504].includes(error.statusCode); } +// `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. Shared by the +// Anthropic and OpenAI providers so a raw, non-SDK transport error classifies +// the same way regardless of which provider was streaming. +const NETWORK_RE = /network|connection|connect|disconnect|terminated/i; +const TIMEOUT_RE = /timed?\s*out|timeout|deadline/i; + +/** + * Classify a raw (non-SDK) error message into the right transport-layer + * `ChatProviderError` subclass: a timeout becomes a retryable `APITimeoutError`, + * a dropped connection / undici `terminated` becomes a retryable + * `APIConnectionError`, and anything else stays a non-retryable base + * `ChatProviderError`. Timeout is checked first so "connection timed out" + * classifies as a timeout rather than a bare connection error. + */ +export function classifyBaseApiError(message: string): ChatProviderError { + if (TIMEOUT_RE.test(message)) { + return new APITimeoutError(message); + } + if (NETWORK_RE.test(message)) { + return new APIConnectionError(message); + } + return new ChatProviderError(`Error: ${message}`); +} + const CONTEXT_OVERFLOW_MESSAGE_PATTERNS = [ /context[ _-]?length/, /(?:context[ _-]?window.*exceed|exceed.*context[ _-]?window)/, diff --git a/packages/kosong/src/providers/anthropic.ts b/packages/kosong/src/providers/anthropic.ts index 72f243941..a031b75ee 100644 --- a/packages/kosong/src/providers/anthropic.ts +++ b/packages/kosong/src/providers/anthropic.ts @@ -2,6 +2,7 @@ import { APIConnectionError, APITimeoutError, ChatProviderError, + classifyBaseApiError, normalizeAPIStatusError, } from '#/errors'; import type { ContentPart, Message, StreamedMessagePart, ToolCall } from '#/message'; @@ -652,8 +653,13 @@ export function convertAnthropicError(error: unknown): ChatProviderError { if (error instanceof AnthropicError) { return new ChatProviderError(`Anthropic error: ${error.message}`); } + // Raw, non-SDK errors (e.g. undici's `TypeError: terminated` raised when a + // streaming response body is dropped mid-flight) are never wrapped by the + // Anthropic SDK during stream iteration. Route them through the shared + // 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/src/providers/openai-common.ts b/packages/kosong/src/providers/openai-common.ts index caba437c0..b83377e4e 100644 --- a/packages/kosong/src/providers/openai-common.ts +++ b/packages/kosong/src/providers/openai-common.ts @@ -2,6 +2,7 @@ import { APIConnectionError, APITimeoutError, ChatProviderError, + classifyBaseApiError, normalizeAPIStatusError, } from '#/errors'; import { extractText } from '#/message'; @@ -84,22 +85,6 @@ export function toolToOpenAI(tool: Tool): OpenAIToolParam { }, }; } -// `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 { - if (TIMEOUT_RE.test(message)) { - return new APITimeoutError(message); - } - if (NETWORK_RE.test(message)) { - return new APIConnectionError(message); - } - return new ChatProviderError(`Error: ${message}`); -} /** * Convert an OpenAI SDK error (or raw Error) to a kosong `ChatProviderError`. diff --git a/packages/kosong/test/anthropic-errors.test.ts b/packages/kosong/test/anthropic-errors.test.ts index e39444346..700ba00ff 100644 --- a/packages/kosong/test/anthropic-errors.test.ts +++ b/packages/kosong/test/anthropic-errors.test.ts @@ -5,6 +5,7 @@ import { APIStatusError, APITimeoutError, ChatProviderError, + isRetryableGenerateError, } from '#/errors'; import { convertAnthropicError, AnthropicChatProvider } from '#/providers/anthropic'; import { @@ -104,6 +105,27 @@ describe('convertAnthropicError', () => { expect(result).toBeInstanceOf(ChatProviderError); expect(result.message).toContain('string error'); }); + + 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 Anthropic 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 = convertAnthropicError(err); + + expect(result).toBeInstanceOf(APIConnectionError); + expect(isRetryableGenerateError(result)).toBe(true); + }); + + it('still wraps an unrelated raw Error as a non-retryable ChatProviderError', () => { + const result = convertAnthropicError(new Error('something completely unrelated')); + + expect(result.constructor).toBe(ChatProviderError); + expect(isRetryableGenerateError(result)).toBe(false); + }); }); describe('non-stream error propagation', () => { function createNonStreamProvider(): AnthropicChatProvider { @@ -354,4 +376,32 @@ describe('stream error propagation', () => { expect((error as APIStatusError).statusCode).toBe(401); } }); + + it('undici TypeError("terminated") during stream iteration -> retryable APIConnectionError', async () => { + // 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. + const provider = createStreamProvider(); + (provider as any)._client.messages.create = vi + .fn() + .mockResolvedValue(makeErrorStream(new TypeError('terminated'))) as never; + + const result = await provider.generate( + '', + [], + [{ role: 'user', content: [{ type: 'text', text: 'Hi' }], toolCalls: [] }], + ); + let caught: unknown; + try { + for await (const _ of result) { + void _; + } + } catch (error) { + caught = error; + } + + expect(caught).toBeInstanceOf(APIConnectionError); + expect(isRetryableGenerateError(caught)).toBe(true); + }); });