fix: time out stalled LLM streams

.changeset/fix-stalled-llm-streams.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Fail and retry LLM streams that stop producing data instead of waiting indefinitely.

packages/agent-core-v2/src/kosong/contract/errors.ts

@@ -56,6 +56,30 @@ export class APITimeoutError extends ChatProviderError {
   }
 }
 
+export class StreamIdleTimeoutError extends APITimeoutError {
+  readonly idleTimeoutMs: number;
+  readonly elapsedMs: number;
+  readonly traceId: string | null;
+
+  constructor(
+    providerName: string,
+    modelName: string,
+    idleTimeoutMs: number,
+    elapsedMs: number,
+    traceId: string | null,
+  ) {
+    const traceHint = traceId === null ? '' : `, traceId: ${traceId}`;
+    super(
+      `LLM stream stalled for ${idleTimeoutMs}ms ` +
+        `(provider: ${providerName}, model: ${modelName}, elapsedMs: ${elapsedMs}${traceHint}).`,
+    );
+    this.name = 'StreamIdleTimeoutError';
+    this.idleTimeoutMs = idleTimeoutMs;
+    this.elapsedMs = elapsedMs;
+    this.traceId = traceId;
+  }
+}
+
 export class APIStatusError extends ChatProviderError {
   readonly statusCode: number;
   readonly requestId: string | null;

packages/agent-core-v2/src/kosong/contract/generate.ts

@@ -4,12 +4,11 @@
  * `generate()` is the single place that orchestrates "call
  * `ChatProvider.generate` and normalize the event stream": it merges streamed
  * deltas into a complete assistant `Message`, fires the caller's callbacks,
- * enforces the abort contract (standard abort DOMException, stream cancelled
- * on abort), and rejects empty or thinking-only responses with
- * `APIEmptyResponseError`.
+ * enforces the abort and stream-idle contracts, and rejects empty or
+ * thinking-only responses with `APIEmptyResponseError`.
  */
 
-import { APIEmptyResponseError, createAbortError } from './errors';
+import { APIEmptyResponseError, createAbortError, StreamIdleTimeoutError } from './errors';
 import {
   isContentPart,
   isToolCall,
@@ -24,6 +23,7 @@ import type { Tool } from './tool';
 import type { TokenUsage } from './usage';
 
 type StoredToolCall = Omit<ToolCall, '_streamIndex'>;
+const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 180_000;
 
 export interface GenerateResult {
   readonly id: string | null;
@@ -60,8 +60,17 @@ export async function generate(
     ? tools.filter((tool) => tool.deferred !== true)
     : tools;
 
+  const idleTimeoutMs = resolveStreamIdleTimeoutMs(options?.streamIdleTimeoutMs);
+  const watchdog = new AbortController();
+  const providerSignal =
+    options?.signal === undefined
+      ? watchdog.signal
+      : AbortSignal.any([options.signal, watchdog.signal]);
+  const providerOptions: GenerateOptions = { ...options, signal: providerSignal };
+  delete providerOptions.streamIdleTimeoutMs;
+
   options?.onRequestStart?.();
-  const stream = await provider.generate(systemPrompt, wireTools, history, options);
+  const stream = await provider.generate(systemPrompt, wireTools, history, providerOptions);
   if (stream.traceId !== undefined) {
     options?.onTraceId?.(stream.traceId);
   }
@@ -73,7 +82,13 @@ export async function generate(
   let firstPartAt: number | undefined;
   let lastResumeAt = 0;
 
-  for await (const part of stream) {
+  for await (const part of withStreamIdleTimeout(
+    stream,
+    provider,
+    watchdog,
+    options?.signal,
+    idleTimeoutMs,
+  )) {
     const arrivedAt = Date.now();
     if (firstPartAt === undefined) {
       firstPartAt = arrivedAt;
@@ -186,16 +201,99 @@ type CancelableStream = StreamedMessage & {
   return?: () => unknown;
 };
 
-async function cancelStream(stream: StreamedMessage): Promise<void> {
-  const cancelable = stream as CancelableStream;
-
+function settleWithoutWaiting(action: () => unknown): void {
   try {
-    await cancelable.cancel?.();
+    void Promise.resolve(action()).catch(() => {});
   } catch {}
+}
+
+function abandonStream(stream: StreamedMessage): void {
+  const cancelable = stream as CancelableStream;
+  settleWithoutWaiting(() => cancelable.cancel?.());
+  settleWithoutWaiting(() => cancelable.return?.());
+}
+
+function resolveStreamIdleTimeoutMs(value?: number): number {
+  return value !== undefined && Number.isFinite(value) && value > 0
+    ? value
+    : DEFAULT_STREAM_IDLE_TIMEOUT_MS;
+}
+
+async function* withStreamIdleTimeout(
+  stream: StreamedMessage,
+  provider: ChatProvider,
+  watchdog: AbortController,
+  callerSignal: AbortSignal | undefined,
+  idleTimeoutMs: number,
+): AsyncGenerator<StreamedMessagePart> {
+  const iterator = stream[Symbol.asyncIterator]();
+  const startedAt = Date.now();
+  let completed = false;
 
   try {
-    await cancelable.return?.();
-  } catch {}
+    while (true) {
+      const next = iterator.next();
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      let onCallerAbort: (() => void) | undefined;
+      const timeout = new Promise<never>((_, reject) => {
+        timer = setTimeout(() => {
+          reject(
+            new StreamIdleTimeoutError(
+              provider.name,
+              provider.modelName,
+              idleTimeoutMs,
+              Date.now() - startedAt,
+              stream.traceId ?? null,
+            ),
+          );
+        }, idleTimeoutMs);
+      });
+      const callerAbort =
+        callerSignal === undefined
+          ? undefined
+          : new Promise<never>((_, reject) => {
+              onCallerAbort = () => reject(createAbortError());
+              if (callerSignal.aborted) {
+                onCallerAbort();
+              } else {
+                callerSignal.addEventListener('abort', onCallerAbort, { once: true });
+              }
+            });
+
+      try {
+        const result =
+          callerAbort === undefined
+            ? await Promise.race([next, timeout])
+            : await Promise.race([next, timeout, callerAbort]);
+        if (result.done === true) {
+          completed = true;
+          return;
+        }
+        yield result.value;
+      } catch (error) {
+        if (callerSignal?.aborted) {
+          next.catch(() => {});
+          abandonStream(stream);
+          throw createAbortError();
+        }
+        if (error instanceof StreamIdleTimeoutError) {
+          next.catch(() => {});
+          watchdog.abort(error);
+          abandonStream(stream);
+        }
+        throw error;
+      } finally {
+        clearTimeout(timer);
+        if (onCallerAbort !== undefined) {
+          callerSignal?.removeEventListener('abort', onCallerAbort);
+        }
+      }
+    }
+  } finally {
+    if (!completed) {
+      settleWithoutWaiting(() => iterator.return?.());
+    }
+  }
 }
 
 async function throwIfAborted(signal?: AbortSignal, stream?: StreamedMessage): Promise<void> {
@@ -204,7 +302,7 @@ async function throwIfAborted(signal?: AbortSignal, stream?: StreamedMessage): P
   }
 
   if (stream !== undefined) {
-    await cancelStream(stream);
+    abandonStream(stream);
   }
 
   throw createAbortError();

packages/agent-core-v2/src/kosong/contract/provider.ts

@@ -9,9 +9,9 @@
  *    `with*` methods; every per-turn intent (prompt-cache key, sampling
  *    overrides, thinking effort/keep, completion-token budget) flows through
  *    `GenerateOptions` on each `generate` call instead of through morphs.
- *  - `GenerateOptions` is the per-turn intent carrier. Each wire dialect
- *    decides how — or whether — to encode an intent (e.g. a cache key may
- *    become `prompt_cache_key`, `metadata.user_id`, or be silently dropped).
+ *  - `GenerateOptions` is the per-turn intent carrier and stream-control
+ *    contract. Each wire dialect decides how — or whether — to encode an
+ *    intent; the generation driver owns the stream idle deadline.
  *
  * Pure types only — no other domain, no I/O, no SDKs.
  */
@@ -126,6 +126,7 @@ export interface VideoUploadInput {
  */
 export interface GenerateOptions {
   signal?: AbortSignal;
+  streamIdleTimeoutMs?: number;
   auth?: ProviderRequestAuth;
   responseFormat?: ResponseFormat;
   /**