fix(core): re-land malformed stream retry with narrower nameless detection

packages/core/src/core/geminiChat.test.ts

@@ -5,6 +5,7 @@
  */
 
 import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
+import type OpenAI from 'openai';
 import type {
   Content,
   GenerateContentConfig,
@@ -23,6 +24,8 @@ import {
 import { RETRYABLE_STREAM_TRANSPORT_CODES } from './stream-transport-retry.js';
 import { classifyRetryError } from '../utils/retryErrorClassification.js';
 import { StreamContentError } from './openaiContentGenerator/pipeline.js';
+import { OpenAIContentGenerator } from './openaiContentGenerator/openaiContentGenerator.js';
+import type { OpenAICompatibleProvider } from './openaiContentGenerator/provider/index.js';
 import type { Config } from '../config/config.js';
 import { setSimulate429 } from '../utils/testUtils.js';
 import { uiTelemetryService } from '../telemetry/uiTelemetry.js';
@@ -4729,53 +4732,97 @@ describe('GeminiChat', async () => {
       }
     });
 
-    it('should keep protocol tag leak retries at the existing budget', async () => {
-      vi.useFakeTimers();
-      try {
-        vi.mocked(
-          mockContentGenerator.generateContentStream,
-        ).mockImplementation(async () =>
-          (async function* () {
-            yield {
-              candidates: [
-                {
-                  content: {
-                    parts: [
-                      {
-                        text: '<analysis>hidden</analysis><summary>leaked</summary>',
-                      },
-                    ],
+    it.each([
+      {
+        name: 'protocol tag leaks',
+        errorType: 'PROTOCOL_TAG_LEAK',
+        delta: {
+          reasoning_content: 'hidden reasoning',
+          content: '</think> leaked visible reasoning',
+        },
+        finishReason: 'stop',
+        retryCount: 2,
+      },
+      {
+        name: 'malformed tool calls',
+        errorType: 'MALFORMED_TOOL_CALL',
+        delta: {
+          tool_calls: [
+            {
+              index: 0,
+              id: 'call_without_name',
+              type: 'function',
+              function: { arguments: '{}' },
+            },
+          ],
+        },
+        finishReason: 'tool_calls',
+        retryCount: 4,
+      },
+    ] as const)(
+      'should retry $name through the OpenAI pipeline',
+      async (testCase) => {
+        vi.useFakeTimers();
+        try {
+          const create = vi.fn().mockImplementation(async () =>
+            (async function* () {
+              yield {
+                id: 'protocol-tag-leak',
+                created: 1,
+                model: 'test-model',
+                choices: [
+                  {
+                    index: 0,
+                    delta: testCase.delta,
+                    finish_reason: testCase.finishReason,
                   },
-                  finishReason: 'STOP',
-                },
-              ],
-            } as unknown as GenerateContentResponse;
-          })(),
-        );
+                ],
+              } as unknown as OpenAI.Chat.ChatCompletionChunk;
+            })(),
+          );
+          const provider = {
+            buildClient: () =>
+              ({ chat: { completions: { create } } }) as unknown as OpenAI,
+            buildRequest: (request: OpenAI.Chat.ChatCompletionCreateParams) =>
+              request,
+            buildHeaders: () => ({}),
+            getDefaultGenerationConfig: () => ({}),
+          } as OpenAICompatibleProvider;
+          const generator = new OpenAIContentGenerator(
+            { model: 'test-model', authType: AuthType.USE_OPENAI },
+            mockConfig,
+            provider,
+          );
+          vi.mocked(mockConfig.getContentGenerator).mockReturnValue(generator);
+          vi.mocked(mockConfig.getContentGeneratorConfig).mockReturnValue({
+            model: 'test-model',
+            authType: AuthType.USE_OPENAI,
+          });
 
-        const stream = await chat.sendMessageStream(
-          'test-model',
-          { message: 'test' },
-          'prompt-id-protocol-leak-budget',
-        );
-        await expectStreamExhaustion(stream);
+          const stream = await chat.sendMessageStream(
+            'test-model',
+            { message: 'test' },
+            `prompt-id-${testCase.errorType.toLowerCase()}-budget`,
+          );
+          await expectStreamExhaustion(stream);
 
-        expect(
-          mockContentGenerator.generateContentStream,
-        ).toHaveBeenCalledTimes(3);
-        expect(mockLogContentRetry).toHaveBeenCalledTimes(2);
-        expect(mockLogContentRetryFailure).toHaveBeenCalledWith(
-          mockConfig,
-          expect.objectContaining({
-            total_attempts: 3,
-            final_error_type: 'PROTOCOL_TAG_LEAK',
-            model: 'test-model',
-          }),
-        );
-      } finally {
-        vi.useRealTimers();
-      }
-    });
+          expect(create).toHaveBeenCalledTimes(testCase.retryCount + 1);
+          expect(mockLogContentRetry).toHaveBeenCalledTimes(
+            testCase.retryCount,
+          );
+          expect(mockLogContentRetryFailure).toHaveBeenCalledWith(
+            mockConfig,
+            expect.objectContaining({
+              total_attempts: testCase.retryCount + 1,
+              final_error_type: testCase.errorType,
+              model: 'test-model',
+            }),
+          );
+        } finally {
+          vi.useRealTimers();
+        }
+      },
+    );
 
     it('keeps invalid stream retry budgets independent across error types', async () => {
       vi.useFakeTimers();
@@ -6198,17 +6245,6 @@ describe('GeminiChat', async () => {
       }
     });
 
-    // NOTE: no test for the InvalidStreamError content-retry branch
-    // (geminiChat.ts ~line 1399). Verified unreachable for that error
-    // class: `isTransientStreamError` and `isContentError` are the
-    // same predicate (`error instanceof InvalidStreamError`), so the
-    // transient branch above always either `continue`s or `break`s
-    // before control reaches the content branch. The
-    // `popPartialIfPushed()` call there is preserved as
-    // defense-in-depth for a future error class that should diverge
-    // the predicates; see the comment block at that call site for
-    // the full analysis.
-
     it('rolls back the chat-recording entry too when the retry succeeds', async () => {
       // The in-memory rollback test above asserts `this.history` ends
       // clean after a retry-success. This test asserts the same about
@@ -8595,7 +8631,7 @@ describe('GeminiChat', async () => {
     }
 
     function invalidStream(
-      type: 'NO_FINISH_REASON' | 'PROTOCOL_TAG_LEAK',
+      type: InvalidStreamError['type'],
     ): AsyncGenerator<GenerateContentResponse> {
       return {
         [Symbol.asyncIterator]() {
@@ -9907,8 +9943,6 @@ describe('GeminiChat', async () => {
           invalidStream('PROTOCOL_TAG_LEAK'),
           invalidStream('PROTOCOL_TAG_LEAK'),
           invalidStream('PROTOCOL_TAG_LEAK'),
-          invalidStream('PROTOCOL_TAG_LEAK'),
-          invalidStream('PROTOCOL_TAG_LEAK'),
         ];
         let callIndex = 0;
         vi.mocked(

packages/core/src/core/geminiChat.ts

@@ -102,6 +102,9 @@ import {
   collectToolCallIdsFromHistory,
   normalizeModelToolCallIds,
 } from './toolCallIdUtils.js';
+import { InvalidStreamError } from './invalid-stream-error.js';
+
+export { InvalidStreamError };
 
 const debugLogger = createDebugLogger('QWEN_CODE_CHAT');
 
@@ -292,16 +295,6 @@ export type StreamEvent =
   | { type: StreamEventType.COMPRESSED; info: ChatCompressionInfo }
   | { type: StreamEventType.MODEL_FALLBACK; info: ModelFallbackInfo };
 
-/**
- * Options for retrying due to invalid content from the model.
- */
-interface ContentRetryOptions {
-  /** Total number of attempts to make (1 initial + N retries). */
-  maxAttempts: number;
-  /** The base delay in milliseconds for linear backoff. */
-  initialDelayMs: number;
-}
-
 interface TryCompressOptions {
   originalTokenCountOverride?: number;
   trigger?: CompactTrigger;
@@ -332,20 +325,12 @@ interface TryCompressOptions {
   customInstructions?: string;
 }
 
-const INVALID_CONTENT_RETRY_OPTIONS: ContentRetryOptions = {
-  maxAttempts: 2, // 1 initial call + 1 retry
-  initialDelayMs: 500,
-};
-
-// Some providers occasionally return transient stream anomalies: either an
-// empty stream (usage metadata only, no candidates), a stream that finishes
-// normally but contains no usable text, or a stream cut off without a finish
-// reason. All are retried with an independent budget (similar to rate-limit
-// retries) so they do not consume each other's retry budgets.
+// Model-output validation errors (protocol tag leaks, malformed tool calls)
+// and transient stream anomalies (empty streams, no usable text, missing
+// finish reason) use an independent retry budget so they do not consume each
+// other's or HTTP retries' budgets.
 const INVALID_STREAM_RETRY_CONFIG = {
   transientMaxRetries: 4,
-  // Protocol-tag leaks are model-output validation failures, not the
-  // provider-side empty/truncated streams covered by issue #6670.
   protocolTagLeakMaxRetries: 2,
   initialDelayMs: 2000,
 };
@@ -1041,23 +1026,6 @@ function stripThoughtPartsFromContent(content: Content): Content | null {
   };
 }
 
-/**
- * Custom error to signal that a stream completed with invalid content,
- * which should trigger a retry.
- */
-export class InvalidStreamError extends Error {
-  readonly type: 'NO_FINISH_REASON' | 'NO_RESPONSE_TEXT' | 'PROTOCOL_TAG_LEAK';
-
-  constructor(
-    message: string,
-    type: 'NO_FINISH_REASON' | 'NO_RESPONSE_TEXT' | 'PROTOCOL_TAG_LEAK',
-  ) {
-    super(message);
-    this.name = 'InvalidStreamError';
-    this.type = type;
-  }
-}
-
 const PROTOCOL_TAG_PREFIXES = [
   '<analysis',
   '</analysis',
@@ -2258,17 +2226,12 @@ export class GeminiChat {
 
         let lastFinishReason: string | undefined;
 
-        for (
-          let attempt = 0;
-          attempt < INVALID_CONTENT_RETRY_OPTIONS.maxAttempts;
-          attempt++
-        ) {
+        for (;;) {
           let streamYieldedChunk = false;
           try {
             if (suppressNextRetryEvent) {
               suppressNextRetryEvent = false;
             } else if (
-              attempt > 0 ||
               rateLimitRetryCount > 0 ||
               totalInvalidStreamRetryCount() > 0 ||
               transportStreamRetryCount > 0
@@ -2355,8 +2318,6 @@ export class GeminiChat {
                     skipDelay: skip,
                   },
                 };
-                // Don't count rate-limit retries against the content retry limit
-                attempt--;
                 await delayPromise;
                 continue;
               }
@@ -2400,8 +2361,6 @@ export class GeminiChat {
               });
               yield { type: StreamEventType.RETRY };
               suppressNextRetryEvent = true;
-              // Don't count transport retries against the content retry limit.
-              attempt--;
               await delay(delayMs, params.config?.abortSignal).promise;
               continue;
             }
@@ -2468,9 +2427,6 @@ export class GeminiChat {
                     };
                     yield { type: StreamEventType.RETRY };
                     suppressNextRetryEvent = true;
-                    // Do not count reactive compression against the content
-                    // validation retry budget.
-                    attempt--;
                     continue;
                   }
 
@@ -2518,8 +2474,8 @@ export class GeminiChat {
               break;
             }
 
-            // Invalid stream responses use an independent retry budget and do
-            // not consume the content retry budget.
+            // Invalid stream responses use INVALID_STREAM_RETRY_CONFIG, which
+            // is independent from HTTP retries handled by retryWithBackoff.
             const isInvalidStreamError = error instanceof InvalidStreamError;
             const maxInvalidStreamRetries =
               isInvalidStreamError && error.type === 'PROTOCOL_TAG_LEAK'
@@ -2558,46 +2514,9 @@ export class GeminiChat {
                 ),
               );
               yield { type: StreamEventType.RETRY };
-              // Don't count transient retries against content retry limit.
-              attempt--;
               await delay(delayMs, params.config?.abortSignal).promise;
               continue;
             }
-            // Invalid-stream budget exhausted — stop immediately.
-            if (isInvalidStreamError) {
-              break;
-            }
-
-            // Currently unreachable for `InvalidStreamError`. The
-            // `isContentError` predicate is identical to
-            // `isInvalidStreamError` (`error instanceof InvalidStreamError`),
-            // and the transient branch above already either continued or
-            // broke for that class. The branch is preserved as
-            // defense-in-depth: a future error class that should consume
-            // its own content-retry budget but NOT the transient one
-            // could be threaded through here without re-deriving the
-            // popPartialIfPushed sequence. No reachable test path until
-            // the predicates diverge.
-            const isContentError = error instanceof InvalidStreamError;
-            if (isContentError) {
-              if (attempt < INVALID_CONTENT_RETRY_OPTIONS.maxAttempts - 1) {
-                self.popPendingPartialAssistantTurn();
-                logContentRetry(
-                  self.config,
-                  new ContentRetryEvent(
-                    attempt,
-                    (error as InvalidStreamError).type,
-                    INVALID_CONTENT_RETRY_OPTIONS.initialDelayMs,
-                    model,
-                  ),
-                );
-                await delay(
-                  INVALID_CONTENT_RETRY_OPTIONS.initialDelayMs * (attempt + 1),
-                  params.config?.abortSignal,
-                ).promise;
-                continue;
-              }
-            }
             break;
           }
         }

packages/core/src/core/invalid-stream-error.ts

@@ -0,0 +1,22 @@
+/**
+ * @license
+ * Copyright 2025 Qwen
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+/**
+ * Custom error to signal invalid model stream content that should be retried.
+ */
+export class InvalidStreamError extends Error {
+  readonly type:
+    | 'NO_FINISH_REASON'
+    | 'NO_RESPONSE_TEXT'
+    | 'PROTOCOL_TAG_LEAK'
+    | 'MALFORMED_TOOL_CALL';
+
+  constructor(message: string, type: InvalidStreamError['type']) {
+    super(message);
+    this.name = 'InvalidStreamError';
+    this.type = type;
+  }
+}