feat: add TPM throttling error handling with 1-minute retry delay

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

@@ -18,6 +18,7 @@ import {
   StreamEventType,
   type StreamEvent,
 } from './geminiChat.js';
+import { StreamContentError } from './openaiContentGenerator/pipeline.js';
 import type { Config } from '../config/config.js';
 import { setSimulate429 } from '../utils/testUtils.js';
 import { uiTelemetryService } from '../telemetry/uiTelemetry.js';
@@ -930,6 +931,80 @@ describe('GeminiChat', () => {
       });
     });
 
+    it('should retry on TPM throttling StreamContentError with fixed delay', async () => {
+      vi.useFakeTimers();
+
+      try {
+        const tpmError = new StreamContentError('Throttling: TPM(1/1)');
+        async function* failingStreamGenerator() {
+          throw tpmError;
+
+          yield {} as GenerateContentResponse;
+        }
+        const failingStream = failingStreamGenerator();
+        const successStream = (async function* () {
+          yield {
+            candidates: [
+              {
+                content: { parts: [{ text: 'Success after TPM retry' }] },
+                finishReason: 'STOP',
+              },
+            ],
+          } as unknown as GenerateContentResponse;
+        })();
+
+        vi.mocked(mockContentGenerator.generateContentStream)
+          .mockResolvedValueOnce(failingStream)
+          .mockResolvedValueOnce(successStream);
+
+        const stream = await chat.sendMessageStream(
+          'test-model',
+          { message: 'test' },
+          'prompt-id-tpm-retry',
+        );
+
+        const iterator = stream[Symbol.asyncIterator]();
+        const first = await iterator.next();
+
+        expect(first.done).toBe(false);
+        expect(first.value.type).toBe(StreamEventType.RETRY);
+
+        // Resume generator to schedule the TPM delay, then advance timers.
+        const secondPromise = iterator.next();
+        await vi.advanceTimersByTimeAsync(60_000);
+        const second = await secondPromise;
+
+        expect(second.done).toBe(false);
+        expect(second.value.type).toBe(StreamEventType.RETRY);
+
+        const events: StreamEvent[] = [first.value, second.value];
+
+        for (;;) {
+          const next = await iterator.next();
+          if (next.done) break;
+          events.push(next.value);
+        }
+
+        expect(
+          mockContentGenerator.generateContentStream,
+        ).toHaveBeenCalledTimes(2);
+        expect(
+          events.filter((e) => e.type === StreamEventType.RETRY),
+        ).toHaveLength(2);
+        expect(
+          events.some(
+            (e) =>
+              e.type === StreamEventType.CHUNK &&
+              e.value.candidates?.[0]?.content?.parts?.[0]?.text ===
+                'Success after TPM retry',
+          ),
+        ).toBe(true);
+        expect(mockLogContentRetry).not.toHaveBeenCalled();
+      } finally {
+        vi.useRealTimers();
+      }
+    });
+
     describe('API error retry behavior', () => {
       beforeEach(() => {
         // Use a more direct mock for retry testing

packages/core/src/core/geminiChat.ts

@@ -17,7 +17,13 @@ import type {
   GenerateContentResponseUsageMetadata,
 } from '@google/genai';
 import { createUserContent } from '@google/genai';
-import { getErrorStatus, retryWithBackoff } from '../utils/retry.js';
+import {
+  getErrorStatus,
+  retryWithBackoff,
+  isTPMThrottlingError,
+} from '../utils/retry.js';
+import { StreamContentError } from './openaiContentGenerator/pipeline.js';
+import { createDebugLogger } from '../utils/debugLogger.js';
 import type { Config } from '../config/config.js';
 import { hasCycleInSchema } from '../tools/tools.js';
 import type { StructuredError } from './turn.js';
@@ -32,6 +38,8 @@ import {
 } from '../telemetry/types.js';
 import { uiTelemetryService } from '../telemetry/uiTelemetry.js';
 
+const debugLogger = createDebugLogger('QWEN_CODE_CHAT');
+
 export enum StreamEventType {
   /** A regular content chunk from the API. */
   CHUNK = 'chunk',
@@ -58,6 +66,16 @@ const INVALID_CONTENT_RETRY_OPTIONS: ContentRetryOptions = {
   maxAttempts: 2, // 1 initial call + 1 retry
   initialDelayMs: 500,
 };
+
+/**
+ * Options for retrying on TPM (Tokens Per Minute) throttling errors.
+ * These errors occur when the API rate limit is exceeded and are returned
+ * as stream content (finish_reason="error_finish") rather than HTTP errors.
+ */
+const TPM_RETRY_OPTIONS = {
+  maxRetries: 3,
+  delayMs: 60_000, // 1 minute - TPM quota typically resets within a minute window
+};
 /**
  * Returns true if the response is valid, false otherwise.
  *
@@ -268,14 +286,15 @@ export class GeminiChat {
     return (async function* () {
       try {
         let lastError: unknown = new Error('Request failed after all retries.');
+        let tpmRetryCount = 0;
 
         for (
           let attempt = 0;
           attempt < INVALID_CONTENT_RETRY_OPTIONS.maxAttempts;
           attempt++
         ) {
           try {
-            if (attempt > 0) {
+            if (attempt > 0 || tpmRetryCount > 0) {
               yield { type: StreamEventType.RETRY };
             }
 
@@ -294,6 +313,29 @@ export class GeminiChat {
             break;
           } catch (error) {
             lastError = error;
+
+            // Handle TPM throttling errors returned as stream content.
+            // These arrive as StreamContentError with finish_reason="error_finish"
+            // from the pipeline, containing the throttling message in the content.
+            if (
+              (error instanceof StreamContentError ||
+                isTPMThrottlingError(error)) &&
+              tpmRetryCount < TPM_RETRY_OPTIONS.maxRetries
+            ) {
+              tpmRetryCount++;
+              debugLogger.warn(
+                `TPM throttling detected (retry ${tpmRetryCount}/${TPM_RETRY_OPTIONS.maxRetries}). ` +
+                  `Waiting ${TPM_RETRY_OPTIONS.delayMs / 1000}s before retrying...`,
+              );
+              yield { type: StreamEventType.RETRY };
+              // Don't count TPM retries against the content retry limit
+              attempt--;
+              await new Promise((res) =>
+                setTimeout(res, TPM_RETRY_OPTIONS.delayMs),
+              );
+              continue;
+            }
+
             const isContentError = error instanceof InvalidStreamError;
 
             if (isContentError) {

packages/core/src/core/openaiContentGenerator/pipeline.test.ts

@@ -10,7 +10,7 @@ import type OpenAI from 'openai';
 import type { GenerateContentParameters } from '@google/genai';
 import { GenerateContentResponse, Type, FinishReason } from '@google/genai';
 import type { PipelineConfig } from './pipeline.js';
-import { ContentGenerationPipeline } from './pipeline.js';
+import { ContentGenerationPipeline, StreamContentError } from './pipeline.js';
 import { OpenAIContentConverter } from './converter.js';
 import type { Config } from '../../config/config.js';
 import type { ContentGeneratorConfig, AuthType } from '../contentGenerator.js';
@@ -510,6 +510,51 @@ describe('ContentGenerationPipeline', () => {
       );
     });
 
+    it('should throw StreamContentError when stream chunk contains error_finish', async () => {
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const userPromptId = 'test-prompt-id';
+
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'chunk-1',
+            object: 'chat.completion.chunk',
+            created: Date.now(),
+            model: 'test-model',
+            choices: [
+              {
+                index: 0,
+                delta: { content: 'Throttling: TPM(1/1)' },
+                finish_reason: 'error_finish',
+              },
+            ],
+          } as unknown as OpenAI.Chat.ChatCompletionChunk;
+        },
+      };
+
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        userPromptId,
+      );
+
+      await expect(async () => {
+        for await (const _ of resultGenerator) {
+          // consume stream
+        }
+      }).rejects.toThrow(StreamContentError);
+
+      expect(mockErrorHandler.handle).not.toHaveBeenCalled();
+      expect(mockConverter.convertOpenAIChunkToGemini).not.toHaveBeenCalled();
+    });
+
     it('should pass abort signal to OpenAI client for streaming requests', async () => {
       const abortController = new AbortController();
       const request: GenerateContentParameters = {

packages/core/src/core/openaiContentGenerator/pipeline.ts

@@ -15,6 +15,19 @@ import type { OpenAICompatibleProvider } from './provider/index.js';
 import { OpenAIContentConverter } from './converter.js';
 import type { ErrorHandler, RequestContext } from './errorHandler.js';
 
+/**
+ * Error thrown when the API returns an error embedded as stream content
+ * instead of a proper HTTP error. Some providers (e.g., certain OpenAI-compatible
+ * endpoints) return throttling errors as a normal SSE chunk with
+ * finish_reason="error_finish" and the error message in delta.content.
+ */
+export class StreamContentError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = 'StreamContentError';
+  }
+}
+
 export interface PipelineConfig {
   cliConfig: Config;
   provider: OpenAICompatibleProvider;
@@ -117,6 +130,17 @@ export class ContentGenerationPipeline {
     try {
       // Stage 2a: Convert and yield each chunk while preserving original
       for await (const chunk of stream) {
+        // Detect API errors returned as stream content.
+        // Some providers return errors (e.g., TPM throttling) as a normal SSE chunk
+        // with finish_reason="error_finish" and the error in delta.content,
+        // instead of returning a proper HTTP error status.
+        if ((chunk.choices?.[0]?.finish_reason as string) === 'error_finish') {
+          const errorContent =
+            chunk.choices?.[0]?.delta?.content?.trim() ||
+            'Unknown stream error';
+          throw new StreamContentError(errorContent);
+        }
+
         const response = this.converter.convertOpenAIChunkToGemini(chunk);
 
         // Stage 2b: Filter empty responses to avoid downstream issues
@@ -156,6 +180,12 @@ export class ContentGenerationPipeline {
       // Stage 2e: Stream completed successfully
       context.duration = Date.now() - context.startTime;
     } catch (error) {
+      // Re-throw StreamContentError directly so it can be handled by
+      // the caller's retry logic (e.g., TPM throttling retry in sendMessageStream)
+      if (error instanceof StreamContentError) {
+        throw error;
+      }
+
       // Clear streaming tool calls on error to prevent data pollution
       this.converter.resetStreamingToolCalls();