From a3f3e3744b234580b198d4af12eda7497206f581 Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Mon, 24 Aug 2026 22:01:36 +0800
Subject: [PATCH 01/10] fix(core): guard Anthropic streams with idle and
 lifetime watchdogs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The OpenAI wire wraps its stream in an inactivity watchdog plus a
non-resetting lifetime cap, while the Anthropic wire had neither: a
stream that returns 200 and then goes silent — or drip-feeds
low-content thinking_delta frames that keep resetting any idle-only
timer — hangs the CLI until the process is killed. Wire the same
guards into the Anthropic generator so a stalled stream aborts with a
retryable ETIMEDOUT instead of hanging, and the two wires stop
differing on whether a stalled stream is recoverable.

The guard mechanics, error classes, and timeout resolvers move to a
shared stream-guards module used by both wires; the OpenAI pipeline
re-exports the error classes so existing imports keep working. Also
rename six stale popPartialIfPushed comment references in
geminiChat.ts to the real popPendingPartialAssistantTurn method.

Issue #9005 findings 4 and 6.
---
 packages/cli/src/config/settingsSchema.ts     |   2 +-
 .../anthropicContentGenerator.test.ts         | 327 ++++++++++++++++++
 .../anthropicContentGenerator.ts              |  42 ++-
 packages/core/src/core/contentGenerator.ts    |   7 +-
 packages/core/src/core/geminiChat.ts          |  12 +-
 .../core/openaiContentGenerator/pipeline.ts   | 282 +--------------
 packages/core/src/core/stream-guards.ts       | 291 ++++++++++++++++
 7 files changed, 685 insertions(+), 278 deletions(-)
 create mode 100644 packages/core/src/core/stream-guards.ts

diff --git a/packages/cli/src/config/settingsSchema.ts b/packages/cli/src/config/settingsSchema.ts
index 7d8ffb67329..308bfdf4d53 100644
--- a/packages/cli/src/config/settingsSchema.ts
+++ b/packages/cli/src/config/settingsSchema.ts
@@ -1735,7 +1735,7 @@ const SETTINGS_SCHEMA = {
             requiresRestart: false,
             default: undefined as number | undefined,
             description:
-              'Maximum inactivity between streamed chunks for OpenAI-compatible models, in milliseconds. Set to 0 to disable the idle guard. For provider-backed models, configure this field in the selected modelProviders entry.',
+              'Maximum inactivity between streamed chunks for OpenAI-compatible and Anthropic models, in milliseconds. Set to 0 to disable the idle guard. For provider-backed models, configure this field in the selected modelProviders entry.',
             minimum: 0,
             maximum: 2_147_483_647,
             parentKey: 'generationConfig',
diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
index 6f8408be1d4..34de875c2dd 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
@@ -10,8 +10,11 @@ import type { GenerateContentParameters } from '@google/genai';
 import { FinishReason, GenerateContentResponse } from '@google/genai';
 import type { ContentGeneratorConfig } from '../contentGenerator.js';
 import {
+  DEFAULT_STREAM_IDLE_TIMEOUT_MS,
   DEFAULT_TIMEOUT,
   DISABLED_REQUEST_TIMEOUT_MS,
+  QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
+  QWEN_STREAM_MAX_LIFETIME_MS_ENV,
 } from '../openaiContentGenerator/constants.js';
 
 const mockReportAnthropicRequest = vi.hoisted(() => vi.fn());
@@ -5001,6 +5004,330 @@ describe('AnthropicContentGenerator', () => {
     });
   });
 
+  // Issue #9005 finding 4: the OpenAI wire wraps its stream in an idle
+  // watchdog plus a non-resetting lifetime cap (openaiContentGenerator
+  // `withStreamGuards`), while the Anthropic wire had neither — a stream that
+  // returns 200 and then goes silent (or drip-feeds `thinking_delta` frames
+  // forever, which keep resetting any idle-only timer) hangs the CLI until
+  // the process is killed. These tests pin the Anthropic wire to the same
+  // guards, mirroring the OpenAI pipeline's watchdog suite.
+  describe('stream watchdog guards (issue #9005 finding 4)', () => {
+    // A manually gated SSE event source: events arrive only when pushed, so
+    // tests can model a stream that goes silent or is drip-fed. Mirrors the
+    // `gatedStream` helper in openaiContentGenerator/pipeline.test.ts.
+    function gatedEventStream() {
+      let resolveNext: ((r: IteratorResult<unknown>) => void) | null = null;
+      const buffered: unknown[] = [];
+      let ended = false;
+      let returned = false;
+      const deliver = (r: IteratorResult<unknown>) => {
+        const r2 = resolveNext;
+        resolveNext = null;
+        r2?.(r);
+      };
+      return {
+        push(event: unknown) {
+          if (resolveNext) deliver({ done: false, value: event });
+          else buffered.push(event);
+        },
+        end() {
+          ended = true;
+          if (resolveNext) deliver({ done: true, value: undefined as never });
+        },
+        wasReturned() {
+          return returned;
+        },
+        stream: {
+          [Symbol.asyncIterator]() {
+            return {
+              next(): Promise<IteratorResult<unknown>> {
+                if (buffered.length) {
+                  return Promise.resolve({
+                    done: false,
+                    value: buffered.shift()!,
+                  });
+                }
+                if (ended) {
+                  return Promise.resolve({
+                    done: true,
+                    value: undefined as never,
+                  });
+                }
+                return new Promise((res) => {
+                  resolveNext = res;
+                });
+              },
+              return(): Promise<IteratorResult<unknown>> {
+                returned = true;
+                ended = true;
+                if (resolveNext) {
+                  deliver({ done: true, value: undefined as never });
+                }
+                return Promise.resolve({
+                  done: true,
+                  value: undefined as never,
+                });
+              },
+            };
+          },
+        },
+      };
+    }
+
+    const buildGenerator = async (guardConfig: {
+      streamIdleTimeoutMs?: number;
+      streamMaxLifetimeMs?: number;
+    }) => {
+      const { AnthropicContentGenerator } = await importGenerator();
+      return new AnthropicContentGenerator(
+        {
+          model: 'claude-test',
+          apiKey: 'test-key',
+          timeout: 10_000,
+          maxRetries: 2,
+          samplingParams: { max_tokens: 100 },
+          schemaCompliance: 'auto',
+          ...(guardConfig.streamIdleTimeoutMs !== undefined
+            ? { streamIdleTimeoutMs: guardConfig.streamIdleTimeoutMs }
+            : {}),
+          ...(guardConfig.streamMaxLifetimeMs !== undefined
+            ? { streamMaxLifetimeMs: guardConfig.streamMaxLifetimeMs }
+            : {}),
+        },
+        mockConfig,
+      );
+    };
+
+    const streamRequest = {
+      model: 'models/ignored',
+      contents: 'Hello',
+    } as unknown as GenerateContentParameters;
+
+    // Drain the stream and capture whatever it ends with. Without the guards a
+    // silent/drip-fed source never settles, so callers race the result
+    // against a sentinel instead of awaiting it directly — a missing
+    // watchdog then fails the assertion instead of hanging the test.
+    const consumeUntilSettled = (
+      stream: AsyncGenerator<GenerateContentResponse>,
+    ) =>
+      (async () => {
+        for await (const _chunk of stream) {
+          /* drain */
+        }
+      })().catch((e: unknown) => e);
+
+    beforeEach(() => {
+      // Ignore ambient QWEN_STREAM_* knobs from the dev/CI shell so the
+      // explicit-config tests aren't silently overridden.
+      vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, undefined);
+      vi.stubEnv(QWEN_STREAM_MAX_LIFETIME_MS_ENV, undefined);
+      vi.useFakeTimers();
+    });
+
+    afterEach(() => {
+      vi.useRealTimers();
+      vi.unstubAllEnvs();
+    });
+
+    it('aborts and throws a retryable ETIMEDOUT when the stream is silent past the idle timeout', async () => {
+      const gated = gatedEventStream(); // never pushes → silent
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({
+        streamIdleTimeoutMs: 1000,
+        streamMaxLifetimeMs: 0,
+      });
+      const stream = await generator.generateContentStream(streamRequest);
+      const captured = consumeUntilSettled(stream);
+      await vi.advanceTimersByTimeAsync(1000);
+      await vi.advanceTimersByTimeAsync(0);
+      const sentinel = Symbol('idle-watchdog-did-not-fire');
+      const err = await Promise.race([captured, Promise.resolve(sentinel)]);
+      expect(err).toMatchObject({
+        name: 'StreamInactivityTimeoutError',
+        code: 'ETIMEDOUT',
+        idleMs: 1000,
+        chunksReceived: 0,
+      });
+      expect((err as Error).message).toContain('QWEN_STREAM_IDLE_TIMEOUT_MS');
+      expect(gated.wasReturned()).toBe(true);
+    });
+
+    it('uses the shared default idle timeout when no override is configured', async () => {
+      const gated = gatedEventStream(); // never pushes → silent
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({});
+      const stream = await generator.generateContentStream(streamRequest);
+      const captured = consumeUntilSettled(stream);
+      await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS);
+      await vi.advanceTimersByTimeAsync(0);
+      const sentinel = Symbol('default-idle-watchdog-did-not-fire');
+      const err = await Promise.race([captured, Promise.resolve(sentinel)]);
+      expect(err).toMatchObject({
+        name: 'StreamInactivityTimeoutError',
+        code: 'ETIMEDOUT',
+        idleMs: DEFAULT_STREAM_IDLE_TIMEOUT_MS,
+        chunksReceived: 0,
+      });
+    });
+
+    it('does not interrupt a stream whose events keep arriving inside the idle window', async () => {
+      const gated = gatedEventStream();
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({
+        streamIdleTimeoutMs: 1000,
+        streamMaxLifetimeMs: 0,
+      });
+      const stream = await generator.generateContentStream(streamRequest);
+      let done = false;
+      let error: unknown;
+      const texts: string[] = [];
+      const consume = (async () => {
+        for await (const chunk of stream) {
+          for (const candidate of chunk.candidates ?? []) {
+            for (const part of candidate.content?.parts ?? []) {
+              if (part.text) texts.push(part.text);
+            }
+          }
+        }
+      })().then(
+        () => (done = true),
+        (e: unknown) => (error = e),
+      );
+      gated.push({
+        type: 'message_start',
+        message: {
+          id: 'msg-1',
+          model: 'claude-test',
+          usage: { input_tokens: 1 },
+        },
+      });
+      gated.push({
+        type: 'content_block_start',
+        index: 0,
+        content_block: { type: 'text', text: '' },
+      });
+      gated.push({
+        type: 'content_block_delta',
+        index: 0,
+        delta: { type: 'text_delta', text: 'hel' },
+      });
+      await vi.advanceTimersByTimeAsync(500); // < 1000ms idle window
+      gated.push({
+        type: 'content_block_delta',
+        index: 0,
+        delta: { type: 'text_delta', text: 'lo' },
+      });
+      await vi.advanceTimersByTimeAsync(500);
+      gated.push({ type: 'content_block_stop', index: 0 });
+      gated.push({
+        type: 'message_delta',
+        delta: { stop_reason: 'end_turn' },
+        usage: { output_tokens: 1 },
+      });
+      gated.push({ type: 'message_stop' });
+      gated.end();
+      await vi.advanceTimersByTimeAsync(0);
+      await consume;
+      expect(error).toBeUndefined();
+      expect(done).toBe(true);
+      expect(texts).toEqual(['hel', 'lo']);
+    });
+
+    it('caps total stream lifetime when thinking deltas keep resetting the idle watchdog', async () => {
+      // The issue #9005 finding-4 shape: adaptive thinking emits long runs of
+      // `thinking_delta` frames, each resetting an idle-only timer, while the
+      // message never completes (the #8597 drip-fed hang). The lifetime cap
+      // does not reset.
+      const gated = gatedEventStream(); // drip-fed, never ends
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({
+        streamIdleTimeoutMs: 1000,
+        streamMaxLifetimeMs: 3000,
+      });
+      const stream = await generator.generateContentStream(streamRequest);
+      const captured = consumeUntilSettled(stream);
+      gated.push({
+        type: 'message_start',
+        message: {
+          id: 'msg-1',
+          model: 'claude-test',
+          usage: { input_tokens: 1 },
+        },
+      });
+      await vi.advanceTimersByTimeAsync(500);
+      gated.push({
+        type: 'content_block_start',
+        index: 0,
+        content_block: { type: 'thinking', thinking: '' },
+      });
+      await vi.advanceTimersByTimeAsync(500);
+      for (let i = 0; i < 4; i++) {
+        gated.push({
+          type: 'content_block_delta',
+          index: 0,
+          delta: { type: 'thinking_delta', thinking: 't' },
+        });
+        await vi.advanceTimersByTimeAsync(500); // each drip resets the 1s idle watchdog
+      }
+      await vi.advanceTimersByTimeAsync(1000); // now past the 3000ms cap
+      await vi.advanceTimersByTimeAsync(0);
+      const sentinel = Symbol('lifetime-cap-did-not-fire');
+      const err = await Promise.race([captured, Promise.resolve(sentinel)]);
+      expect(err).toMatchObject({
+        name: 'StreamLifetimeExceededError',
+        code: 'ETIMEDOUT',
+        maxLifetimeMs: 3000,
+        chunksReceived: 6,
+      });
+      expect((err as Error).message).toContain('QWEN_STREAM_MAX_LIFETIME_MS');
+      expect(gated.wasReturned()).toBe(true);
+    });
+
+    it('leaves streams unguarded when both timeouts are disabled (<= 0)', async () => {
+      const gated = gatedEventStream();
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({
+        streamIdleTimeoutMs: 0,
+        streamMaxLifetimeMs: 0,
+      });
+      const stream = await generator.generateContentStream(streamRequest);
+      let done = false;
+      let error: unknown;
+      const consume = (async () => {
+        for await (const _chunk of stream) {
+          /* drain */
+        }
+      })().then(
+        () => (done = true),
+        (e: unknown) => (error = e),
+      );
+      gated.push({
+        type: 'content_block_start',
+        index: 0,
+        content_block: { type: 'text', text: '' },
+      });
+      // A silence far beyond the default idle timeout — survivable only
+      // when the guards are explicitly disabled.
+      await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS + 1000);
+      gated.push({
+        type: 'content_block_delta',
+        index: 0,
+        delta: { type: 'text_delta', text: 'still here' },
+      });
+      gated.push({ type: 'content_block_stop', index: 0 });
+      gated.push({
+        type: 'message_delta',
+        delta: { stop_reason: 'end_turn' },
+        usage: { output_tokens: 1 },
+      });
+      gated.end();
+      await vi.advanceTimersByTimeAsync(0);
+      await consume;
+      expect(error).toBeUndefined();
+      expect(done).toBe(true);
+    });
+  });
+
   describe('tool_choice mapping from Gemini toolConfig', () => {
     async function sendWithToolConfig(
       mode: string | undefined,
diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts
index 1c6d0887795..da41ffdb32a 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts
@@ -34,6 +34,11 @@ import {
   redactProxyError,
 } from '../../utils/runtimeFetchOptions.js';
 import { resolveRequestTimeout } from '../openaiContentGenerator/constants.js';
+import {
+  resolveStreamIdleTimeoutMs,
+  resolveStreamMaxLifetimeMs,
+  withStreamGuards,
+} from '../stream-guards.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 import { runtimeDiagnostics } from '../../utils/runtimeDiagnostics.js';
 import { createChildAbortController } from '../../utils/abortController.js';
@@ -300,6 +305,13 @@ export class AnthropicContentGenerator implements ContentGenerator {
   private effortClampWarned = false;
   private budgetDropWarned = false;
   private temperatureDropWarned = false;
+  // Stream watchdog tuning, resolved once (config field > env > default) so
+  // the env read + any invalid-value warning happen per generator, not per
+  // streaming request. Same guards the OpenAI pipeline applies — the two
+  // wires must not differ on whether a stalled stream is recoverable
+  // (issue #9005 finding 4).
+  private readonly streamIdleTimeoutMs: number;
+  private readonly streamMaxLifetimeMs: number;
 
   constructor(
     private contentGeneratorConfig: ContentGeneratorConfig,
@@ -355,6 +367,13 @@ export class AnthropicContentGenerator implements ContentGenerator {
       contentGeneratorConfig.schemaCompliance,
       contentGeneratorConfig.enableCacheControl,
     );
+
+    this.streamIdleTimeoutMs = resolveStreamIdleTimeoutMs(
+      contentGeneratorConfig,
+    );
+    this.streamMaxLifetimeMs = resolveStreamMaxLifetimeMs(
+      contentGeneratorConfig,
+    );
   }
 
   async generateContent(
@@ -425,8 +444,29 @@ export class AnthropicContentGenerator implements ContentGenerator {
       throw redactProxyError(error);
     }
 
+    // Two guards wrap the stream, identical to the OpenAI pipeline (the SDK
+    // `timeout` only bounds connect + first response). The inactivity
+    // watchdog aborts + surfaces a retryable ETIMEDOUT after `idleMs` of no
+    // events; the lifetime cap covers what the watchdog cannot — a drip-fed
+    // stream (e.g. long runs of low-content `thinking_delta` frames) resets
+    // the idle timer forever while never completing (issue #8597), so it
+    // aborts once `maxLifetimeMs` of accumulated upstream-wait has passed.
+    // `<= 0` disables each guard. Issue #9005 finding 4.
+    const idleMs = this.streamIdleTimeoutMs;
+    const maxLifetimeMs = this.streamMaxLifetimeMs;
+    const guardedStream =
+      idleMs > 0 || maxLifetimeMs > 0
+        ? withStreamGuards(
+            stream,
+            idleMs,
+            maxLifetimeMs,
+            () => perRequestAc.abort(),
+            request.config?.abortSignal,
+          )
+        : stream;
+
     const inner = this.processStreamWithEmptyFallback(
-      this.redactStreamErrors(stream),
+      this.redactStreamErrors(guardedStream),
       anthropicRequest,
       perRequestAc.signal,
       headers,
diff --git a/packages/core/src/core/contentGenerator.ts b/packages/core/src/core/contentGenerator.ts
index 99bd32f2cf8..ccd5c1ac500 100644
--- a/packages/core/src/core/contentGenerator.ts
+++ b/packages/core/src/core/contentGenerator.ts
@@ -87,9 +87,10 @@ export type ContentGeneratorConfig = {
   // Total-lifetime cap for one streaming response, NOT refreshed by chunk
   // arrival: a drip-fed stream resets the idle watchdog forever while never
   // completing the message (issue #8597), so that shape needs a bound the
-  // chunks cannot reset. `<= 0` disables it. Honored only by the
-  // OpenAI-compatible pipeline today — the Anthropic/Gemini generators do not
-  // implement it, so on those auth types the drip-fed shape stays unbounded.
+  // chunks cannot reset. `<= 0` disables it. Honored by the OpenAI-compatible
+  // pipeline and the Anthropic generator (shared `withStreamGuards`,
+  // issue #9005 finding 4); the Gemini generator does not implement it, so on
+  // that auth type the drip-fed shape stays unbounded.
   streamMaxLifetimeMs?: number;
   maxRetries?: number; // Maximum retries for rate-limit errors
   retryInitialDelayMs?: number; // Initial delay for stream rate-limit retries
diff --git a/packages/core/src/core/geminiChat.ts b/packages/core/src/core/geminiChat.ts
index 21acebb22d6..fa299104d26 100644
--- a/packages/core/src/core/geminiChat.ts
+++ b/packages/core/src/core/geminiChat.ts
@@ -1541,14 +1541,14 @@ export const ORPHAN_TOOL_USE_REPAIR_REASON =
  * --- Partial-push marker lifecycle ---------------------------------------
  *
  * Set together on (streamError + hasToolCall + hasContent) inside
- * `processStreamResponse`. Cleared together by `popPartialIfPushed` on a
+ * `processStreamResponse`. Cleared together by `popPendingPartialAssistantTurn` on a
  * retryable error rollback, or flushed together to JSONL by the outer
  * `finally` after the retry loop exits. Defense-in-depth: every
  * history-mutation method (clearHistory / addHistory / setHistory /
  * truncateHistory / stripThoughtsFromHistory /
  * stripOrphanedUserEntriesFromHistory) resets both markers in lockstep so
  * a stale index can't shift onto an unrelated model turn and cause
- * `popPartialIfPushed` to splice the wrong entry. Any single-field reset
+ * `popPendingPartialAssistantTurn` to splice the wrong entry. Any single-field reset
  * is a bug.
  * ============================================================================
  */
@@ -3713,7 +3713,7 @@ export class GeminiChat {
           // Pop the partial `model[fc]` FIRST (if processStreamResponse
           // pushed one before re-throwing), THEN the recovery user turn.
           // Reversed order would strand `OUTPUT_RECOVERY_MESSAGE` as a real
-          // user turn. Index-checked pop mirrors `popPartialIfPushed`
+          // user turn. Index-checked pop mirrors `popPendingPartialAssistantTurn`
           // above — see the design note above
           // `ORPHAN_TOOL_USE_REPAIR_REASON` for the wedge mechanism and
           // the partial-push marker lifecycle.
@@ -4673,7 +4673,7 @@ export class GeminiChat {
   clearHistory(): void {
     this.history = [];
     // Any pending partial-push state points into the now-empty history;
-    // resetting prevents `popPartialIfPushed` from splicing whatever
+    // resetting prevents `popPendingPartialAssistantTurn` from splicing whatever
     // shows up at that index in a future send (defense-in-depth — the
     // helper also bounds-checks, but a stale marker that happens to
     // line up with a real model turn could otherwise pop the wrong
@@ -4827,7 +4827,7 @@ export class GeminiChat {
     this.history = history;
     // History replacement (compression, /clear, --resume reload) wipes
     // the index basis the partial-push marker was captured against. The
-    // marker MUST be cleared — otherwise `popPartialIfPushed` could find
+    // marker MUST be cleared — otherwise `popPendingPartialAssistantTurn` could find
     // a model turn at the stale index in the replacement history and
     // splice an entry that has nothing to do with the original partial
     // push, corrupting the conversation. Drop the paired deferred-record
@@ -5578,7 +5578,7 @@ export class GeminiChat {
       if (streamError !== null) {
         // Stream-error + tool-use partial: defer the JSONL append until
         // the outer retry loop decides whether to roll back this attempt.
-        // If the same send retries successfully, popPartialIfPushed clears
+        // If the same send retries successfully, popPendingPartialAssistantTurn clears
         // this stash and the failed attempt never lands on disk; if the
         // retry path doesn't apply (unretryable break), the stash is
         // flushed at the rethrow site so JSONL stays aligned with the
diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts
index c9966410166..b4f44e3cb89 100644
--- a/packages/core/src/core/openaiContentGenerator/pipeline.ts
+++ b/packages/core/src/core/openaiContentGenerator/pipeline.ts
@@ -33,12 +33,12 @@ import {
   isTieredEffortWireModel,
 } from '../modalityDefaults.js';
 import {
-  DEFAULT_STREAM_IDLE_TIMEOUT_MS,
-  DEFAULT_STREAM_MAX_LIFETIME_MS,
-  MAX_STREAM_GUARD_TIMEOUT_MS,
-  QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
-  QWEN_STREAM_MAX_LIFETIME_MS_ENV,
-} from './constants.js';
+  resolveStreamIdleTimeoutMs,
+  resolveStreamMaxLifetimeMs,
+  StreamInactivityTimeoutError,
+  StreamLifetimeExceededError,
+  withStreamGuards,
+} from '../stream-guards.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 import { getToolCallPreparations } from '../tool-call-preparation.js';
 import { InvalidStreamError } from '../invalid-stream-error.js';
@@ -175,59 +175,13 @@ export class StreamContentError extends Error {
   }
 }
 
-/**
- * Thrown when a streaming response goes silent past the inactivity timeout.
- * `code: 'ETIMEDOUT'` makes `classifyRetryError` treat it as a retryable
- * transport error, identical to a real socket read timeout.
- */
-export class StreamInactivityTimeoutError extends Error {
-  readonly code = 'ETIMEDOUT' as const;
-
-  constructor(
-    readonly idleMs: number,
-    readonly chunksReceived: number,
-    readonly streamLifetimeMs: number,
-  ) {
-    super(
-      `No stream activity for ${idleMs}ms after ${chunksReceived} chunks ` +
-        `(stream lifetime: ${streamLifetimeMs}ms). Set ` +
-        `${QWEN_STREAM_IDLE_TIMEOUT_MS_ENV} to increase this window ` +
-        `(or 0 to disable it).`,
-    );
-    this.name = 'StreamInactivityTimeoutError';
-  }
-}
-
-/**
- * Thrown when a streaming response exceeds its upstream-wait budget without
- * completing. The cap charges accumulated time blocked in `await it.next()`
- * (upstream latency), never the consumer's processing, so a buffered,
- * already-complete stream never trips it — the shape it catches is a
- * never-completing stream the inactivity watchdog cannot see: a drip-fed
- * gateway or a model crawling through an oversized response resets that
- * watchdog forever (issue #8597). Same retryable `ETIMEDOUT` code, so a
- * text-only generation resumes via the transport-continuation recovery; a
- * turn that already streamed a functionCall surfaces as a visible,
- * classified error instead.
- */
-export class StreamLifetimeExceededError extends Error {
-  readonly code = 'ETIMEDOUT' as const;
-
-  constructor(
-    readonly maxLifetimeMs: number,
-    readonly chunksReceived: number,
-    readonly streamLifetimeMs: number,
-  ) {
-    super(
-      `Stream exceeded its ${maxLifetimeMs}ms upstream-wait cap after ` +
-        `${chunksReceived} chunks without completing (wall clock: ` +
-        `${streamLifetimeMs}ms). Set ` +
-        `${QWEN_STREAM_MAX_LIFETIME_MS_ENV} to increase this cap ` +
-        `(or 0 to disable it).`,
-    );
-    this.name = 'StreamLifetimeExceededError';
-  }
-}
+// Stream watchdog errors are shared with the Anthropic wire — see
+// ../stream-guards.ts (issue #9005 finding 4). Re-exported so existing
+// imports from this module keep working.
+export {
+  StreamInactivityTimeoutError,
+  StreamLifetimeExceededError,
+} from '../stream-guards.js';
 
 /**
  * Maximum bytes of response body to include in NonSSEResponseError diagnostics.
@@ -318,214 +272,8 @@ function clampProviderOutputBudgetKeys(
   return samplingParams;
 }
 
-/**
- * Resolve a stream-guard timeout (ms). Precedence, for both guards: explicit
- * `ContentGeneratorConfig` field (programmatic, wins — including `0` to
- * disable) > the env deployment knob > the built-in default. A malformed env
- * value is ignored (with a `console.warn`) rather than failing the request.
- */
-function resolveStreamGuardMs(
-  fromConfig: number | undefined,
-  configLabel: string,
-  envName: string,
-  defaultMs: number,
-): number {
-  // 1. Explicit config field (programmatic) wins:
-  //    - `<= 0` disables the watchdog (downstream `> 0` guards skip it).
-  //    - Values above the JS timer ceiling are rejected: setTimeout silently
-  //      compresses them to 1ms, which would fire near-immediately.
-  //    - NaN/Infinity/non-integer are invalid.
-  if (typeof fromConfig === 'number') {
-    if (
-      Number.isInteger(fromConfig) &&
-      fromConfig <= MAX_STREAM_GUARD_TIMEOUT_MS
-    ) {
-      return fromConfig;
-    }
-    // eslint-disable-next-line no-console
-    console.warn(
-      `[qwen-code] Ignoring out-of-range ${configLabel}=${fromConfig} ` +
-        `(expected an integer in (-∞, ${MAX_STREAM_GUARD_TIMEOUT_MS}]); ` +
-        `falling back to ${envName}/default.`,
-    );
-  }
-  // 2. Env deployment knob. Strict decimal integer only — reject hex/scientific
-  //    notation/floats/signs so a typo can't silently become a surprising
-  //    timeout. `0` disables; values above the timer ceiling are rejected.
-  const raw = process.env[envName];
-  const trimmed = raw?.trim();
-  if (trimmed) {
-    if (/^\d+$/.test(trimmed)) {
-      const parsed = Number(trimmed);
-      if (parsed <= MAX_STREAM_GUARD_TIMEOUT_MS) {
-        return parsed;
-      }
-    }
-    // eslint-disable-next-line no-console
-    console.warn(
-      `[qwen-code] Ignoring invalid ${envName}="${raw}" ` +
-        `(expected an integer of milliseconds in [0, ${MAX_STREAM_GUARD_TIMEOUT_MS}]); ` +
-        `using default ${defaultMs}ms.`,
-    );
-  }
-  return defaultMs;
-}
-
-function resolveStreamIdleTimeoutMs(config: ContentGeneratorConfig): number {
-  return resolveStreamGuardMs(
-    config.streamIdleTimeoutMs,
-    'streamIdleTimeoutMs',
-    QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
-    DEFAULT_STREAM_IDLE_TIMEOUT_MS,
-  );
-}
-
-function resolveStreamMaxLifetimeMs(config: ContentGeneratorConfig): number {
-  return resolveStreamGuardMs(
-    config.streamMaxLifetimeMs,
-    'streamMaxLifetimeMs',
-    QWEN_STREAM_MAX_LIFETIME_MS_ENV,
-    DEFAULT_STREAM_MAX_LIFETIME_MS,
-  );
-}
-
-/**
- * Wraps a streaming chunk source with two guards. The inactivity watchdog: if
- * no chunk arrives for `idleMs`, `abortRequest()` is invoked (to abort the
- * underlying request and free the socket) and the iterator throws — a user
- * `AbortError` when the parent signal was cancelled, otherwise a retryable
- * ETIMEDOUT. The idle timer resets on every chunk (including
- * thinking/reasoning deltas), so an actively streaming model is never
- * interrupted by it. The lifetime cap does NOT reset: once the stream has
- * accumulated `maxLifetimeMs` of upstream-wait time (the time spent blocked
- * on the source, never the consumer's time after a yield) without
- * completing, the iterator throws the same way — the bound a drip-fed
- * stream cannot reset (issue #8597). `<= 0` disables each guard
- * independently.
- */
-async function* withStreamGuards(
-  source: AsyncIterable<OpenAI.Chat.ChatCompletionChunk>,
-  idleMs: number,
-  maxLifetimeMs: number,
-  abortRequest: () => void,
-  parentSignal: AbortSignal | undefined,
-): AsyncGenerator<OpenAI.Chat.ChatCompletionChunk> {
-  // Both guards off: pass the source through untouched. The caller's
-  // `idleMs > 0 || maxLifetimeMs > 0` already prevents this, but the invariant
-  // must live here too — with both `<= 0`, `wait` computes to `Infinity` and
-  // Node clamps `setTimeout(Infinity)` to ~1ms, so every stream would die
-  // instantly with a bogus lifetime error.
-  if (idleMs <= 0 && maxLifetimeMs <= 0) {
-    yield* source;
-    return;
-  }
-  const it = source[Symbol.asyncIterator]();
-  // Monotonic, never `Date.now()`: an NTP step must not kill a healthy
-  // generation on the next iteration (a forward jump) nor silently disable
-  // the cap until the clock catches up (a backward jump) — the hang this
-  // guard exists to bound. The setTimeout this races is a monotonic clock
-  // too, so the two agree.
-  const streamStartedAt = performance.now();
-  // The lifetime cap is on ACCUMULATED UPSTREAM-WAIT — the wall-clock time
-  // this loop spends blocked in `await it.next()`. It is deliberately NOT
-  // end-to-end delivery time: an upstream that finished and buffered its
-  // chunks owes nothing, however slowly the consumer drains (a paused IDE
-  // client, a big TUI render), and a stream whose terminal `done` resolves
-  // at the boundary completes rather than becoming a retry. The cap only
-  // bites while the consumer is actually waiting on the model — which is
-  // exactly where #8597's drip-fed, never-completing stream spends its time.
-  let upstreamMs = 0;
-  let chunksReceived = 0;
-  try {
-    while (true) {
-      const remainingMs =
-        maxLifetimeMs > 0
-          ? maxLifetimeMs - upstreamMs
-          : Number.POSITIVE_INFINITY;
-      // The upstream-wait budget is already spent; a further wait can only
-      // lose, so fail it here (the lifetime timer below normally wins first).
-      if (remainingMs <= 0) {
-        // Same precedence as the timer below: a user cancellation wins over
-        // the cap's retryable ETIMEDOUT.
-        if (parentSignal?.aborted) {
-          const abortErr = new Error('Aborted');
-          abortErr.name = 'AbortError';
-          throw abortErr;
-        }
-        abortRequest();
-        throw new StreamLifetimeExceededError(
-          maxLifetimeMs,
-          chunksReceived,
-          performance.now() - streamStartedAt,
-        );
-      }
-      const nextPromise = it.next();
-      const awaitedAt = performance.now();
-      let timer: ReturnType<typeof setTimeout> | undefined;
-      const timeout = new Promise<never>((_resolve, reject) => {
-        // The caller wraps only when at least one guard is positive, so at
-        // least one of these is finite.
-        const idleIn = idleMs > 0 ? idleMs : Number.POSITIVE_INFINITY;
-        const wait = Math.min(idleIn, remainingMs);
-        timer = setTimeout(
-          () => {
-            if (parentSignal?.aborted) {
-              // Plain Error (not DOMException) so error redaction's prototype
-              // clone cannot corrupt it; name 'AbortError' satisfies isAbortError.
-              const abortErr = new Error('Aborted');
-              abortErr.name = 'AbortError';
-              reject(abortErr);
-            } else if (remainingMs <= idleIn) {
-              abortRequest();
-              reject(
-                new StreamLifetimeExceededError(
-                  maxLifetimeMs,
-                  chunksReceived,
-                  performance.now() - streamStartedAt,
-                ),
-              );
-            } else {
-              abortRequest();
-              reject(
-                new StreamInactivityTimeoutError(
-                  idleMs,
-                  chunksReceived,
-                  performance.now() - streamStartedAt,
-                ),
-              );
-            }
-          },
-          Math.max(wait, 0),
-        );
-        timer.unref?.();
-      });
-      let result: IteratorResult<OpenAI.Chat.ChatCompletionChunk>;
-      try {
-        result = await Promise.race([nextPromise, timeout]);
-      } catch (err) {
-        // Once abortRequest() aborts the request, the orphaned next() rejects
-        // with an AbortError; swallow it so it is not an unhandled rejection.
-        void Promise.resolve(nextPromise).catch(() => {});
-        throw err;
-      } finally {
-        if (timer !== undefined) clearTimeout(timer);
-      }
-      if (result.done) return;
-      // Charge only the time this chunk took to arrive — the upstream
-      // latency — never the time the consumer spent after the previous yield.
-      upstreamMs += performance.now() - awaitedAt;
-      chunksReceived += 1;
-      yield result.value;
-    }
-  } finally {
-    abortRequest();
-    try {
-      await it.return?.();
-    } catch {
-      // The abort above is the cleanup that matters; ignore return failures.
-    }
-  }
-}
+// The stream-guard timeout resolvers and `withStreamGuards` are shared with
+// the Anthropic wire — see ../stream-guards.ts (issue #9005 finding 4).
 
 export type { PipelineConfig } from './types.js';
 
diff --git a/packages/core/src/core/stream-guards.ts b/packages/core/src/core/stream-guards.ts
new file mode 100644
index 00000000000..d1a98e84be6
--- /dev/null
+++ b/packages/core/src/core/stream-guards.ts
@@ -0,0 +1,291 @@
+/**
+ * @license
+ * Copyright 2026 Google LLC
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * Stream-safety guards shared by every streaming wire (OpenAI pipeline and
+ * Anthropic generator). Both wires face the same hazards once a streaming
+ * response has returned 200 — a stream that goes silent, and a drip-fed
+ * stream that never completes (issue #8597) — so the watchdog mechanics and
+ * their tuning knobs live here once instead of drifting per wire (issue
+ * #9005 finding 4). The SDK `timeout` only bounds connect + first response,
+ * which is why these guards exist at all.
+ */
+
+import type { ContentGeneratorConfig } from './contentGenerator.js';
+import {
+  DEFAULT_STREAM_IDLE_TIMEOUT_MS,
+  DEFAULT_STREAM_MAX_LIFETIME_MS,
+  MAX_STREAM_GUARD_TIMEOUT_MS,
+  QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
+  QWEN_STREAM_MAX_LIFETIME_MS_ENV,
+} from './openaiContentGenerator/constants.js';
+
+/**
+ * Thrown when a streaming response goes silent past the inactivity timeout.
+ * `code: 'ETIMEDOUT'` makes `classifyRetryError` treat it as a retryable
+ * transport error, identical to a real socket read timeout.
+ */
+export class StreamInactivityTimeoutError extends Error {
+  readonly code = 'ETIMEDOUT' as const;
+
+  constructor(
+    readonly idleMs: number,
+    readonly chunksReceived: number,
+    readonly streamLifetimeMs: number,
+  ) {
+    super(
+      `No stream activity for ${idleMs}ms after ${chunksReceived} chunks ` +
+        `(stream lifetime: ${streamLifetimeMs}ms). Set ` +
+        `${QWEN_STREAM_IDLE_TIMEOUT_MS_ENV} to increase this window ` +
+        `(or 0 to disable it).`,
+    );
+    this.name = 'StreamInactivityTimeoutError';
+  }
+}
+
+/**
+ * Thrown when a streaming response exceeds its upstream-wait budget without
+ * completing. The cap charges accumulated time blocked in `await it.next()`
+ * (upstream latency), never the consumer's processing, so a buffered,
+ * already-complete stream never trips it — the shape it catches is a
+ * never-completing stream the inactivity watchdog cannot see: a drip-fed
+ * gateway or a model crawling through an oversized response resets that
+ * watchdog forever (issue #8597). Same retryable `ETIMEDOUT` code, so a
+ * text-only generation resumes via the transport-continuation recovery; a
+ * turn that already streamed a functionCall surfaces as a visible,
+ * classified error instead.
+ */
+export class StreamLifetimeExceededError extends Error {
+  readonly code = 'ETIMEDOUT' as const;
+
+  constructor(
+    readonly maxLifetimeMs: number,
+    readonly chunksReceived: number,
+    readonly streamLifetimeMs: number,
+  ) {
+    super(
+      `Stream exceeded its ${maxLifetimeMs}ms upstream-wait cap after ` +
+        `${chunksReceived} chunks without completing (wall clock: ` +
+        `${streamLifetimeMs}ms). Set ` +
+        `${QWEN_STREAM_MAX_LIFETIME_MS_ENV} to increase this cap ` +
+        `(or 0 to disable it).`,
+    );
+    this.name = 'StreamLifetimeExceededError';
+  }
+}
+
+/**
+ * Resolve a stream-guard timeout (ms). Precedence, for both guards: explicit
+ * `ContentGeneratorConfig` field (programmatic, wins — including `0` to
+ * disable) > the env deployment knob > the built-in default. A malformed env
+ * value is ignored (with a `console.warn`) rather than failing the request.
+ */
+function resolveStreamGuardMs(
+  fromConfig: number | undefined,
+  configLabel: string,
+  envName: string,
+  defaultMs: number,
+): number {
+  // 1. Explicit config field (programmatic) wins:
+  //    - `<= 0` disables the watchdog (downstream `> 0` guards skip it).
+  //    - Values above the JS timer ceiling are rejected: setTimeout silently
+  //      compresses them to 1ms, which would fire near-immediately.
+  //    - NaN/Infinity/non-integer are invalid.
+  if (typeof fromConfig === 'number') {
+    if (
+      Number.isInteger(fromConfig) &&
+      fromConfig <= MAX_STREAM_GUARD_TIMEOUT_MS
+    ) {
+      return fromConfig;
+    }
+    // eslint-disable-next-line no-console
+    console.warn(
+      `[qwen-code] Ignoring out-of-range ${configLabel}=${fromConfig} ` +
+        `(expected an integer in (-∞, ${MAX_STREAM_GUARD_TIMEOUT_MS}]); ` +
+        `falling back to ${envName}/default.`,
+    );
+  }
+  // 2. Env deployment knob. Strict decimal integer only — reject hex/scientific
+  //    notation/floats/signs so a typo can't silently become a surprising
+  //    timeout. `0` disables; values above the timer ceiling are rejected.
+  const raw = process.env[envName];
+  const trimmed = raw?.trim();
+  if (trimmed) {
+    if (/^\d+$/.test(trimmed)) {
+      const parsed = Number(trimmed);
+      if (parsed <= MAX_STREAM_GUARD_TIMEOUT_MS) {
+        return parsed;
+      }
+    }
+    // eslint-disable-next-line no-console
+    console.warn(
+      `[qwen-code] Ignoring invalid ${envName}="${raw}" ` +
+        `(expected an integer of milliseconds in [0, ${MAX_STREAM_GUARD_TIMEOUT_MS}]); ` +
+        `using default ${defaultMs}ms.`,
+    );
+  }
+  return defaultMs;
+}
+
+export function resolveStreamIdleTimeoutMs(
+  config: ContentGeneratorConfig,
+): number {
+  return resolveStreamGuardMs(
+    config.streamIdleTimeoutMs,
+    'streamIdleTimeoutMs',
+    QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
+    DEFAULT_STREAM_IDLE_TIMEOUT_MS,
+  );
+}
+
+export function resolveStreamMaxLifetimeMs(
+  config: ContentGeneratorConfig,
+): number {
+  return resolveStreamGuardMs(
+    config.streamMaxLifetimeMs,
+    'streamMaxLifetimeMs',
+    QWEN_STREAM_MAX_LIFETIME_MS_ENV,
+    DEFAULT_STREAM_MAX_LIFETIME_MS,
+  );
+}
+
+/**
+ * Wraps a streaming chunk source with two guards. The inactivity watchdog: if
+ * no chunk arrives for `idleMs`, `abortRequest()` is invoked (to abort the
+ * underlying request and free the socket) and the iterator throws — a user
+ * `AbortError` when the parent signal was cancelled, otherwise a retryable
+ * ETIMEDOUT. The idle timer resets on every chunk (including
+ * thinking/reasoning deltas), so an actively streaming model is never
+ * interrupted by it. The lifetime cap does NOT reset: once the stream has
+ * accumulated `maxLifetimeMs` of upstream-wait time (the time spent blocked
+ * on the source, never the consumer's time after a yield) without
+ * completing, the iterator throws the same way — the bound a drip-fed
+ * stream cannot reset (issue #8597). `<= 0` disables each guard
+ * independently.
+ */
+export async function* withStreamGuards<T>(
+  source: AsyncIterable<T>,
+  idleMs: number,
+  maxLifetimeMs: number,
+  abortRequest: () => void,
+  parentSignal: AbortSignal | undefined,
+): AsyncGenerator<T> {
+  // Both guards off: pass the source through untouched. The caller's
+  // `idleMs > 0 || maxLifetimeMs > 0` already prevents this, but the invariant
+  // must live here too — with both `<= 0`, `wait` computes to `Infinity` and
+  // Node clamps `setTimeout(Infinity)` to ~1ms, so every stream would die
+  // instantly with a bogus lifetime error.
+  if (idleMs <= 0 && maxLifetimeMs <= 0) {
+    yield* source;
+    return;
+  }
+  const it = source[Symbol.asyncIterator]();
+  // Monotonic, never `Date.now()`: an NTP step must not kill a healthy
+  // generation on the next iteration (a forward jump) nor silently disable
+  // the cap until the clock catches up (a backward jump) — the hang this
+  // guard exists to bound. The setTimeout this races is a monotonic clock
+  // too, so the two agree.
+  const streamStartedAt = performance.now();
+  // The lifetime cap is on ACCUMULATED UPSTREAM-WAIT — the wall-clock time
+  // this loop spends blocked in `await it.next()`. It is deliberately NOT
+  // end-to-end delivery time: an upstream that finished and buffered its
+  // chunks owes nothing, however slowly the consumer drains (a paused IDE
+  // client, a big TUI render), and a stream whose terminal `done` resolves
+  // at the boundary completes rather than becoming a retry. The cap only
+  // bites while the consumer is actually waiting on the model — which is
+  // exactly where #8597's drip-fed, never-completing stream spends its time.
+  let upstreamMs = 0;
+  let chunksReceived = 0;
+  try {
+    while (true) {
+      const remainingMs =
+        maxLifetimeMs > 0
+          ? maxLifetimeMs - upstreamMs
+          : Number.POSITIVE_INFINITY;
+      // The upstream-wait budget is already spent; a further wait can only
+      // lose, so fail it here (the lifetime timer below normally wins first).
+      if (remainingMs <= 0) {
+        // Same precedence as the timer below: a user cancellation wins over
+        // the cap's retryable ETIMEDOUT.
+        if (parentSignal?.aborted) {
+          const abortErr = new Error('Aborted');
+          abortErr.name = 'AbortError';
+          throw abortErr;
+        }
+        abortRequest();
+        throw new StreamLifetimeExceededError(
+          maxLifetimeMs,
+          chunksReceived,
+          performance.now() - streamStartedAt,
+        );
+      }
+      const nextPromise = it.next();
+      const awaitedAt = performance.now();
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      const timeout = new Promise<never>((_resolve, reject) => {
+        // The caller wraps only when at least one guard is positive, so at
+        // least one of these is finite.
+        const idleIn = idleMs > 0 ? idleMs : Number.POSITIVE_INFINITY;
+        const wait = Math.min(idleIn, remainingMs);
+        timer = setTimeout(
+          () => {
+            if (parentSignal?.aborted) {
+              // Plain Error (not DOMException) so error redaction's prototype
+              // clone cannot corrupt it; name 'AbortError' satisfies isAbortError.
+              const abortErr = new Error('Aborted');
+              abortErr.name = 'AbortError';
+              reject(abortErr);
+            } else if (remainingMs <= idleIn) {
+              abortRequest();
+              reject(
+                new StreamLifetimeExceededError(
+                  maxLifetimeMs,
+                  chunksReceived,
+                  performance.now() - streamStartedAt,
+                ),
+              );
+            } else {
+              abortRequest();
+              reject(
+                new StreamInactivityTimeoutError(
+                  idleMs,
+                  chunksReceived,
+                  performance.now() - streamStartedAt,
+                ),
+              );
+            }
+          },
+          Math.max(wait, 0),
+        );
+        timer.unref?.();
+      });
+      let result: IteratorResult<T>;
+      try {
+        result = await Promise.race([nextPromise, timeout]);
+      } catch (err) {
+        // Once abortRequest() aborts the request, the orphaned next() rejects
+        // with an AbortError; swallow it so it is not an unhandled rejection.
+        void Promise.resolve(nextPromise).catch(() => {});
+        throw err;
+      } finally {
+        if (timer !== undefined) clearTimeout(timer);
+      }
+      if (result.done) return;
+      // Charge only the time this chunk took to arrive — the upstream
+      // latency — never the time the consumer spent after the previous yield.
+      upstreamMs += performance.now() - awaitedAt;
+      chunksReceived += 1;
+      yield result.value;
+    }
+  } finally {
+    abortRequest();
+    try {
+      await it.return?.();
+    } catch {
+      // The abort above is the cleanup that matters; ignore return failures.
+    }
+  }
+}

From fe7e2fb076f6acd7b8ebb4672b16cd255d27bcb4 Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 06:08:05 +0800
Subject: [PATCH 02/10] fix(core): keep Anthropic fallback probe signal live
 past the drain abort

The shared stream guard aborts the per-request controller the moment the source stream drains, but the Anthropic empty-stream fallback probe runs after that drain and was reusing the now-aborted signal, so the SDK rejected it with a spurious AbortError instead of surfacing the provider's real error (e.g. a 402 credit-balance response). Pass the caller's signal into the fallback and derive a fresh short-lived child for the probe, aborting it once the probe settles to release the SDK's abort listener.
---
 .../anthropicContentGenerator.test.ts         | 53 +++++++++++++++++++
 .../anthropicContentGenerator.ts              | 20 ++++++-
 2 files changed, 71 insertions(+), 2 deletions(-)

diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
index 34de875c2dd..0af172e0d16 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
@@ -4896,6 +4896,59 @@ describe('AnthropicContentGenerator', () => {
       );
     });
 
+    it('keeps the fallback probe signal live after the drain abort (no spurious AbortError)', async () => {
+      // Regression for the empty-fallback probe: the shared stream guard aborts
+      // the per-request controller the moment the source stream drains, which
+      // happens before the probe runs. The probe must NOT inherit that
+      // already-aborted signal, or the SDK rejects it immediately with a
+      // spurious AbortError instead of surfacing the provider's real error
+      // (e.g. a 402 credit-balance response). Model the SDK's abort semantics:
+      // the probe's `create` rejects at call time when handed an aborted signal.
+      const { AnthropicContentGenerator } = await importGenerator();
+      anthropicState.createImpl
+        .mockResolvedValueOnce(
+          (async function* () {
+            // Empty stream: drains immediately, after which the guard aborts
+            // the per-request controller and the fallback probe runs.
+          })(),
+        )
+        .mockImplementationOnce(
+          (_req: unknown, opts: { signal?: AbortSignal }) => {
+            if (opts?.signal?.aborted) {
+              const abortErr = new Error('The operation was aborted');
+              abortErr.name = 'AbortError';
+              return Promise.reject(abortErr);
+            }
+            return Promise.reject(new Error('402 credit balance is too low'));
+          },
+        );
+
+      const generator = new AnthropicContentGenerator(
+        {
+          model: 'claude-test',
+          apiKey: 'test-key',
+          timeout: 10_000,
+          maxRetries: 2,
+          samplingParams: { max_tokens: 123 },
+          schemaCompliance: 'auto',
+        },
+        mockConfig,
+      );
+
+      const stream = await generator.generateContentStream({
+        model: 'models/ignored',
+        contents: 'Hello',
+      } as unknown as GenerateContentParameters);
+
+      await expect(async () => {
+        for await (const _chunk of stream) {
+          void _chunk;
+        }
+      }).rejects.toThrow('402 credit balance is too low');
+
+      expect(anthropicState.createImpl).toHaveBeenCalledTimes(2);
+    });
+
     it.each([
       { case: 'an empty buffer', partialJson: '' },
       { case: 'a partial buffer', partialJson: '{"file_path":' },
diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts
index da41ffdb32a..04644c98979 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts
@@ -468,7 +468,12 @@ export class AnthropicContentGenerator implements ContentGenerator {
     const inner = this.processStreamWithEmptyFallback(
       this.redactStreamErrors(guardedStream),
       anthropicRequest,
-      perRequestAc.signal,
+      // The empty-stream fallback probe needs a signal that is still live
+      // after the source stream drains. The shared stream guard aborts
+      // `perRequestAc` in its `finally` the moment the source drains — which
+      // happens before the probe runs — so pass the caller's signal instead;
+      // the probe derives its own short-lived child from it.
+      request.config?.abortSignal,
       headers,
       telemetryAttempt,
     );
@@ -1657,6 +1662,15 @@ export class AnthropicContentGenerator implements ContentGenerator {
     );
 
     let response: Message;
+    // Derive a fresh short-lived child for the probe from the caller's signal.
+    // Reusing the per-request controller is wrong here: the shared stream guard
+    // already aborted it when the source drained (its `finally` cleanup), and
+    // passing an already-aborted signal makes the SDK reject immediately with
+    // a spurious AbortError instead of surfacing the provider's real error
+    // (e.g. a 402 credit-balance response). A child of the caller's signal
+    // keeps the probe cancellable by the user while ignoring the drain abort,
+    // and aborting it once the probe settles releases the SDK's abort listener.
+    const probeAc = createChildAbortController(abortSignal);
     try {
       runtimeDiagnostics.recordAnthropicWireRequest(fallbackRequest);
       const fallbackAttempt = reportAnthropicFollowingRequest(
@@ -1664,13 +1678,15 @@ export class AnthropicContentGenerator implements ContentGenerator {
         telemetryAttempt,
       );
       response = (await this.client.messages.create(fallbackRequest, {
-        signal: abortSignal,
+        signal: probeAc.signal,
         ...(headers ? { headers } : {}),
       })) as Message;
       reportAnthropicResponse(fallbackAttempt, response);
       yield this.converter.convertAnthropicResponseToGemini(response);
     } catch (error) {
       throw redactProxyError(error);
+    } finally {
+      probeAc.abort();
     }
   }
 

From 595293ac0d9b19e52484d18e4977717a74ad87d1 Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 06:08:14 +0800
Subject: [PATCH 03/10] chore(vscode-ide-companion): regenerate settings schema
 artifact

Propagate the streamIdleTimeoutMs description change (OpenAI-compatible and Anthropic models) to the generated settings.schema.json via npm run generate:settings-schema, keeping the CI lockstep check green.
---
 packages/vscode-ide-companion/schemas/settings.schema.json | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/vscode-ide-companion/schemas/settings.schema.json b/packages/vscode-ide-companion/schemas/settings.schema.json
index 593e9d1016a..45197862e5c 100644
--- a/packages/vscode-ide-companion/schemas/settings.schema.json
+++ b/packages/vscode-ide-companion/schemas/settings.schema.json
@@ -770,7 +770,7 @@
               "type": "number"
             },
             "streamIdleTimeoutMs": {
-              "description": "Maximum inactivity between streamed chunks for OpenAI-compatible models, in milliseconds. Set to 0 to disable the idle guard. For provider-backed models, configure this field in the selected modelProviders entry.",
+              "description": "Maximum inactivity between streamed chunks for OpenAI-compatible and Anthropic models, in milliseconds. Set to 0 to disable the idle guard. For provider-backed models, configure this field in the selected modelProviders entry.",
               "type": "integer",
               "minimum": 0,
               "maximum": 2147483647

From e1f0c60cc0fb99cc15d11a85286429ffc7d7fd6c Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 06:08:21 +0800
Subject: [PATCH 04/10] test(core): pin Anthropic user-abort precedence over
 the idle watchdog
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Anthropic twin of the OpenAI pipeline.test.ts case: a user abort landing while the idle-watchdog timer is pending must surface as a non-retryable AbortError, not the retryable ETIMEDOUT the watchdog would otherwise raise (ETIMEDOUT is retryable, so the retry loop would resume the cancelled turn). Pins the parentSignal argument threaded into withStreamGuards — replacing it with undefined leaves this test red.
---
 .../anthropicContentGenerator.test.ts         | 32 +++++++++++++++++++
 1 file changed, 32 insertions(+)

diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
index 0af172e0d16..03c0ed0afdc 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
@@ -5379,6 +5379,38 @@ describe('AnthropicContentGenerator', () => {
       expect(error).toBeUndefined();
       expect(done).toBe(true);
     });
+
+    it('propagates a user AbortError (not ETIMEDOUT) when the parent signal is aborted', async () => {
+      // Anthropic twin of the OpenAI pipeline.test.ts case. A user Ctrl-C that
+      // lands while the idle-watchdog timer is pending must surface as a
+      // non-retryable AbortError, not the watchdog's retryable ETIMEDOUT —
+      // ETIMEDOUT is in the retryable set, so the retry loop would otherwise
+      // resume the turn the user just cancelled. This pins the `parentSignal`
+      // argument threaded into `withStreamGuards`: replacing it with
+      // `undefined` makes the timer reject with StreamInactivityTimeoutError
+      // and this test fail.
+      const callerAc = new AbortController();
+      const gated = gatedEventStream(); // never pushes → silent
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({
+        streamIdleTimeoutMs: 1000,
+        streamMaxLifetimeMs: 0,
+      });
+      const stream = await generator.generateContentStream({
+        ...streamRequest,
+        config: { abortSignal: callerAc.signal },
+      } as unknown as GenerateContentParameters);
+      const captured = consumeUntilSettled(stream);
+      callerAc.abort();
+      await vi.advanceTimersByTimeAsync(1000);
+      await vi.advanceTimersByTimeAsync(0);
+      const sentinel = Symbol('user-abort-did-not-propagate');
+      const err = await Promise.race([captured, Promise.resolve(sentinel)]);
+      expect(err).not.toBe(sentinel);
+      expect((err as Error).name).toBe('AbortError');
+      expect((err as { code?: string }).code).not.toBe('ETIMEDOUT');
+      expect(gated.wasReturned()).toBe(true);
+    });
   });
 
   describe('tool_choice mapping from Gemini toolConfig', () => {

From 8a5f8913c8bc7d7db219ca35dfa6677e2c2f8646 Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 06:08:28 +0800
Subject: [PATCH 05/10] chore(core): finish popPartialIfPushed rename in
 geminiChat.test.ts comments

The rename sweep to popPendingPartialAssistantTurn missed four comment references in geminiChat.test.ts (lines ~7737, ~10919, ~10980, ~13284). Update them so grep for the new method name reaches every site describing the rollback mechanism.
---
 packages/core/src/core/geminiChat.test.ts | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/packages/core/src/core/geminiChat.test.ts b/packages/core/src/core/geminiChat.test.ts
index 959fc544d4f..2aa3763dbab 100644
--- a/packages/core/src/core/geminiChat.test.ts
+++ b/packages/core/src/core/geminiChat.test.ts
@@ -7734,7 +7734,7 @@ describe('GeminiChat', async () => {
     it('rolls back the partial assistant turn when an InvalidStreamError fires after a tool_use chunk on the transient-stream retry budget', async () => {
       // Counterpart to the rate-limit rollback above. The
       // transient-stream retry budget (NO_FINISH_REASON /
-      // NO_RESPONSE_TEXT) has its own popPartialIfPushed call site —
+      // NO_RESPONSE_TEXT) has its own popPendingPartialAssistantTurn call site —
       // separate from the rate-limit branch the existing test
       // covers. Without a regression test, that call could be
       // accidentally removed and the rate-limit test would still
@@ -10916,7 +10916,7 @@ describe('GeminiChat', async () => {
       // chat-recording JSONL: the failed attempt's `recordAssistantTurn`
       // call must NOT have been flushed, so `--resume` won't rehydrate
       // a model[functionCall] turn the live session correctly discarded.
-      // Without the deferred-flush stash + popPartialIfPushed clear,
+      // Without the deferred-flush stash + popPendingPartialAssistantTurn clear,
       // `recordAssistantTurn` was called twice (once for the partial,
       // once for the success) and only the in-memory pop fixed live
       // history; the durable transcript stayed corrupt.
@@ -10977,7 +10977,7 @@ describe('GeminiChat', async () => {
 
         // Exactly one recording: the successful retry's text turn.
         // The failed attempt's partial functionCall must have been
-        // discarded by `popPartialIfPushed` clearing the deferred-flush
+        // discarded by `popPendingPartialAssistantTurn` clearing the deferred-flush
         // stash, never reaching the JSONL.
         expect(recordAssistantTurn).toHaveBeenCalledTimes(1);
         const recordedMessage = recordAssistantTurn.mock.calls[0]![0]
@@ -13281,7 +13281,7 @@ describe('GeminiChat', async () => {
     // stripOrphanedUserEntriesFromHistory). If any site forgets, a
     // stale `pendingPartialAssistantTurnIndex` could line up with an
     // unrelated model turn in the post-mutation history and cause
-    // `popPartialIfPushed` to splice the WRONG entry — silently losing
+    // `popPendingPartialAssistantTurn` to splice the WRONG entry — silently losing
     // a real assistant response.
     //
     // The markers are ephemeral within a single sendMessageStream

From 1096c56d0382a47b127ae46e1b44014655732441 Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 09:27:53 +0800
Subject: [PATCH 06/10] docs(users): extend stream guards prose to Anthropic
 providers

The docblock and regenerated settings schema already declare Anthropic
coverage; settings.md still scoped the guards to OpenAI-compatible
providers and claimed neither guard exists on Anthropic. Update the
heading and scope the remaining gap to the Gemini generator only.
---
 docs/users/configuration/settings.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/users/configuration/settings.md b/docs/users/configuration/settings.md
index ece947f463c..7eb166f2f78 100644
--- a/docs/users/configuration/settings.md
+++ b/docs/users/configuration/settings.md
@@ -229,9 +229,9 @@ These settings are read from operator scopes only (User, System, and SystemDefau
 
 `timeout` is the per-request timeout in milliseconds (default `120000`). Set it to `0` to disable the request timeout — matching the `QWEN_STREAM_IDLE_TIMEOUT_MS=0` convention — rather than aborting the request. It can also be set via the `QWEN_CODE_API_TIMEOUT_MS` environment variable. This is distinct from the two stream guards below.
 
-**stream guards (OpenAI-compatible providers only):**
+**stream guards (OpenAI-compatible and Anthropic providers):**
 
-Two guards bound a streaming response, each accepting `0` to disable. Neither is implemented by the Anthropic/Gemini generators, which leave the drip-fed shape below unbounded.
+Two guards bound a streaming response, each accepting `0` to disable. The Gemini generator does not implement them, which leaves the drip-fed shape below unbounded for Gemini models.
 
 - `streamIdleTimeoutMs` (default `240000`) bounds inactivity _between_ streamed chunks: a stream that goes silent for this long is aborted as a retryable `ETIMEDOUT`. For provider-backed models, set it under the matching `modelProviders[providerId][].generationConfig`; for runtime models, use `model.generationConfig`. An explicit model value takes precedence over `QWEN_STREAM_IDLE_TIMEOUT_MS`, and `0` disables the idle guard.
 - `QWEN_STREAM_MAX_LIFETIME_MS` (default `900000`) caps the _total_ upstream-wait time of one streaming response regardless of chunk flow — the bound a drip-fed stream that never completes cannot reset.

From c706c68cfc8e91c1849692aa1f973e920b60d907 Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 09:39:26 +0800
Subject: [PATCH 07/10] test(core): pin Anthropic stream-guard env knobs and
 probe signal linkage

Close the round-2 review gaps on the Anthropic wire:
- Honour QWEN_STREAM_MAX_LIFETIME_MS and the shared default lifetime cap
  through the Anthropic wiring (both previously untested; mutation-verified).
- Stub QWEN_STREAM_* envs in the fallback-probe regression so ambient
  disable values cannot vacate it.
- Pin the probe's caller-signal linkage (mid-probe cancel surfaces
  AbortError) and its post-probe abort cleanup.
---
 .../anthropicContentGenerator.test.ts         | 245 +++++++++++++++---
 1 file changed, 208 insertions(+), 37 deletions(-)

diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
index 03c0ed0afdc..cfdadd95118 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
@@ -11,6 +11,7 @@ import { FinishReason, GenerateContentResponse } from '@google/genai';
 import type { ContentGeneratorConfig } from '../contentGenerator.js';
 import {
   DEFAULT_STREAM_IDLE_TIMEOUT_MS,
+  DEFAULT_STREAM_MAX_LIFETIME_MS,
   DEFAULT_TIMEOUT,
   DISABLED_REQUEST_TIMEOUT_MS,
   QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
@@ -4904,49 +4905,132 @@ describe('AnthropicContentGenerator', () => {
       // spurious AbortError instead of surfacing the provider's real error
       // (e.g. a 402 credit-balance response). Model the SDK's abort semantics:
       // the probe's `create` rejects at call time when handed an aborted signal.
-      const { AnthropicContentGenerator } = await importGenerator();
-      anthropicState.createImpl
-        .mockResolvedValueOnce(
-          (async function* () {
-            // Empty stream: drains immediately, after which the guard aborts
-            // the per-request controller and the fallback probe runs.
-          })(),
-        )
-        .mockImplementationOnce(
-          (_req: unknown, opts: { signal?: AbortSignal }) => {
-            if (opts?.signal?.aborted) {
-              const abortErr = new Error('The operation was aborted');
-              abortErr.name = 'AbortError';
-              return Promise.reject(abortErr);
-            }
-            return Promise.reject(new Error('402 credit balance is too low'));
+      //
+      // The guards must be ON for this regression to bite — the spurious
+      // AbortError only exists because the guard's drain-time abort precedes
+      // the probe. Stub the env knobs so an ambient `QWEN_STREAM_*=0`
+      // (documented disable values) in the dev/CI shell can't silently switch
+      // the guards off and vacate the test.
+      vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, undefined);
+      vi.stubEnv(QWEN_STREAM_MAX_LIFETIME_MS_ENV, undefined);
+      try {
+        const { AnthropicContentGenerator } = await importGenerator();
+        anthropicState.createImpl
+          .mockResolvedValueOnce(
+            (async function* () {
+              // Empty stream: drains immediately, after which the guard aborts
+              // the per-request controller and the fallback probe runs.
+            })(),
+          )
+          .mockImplementationOnce(
+            (_req: unknown, opts: { signal?: AbortSignal }) => {
+              if (opts?.signal?.aborted) {
+                const abortErr = new Error('The operation was aborted');
+                abortErr.name = 'AbortError';
+                return Promise.reject(abortErr);
+              }
+              return Promise.reject(new Error('402 credit balance is too low'));
+            },
+          );
+
+        const generator = new AnthropicContentGenerator(
+          {
+            model: 'claude-test',
+            apiKey: 'test-key',
+            timeout: 10_000,
+            maxRetries: 2,
+            samplingParams: { max_tokens: 123 },
+            schemaCompliance: 'auto',
           },
+          mockConfig,
         );
 
-      const generator = new AnthropicContentGenerator(
-        {
-          model: 'claude-test',
-          apiKey: 'test-key',
-          timeout: 10_000,
-          maxRetries: 2,
-          samplingParams: { max_tokens: 123 },
-          schemaCompliance: 'auto',
-        },
-        mockConfig,
-      );
+        const stream = await generator.generateContentStream({
+          model: 'models/ignored',
+          contents: 'Hello',
+        } as unknown as GenerateContentParameters);
 
-      const stream = await generator.generateContentStream({
-        model: 'models/ignored',
-        contents: 'Hello',
-      } as unknown as GenerateContentParameters);
+        await expect(async () => {
+          for await (const _chunk of stream) {
+            void _chunk;
+          }
+        }).rejects.toThrow('402 credit balance is too low');
 
-      await expect(async () => {
-        for await (const _chunk of stream) {
-          void _chunk;
-        }
-      }).rejects.toThrow('402 credit balance is too low');
+        expect(anthropicState.createImpl).toHaveBeenCalledTimes(2);
+      } finally {
+        vi.unstubAllEnvs();
+      }
+    });
 
-      expect(anthropicState.createImpl).toHaveBeenCalledTimes(2);
+    it('aborts the fallback probe when the caller signal cancels mid-probe', async () => {
+      // Pins the probe's caller-signal linkage: the probe derives a child of
+      // the caller's signal, so a user Ctrl-C landing while the non-streaming
+      // probe is in flight (the quota/billing-shaped 200-but-empty response)
+      // aborts the probe instead of letting it run to completion against the
+      // provider and spend quota on a turn already cancelled. Mutant check:
+      // deriving the child from `undefined` leaves the hung probe unsettled
+      // and this test fails on the timeout assertion.
+      vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, undefined);
+      vi.stubEnv(QWEN_STREAM_MAX_LIFETIME_MS_ENV, undefined);
+      try {
+        const { AnthropicContentGenerator } = await importGenerator();
+        const callerAc = new AbortController();
+        anthropicState.createImpl
+          .mockResolvedValueOnce(
+            (async function* () {
+              // Empty stream: drains immediately, then the probe runs.
+            })(),
+          )
+          .mockImplementationOnce(
+            // A probe that hangs until its signal aborts, modelling an
+            // in-flight non-streaming request.
+            (_req: unknown, opts: { signal?: AbortSignal }) =>
+              new Promise((_resolve, reject) => {
+                const abortErr = new Error('The operation was aborted');
+                abortErr.name = 'AbortError';
+                if (opts?.signal?.aborted) {
+                  reject(abortErr);
+                  return;
+                }
+                opts?.signal?.addEventListener('abort', () => reject(abortErr));
+              }),
+          );
+
+        const generator = new AnthropicContentGenerator(
+          {
+            model: 'claude-test',
+            apiKey: 'test-key',
+            timeout: 10_000,
+            maxRetries: 2,
+            samplingParams: { max_tokens: 123 },
+            schemaCompliance: 'auto',
+          },
+          mockConfig,
+        );
+
+        const stream = await generator.generateContentStream({
+          model: 'models/ignored',
+          contents: 'Hello',
+          config: { abortSignal: callerAc.signal },
+        } as unknown as GenerateContentParameters);
+
+        const settled = (async () => {
+          for await (const _chunk of stream) {
+            void _chunk;
+          }
+        })().catch((e: unknown) => e);
+
+        // Wait until the probe call is in flight, then cancel like a user.
+        await vi.waitFor(() =>
+          expect(anthropicState.createImpl).toHaveBeenCalledTimes(2),
+        );
+        callerAc.abort();
+        const err = await settled;
+        expect((err as Error).name).toBe('AbortError');
+        expect((err as Error).message).toBe('The operation was aborted');
+      } finally {
+        vi.unstubAllEnvs();
+      }
     });
 
     it.each([
@@ -5054,6 +5138,12 @@ describe('AnthropicContentGenerator', () => {
         streamingAttempt,
         expect.anything(),
       );
+      // The probe's short-lived child must be aborted once the probe settles:
+      // that is what releases the SDK's abort listener instead of leaving it
+      // attached until the caller's long-lived round signal ends.
+      const [, fallbackOptions] = anthropicState.createImpl.mock
+        .calls[1] as AnthropicCreateArgs;
+      expect(fallbackOptions?.signal?.aborted).toBe(true);
     });
   });
 
@@ -5336,6 +5426,87 @@ describe('AnthropicContentGenerator', () => {
       expect(gated.wasReturned()).toBe(true);
     });
 
+    it('honours QWEN_STREAM_MAX_LIFETIME_MS when no explicit config is set', async () => {
+      // Anthropic twin of the OpenAI pipeline.test.ts case: with no explicit
+      // `streamMaxLifetimeMs` config field, the constructor must resolve the
+      // cap from the deployment env knob. Mutant check: replacing the
+      // constructor's `resolveStreamMaxLifetimeMs(contentGeneratorConfig)`
+      // with `contentGeneratorConfig.streamMaxLifetimeMs ?? 0` disables the
+      // cap and this test fails on the sentinel.
+      vi.stubEnv(QWEN_STREAM_MAX_LIFETIME_MS_ENV, '4000');
+      const gated = gatedEventStream(); // drip-fed, never ends
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({});
+      const stream = await generator.generateContentStream(streamRequest);
+      const captured = consumeUntilSettled(stream);
+      gated.push({
+        type: 'message_start',
+        message: {
+          id: 'msg-1',
+          model: 'claude-test',
+          usage: { input_tokens: 1 },
+        },
+      });
+      for (let i = 0; i < 7; i++) {
+        gated.push({
+          type: 'content_block_delta',
+          index: 0,
+          delta: { type: 'thinking_delta', thinking: 't' },
+        });
+        await vi.advanceTimersByTimeAsync(500); // each drip resets the idle watchdog
+      }
+      await vi.advanceTimersByTimeAsync(1000); // t=4500 — past the 4s env cap
+      await vi.advanceTimersByTimeAsync(0);
+      const sentinel = Symbol('env-lifetime-cap-did-not-fire');
+      const err = await Promise.race([captured, Promise.resolve(sentinel)]);
+      expect(err).toMatchObject({
+        name: 'StreamLifetimeExceededError',
+        code: 'ETIMEDOUT',
+        maxLifetimeMs: 4000,
+      });
+      expect((err as Error).message).toContain('QWEN_STREAM_MAX_LIFETIME_MS');
+    });
+
+    it('uses the default lifetime cap when nothing overrides it', async () => {
+      // No explicit config, no QWEN_STREAM_* env: the cap must resolve to the
+      // shared default. The drips land every 200s — inside the 240s default
+      // idle window, so the idle watchdog stays quiet while the lifetime cap
+      // accumulates upstream wait up to the 900s default.
+      const gated = gatedEventStream(); // drip-fed, never ends
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({});
+      const stream = await generator.generateContentStream(streamRequest);
+      const captured = consumeUntilSettled(stream);
+      gated.push({
+        type: 'message_start',
+        message: {
+          id: 'msg-1',
+          model: 'claude-test',
+          usage: { input_tokens: 1 },
+        },
+      });
+      for (let i = 0; i < 5; i++) {
+        gated.push({
+          type: 'content_block_delta',
+          index: 0,
+          delta: { type: 'thinking_delta', thinking: 't' },
+        });
+        // Inside the 240s default idle window, so only the cap can fire.
+        await vi.advanceTimersByTimeAsync(200_000);
+      }
+      // The cap is reached at t=900s during the last advance; the final drip
+      // at t=800s keeps the 240s idle watchdog pending until t=1040s.
+      await vi.advanceTimersByTimeAsync(0);
+      const sentinel = Symbol('default-lifetime-cap-did-not-fire');
+      const err = await Promise.race([captured, Promise.resolve(sentinel)]);
+      expect(err).toMatchObject({
+        name: 'StreamLifetimeExceededError',
+        code: 'ETIMEDOUT',
+        maxLifetimeMs: DEFAULT_STREAM_MAX_LIFETIME_MS,
+      });
+      expect((err as Error).message).toContain('QWEN_STREAM_MAX_LIFETIME_MS');
+    });
+
     it('leaves streams unguarded when both timeouts are disabled (<= 0)', async () => {
       const gated = gatedEventStream();
       anthropicState.createImpl.mockResolvedValue(gated.stream);

From d254e7fe469d86263a936c00295df7b877fb965d Mon Sep 17 00:00:00 2001
From: "jinjing.zzj" <jinjing.zzj@alibaba-inc.com>
Date: Tue, 25 Aug 2026 12:16:24 +0800
Subject: [PATCH 08/10] test(core): pin QWEN_STREAM_IDLE_TIMEOUT_MS env
 resolution on the Anthropic wire

Add the idle-knob twin of the lifetime env test in the watchdog suite:
with no explicit streamIdleTimeoutMs config, the constructor must
resolve the idle window from QWEN_STREAM_IDLE_TIMEOUT_MS. Mirrors the
OpenAI pipeline.test.ts case. Mutant check: replacing the constructor's
resolveStreamIdleTimeoutMs(contentGeneratorConfig) with the direct
config ?? DEFAULT fallback leaves this test red (env 3000 no longer
fires at 3000ms).
---
 .../anthropicContentGenerator.test.ts         | 36 +++++++++++++++++++
 1 file changed, 36 insertions(+)

diff --git a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
index cfdadd95118..dd3de2a9caa 100644
--- a/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
+++ b/packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts
@@ -5313,6 +5313,42 @@ describe('AnthropicContentGenerator', () => {
       });
     });
 
+    it('honours QWEN_STREAM_IDLE_TIMEOUT_MS when no explicit config is set', async () => {
+      // Anthropic twin of the OpenAI pipeline.test.ts case: with no explicit
+      // `streamIdleTimeoutMs` config field, the constructor must resolve the
+      // idle window from the deployment env knob. Mutant check: replacing the
+      // constructor's `resolveStreamIdleTimeoutMs(contentGeneratorConfig)`
+      // with `contentGeneratorConfig.streamIdleTimeoutMs ??
+      // DEFAULT_STREAM_IDLE_TIMEOUT_MS` ignores the env knob (falls back to
+      // the 240s default) and this test fails on the sentinel.
+      vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '3000');
+      const gated = gatedEventStream(); // never pushes → silent
+      anthropicState.createImpl.mockResolvedValue(gated.stream);
+      const generator = await buildGenerator({});
+      const stream = await generator.generateContentStream(streamRequest);
+      const captured = consumeUntilSettled(stream);
+      await vi.advanceTimersByTimeAsync(2999); // inside the env window
+      await vi.advanceTimersByTimeAsync(0);
+      const earlySentinel = Symbol('idle-watchdog-fired-before-env-value');
+      const early = await Promise.race([
+        captured,
+        Promise.resolve(earlySentinel),
+      ]);
+      expect(early).toBe(earlySentinel); // not yet at the env value
+      await vi.advanceTimersByTimeAsync(1); // t=3000 — the env value
+      await vi.advanceTimersByTimeAsync(0);
+      const lateSentinel = Symbol('env-idle-watchdog-did-not-fire');
+      const err = await Promise.race([captured, Promise.resolve(lateSentinel)]);
+      expect(err).toMatchObject({
+        name: 'StreamInactivityTimeoutError',
+        code: 'ETIMEDOUT',
+        idleMs: 3000,
+        chunksReceived: 0,
+      });
+      expect((err as Error).message).toContain('QWEN_STREAM_IDLE_TIMEOUT_MS');
+      expect(gated.wasReturned()).toBe(true);
+    });
+
     it('does not interrupt a stream whose events keep arriving inside the idle window', async () => {
       const gated = gatedEventStream();
       anthropicState.createImpl.mockResolvedValue(gated.stream);

From 7b4e0c98285f450e56f59e9fc32ec3d8eb0f8cb5 Mon Sep 17 00:00:00 2001
From: yiliang114 <effortyiliang@gmail.com>
Date: Fri, 28 Aug 2026 14:04:58 +0800
Subject: [PATCH 09/10] ci: serialize helper test files

---
 .github/workflows/ci.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bdc74da446b..e99774e64bc 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -355,7 +355,7 @@ jobs:
           node scripts/lint.js --setup
           node scripts/lint.js --actionlint
           node scripts/lint.js --yamllint
-          node --test ${{ env.HELPER_TESTS }}
+          node --test --test-concurrency=1 ${{ env.HELPER_TESTS }}
 
       # Avoid setup-node downloads on ECS, where nodejs.org may be unreachable
       # through the egress proxy; reuse the machine's Node instead.
@@ -496,7 +496,7 @@ jobs:
       # regression tests. Linux-only (they're platform-independent).
       - name: 'Run .github/scripts helper tests'
         if: "${{ needs.classify_pr.outputs.skip_ci != 'true' && steps.ci_profile.outputs.ci_profile == 'full' }}"
-        run: 'node --test ${{ env.HELPER_TESTS }}'
+        run: 'node --test --test-concurrency=1 ${{ env.HELPER_TESTS }}'
 
       # The install-script packaging suite needs zip/unzip, and throws on a
       # CI host that ships neither, so a silent skip there is impossible.

From 4afc17a1112ca76a8cd8a4f52ba3b345c3232242 Mon Sep 17 00:00:00 2001
From: yiliang114 <effortyiliang@gmail.com>
Date: Fri, 28 Aug 2026 15:29:16 +0800
Subject: [PATCH 10/10] Revert "ci: serialize helper test files"

This reverts commit 7b4e0c98285f450e56f59e9fc32ec3d8eb0f8cb5.
---
 .github/workflows/ci.yml | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index e99774e64bc..bdc74da446b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -355,7 +355,7 @@ jobs:
           node scripts/lint.js --setup
           node scripts/lint.js --actionlint
           node scripts/lint.js --yamllint
-          node --test --test-concurrency=1 ${{ env.HELPER_TESTS }}
+          node --test ${{ env.HELPER_TESTS }}
 
       # Avoid setup-node downloads on ECS, where nodejs.org may be unreachable
       # through the egress proxy; reuse the machine's Node instead.
@@ -496,7 +496,7 @@ jobs:
       # regression tests. Linux-only (they're platform-independent).
       - name: 'Run .github/scripts helper tests'
         if: "${{ needs.classify_pr.outputs.skip_ci != 'true' && steps.ci_profile.outputs.ci_profile == 'full' }}"
-        run: 'node --test --test-concurrency=1 ${{ env.HELPER_TESTS }}'
+        run: 'node --test ${{ env.HELPER_TESTS }}'
 
       # The install-script packaging suite needs zip/unzip, and throws on a
       # CI host that ships neither, so a silent skip there is impossible.