fix(kosong): retry a dropped provider stream (terminated)

.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.

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)/,

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)}`);
 }

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`.

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);
+  });
 });