fix(kosong): retry when a response stream is terminated mid-flight

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

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<typeof chatWithRetry>[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<LLMChatResponse> {
+        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<LLMChatResponse> {
+        calls += 1;
+        throw new APIConnectionError('terminated');
+      },
+    };
+
+    await expect(chatWithRetry(makeInput(llm, ac.signal))).rejects.toMatchObject({
+      name: 'AbortError',
+    });
+    expect(calls).toBe(1);
+  });
+});

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

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<never> {
+      throw new TypeError('terminated');
+      yield undefined as never;
+    }
+
+    const msg = new OpenAILegacyStreamedMessage(
+      terminatedStream() as AsyncIterable<never>,
+      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({