fix(core): retry requests when providers require thinking

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

@@ -826,6 +826,168 @@ describe('ContentGenerationPipeline', () => {
       expect(apiCall.tool_choice).toBe(testCase.expectedToolChoice);
     });
 
+    it('learns required thinking from a provider error and retries once', async () => {
+      mockContentGeneratorConfig = {
+        ...mockContentGeneratorConfig,
+        baseUrl:
+          'https://token-plan.cn-beijing.maas.aliyuncs.com/compatible-mode/v1',
+        model: 'qwen3.8-max-preview',
+        extra_body: { enable_thinking: true },
+      } as ContentGeneratorConfig;
+      mockConfig = {
+        ...mockConfig,
+        contentGeneratorConfig: mockContentGeneratorConfig,
+      };
+      pipeline = new ContentGenerationPipeline(mockConfig);
+
+      (mockProvider.buildRequest as Mock).mockImplementation((req) => ({
+        ...req,
+        enable_thinking: true,
+      }));
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([
+        { role: 'user', content: 'What is 2+2?' },
+      ]);
+      (mockConverter.convertGeminiToolsToOpenAI as Mock).mockResolvedValue([
+        { type: 'function', function: { name: 'respond_in_schema' } },
+      ]);
+      (mockConverter.convertOpenAIResponseToGemini as Mock).mockReturnValue(
+        new GenerateContentResponse(),
+      );
+
+      const requiredThinkingError = Object.assign(
+        new Error(
+          'The value of the enable_thinking parameter is restricted to True.',
+        ),
+        { status: 400 },
+      );
+      (mockClient.chat.completions.create as Mock)
+        .mockRejectedValueOnce(requiredThinkingError)
+        .mockResolvedValue({
+          id: 'r',
+          choices: [{ message: { content: '4' }, finish_reason: 'stop' }],
+        } as OpenAI.Chat.ChatCompletion);
+
+      const request: GenerateContentParameters = {
+        model: 'qwen3.8-max-preview',
+        contents: [{ parts: [{ text: 'What is 2+2?' }], role: 'user' }],
+        config: {
+          thinkingConfig: { includeThoughts: false },
+          tools: [
+            {
+              functionDeclarations: [
+                {
+                  name: 'respond_in_schema',
+                  parameters: { type: Type.OBJECT, properties: {} },
+                },
+              ],
+            },
+          ],
+          toolConfig: {
+            functionCallingConfig: { mode: FunctionCallingConfigMode.ANY },
+          },
+        },
+      };
+
+      await pipeline.execute(request, 'forked_query');
+      await pipeline.execute(request, 'forked_query');
+
+      const calls = (mockClient.chat.completions.create as Mock).mock.calls;
+      expect(calls).toHaveLength(3);
+      expect(calls[0][0]).toMatchObject({
+        enable_thinking: false,
+        tool_choice: 'required',
+      });
+      expect(calls[1][0].enable_thinking).toBe(true);
+      expect(calls[1][0].tool_choice).toBeUndefined();
+      expect(calls[2][0].enable_thinking).toBe(true);
+      expect(calls[2][0].tool_choice).toBeUndefined();
+      expect(mockErrorHandler.handle).not.toHaveBeenCalled();
+    });
+
+    it('handles the retry error when required-thinking retry fails', async () => {
+      mockContentGeneratorConfig = {
+        ...mockContentGeneratorConfig,
+        baseUrl:
+          'https://token-plan.cn-beijing.maas.aliyuncs.com/compatible-mode/v1',
+        model: 'qwen3.8-max-preview',
+        extra_body: { enable_thinking: true },
+      } as ContentGeneratorConfig;
+      mockConfig = {
+        ...mockConfig,
+        contentGeneratorConfig: mockContentGeneratorConfig,
+      };
+      pipeline = new ContentGenerationPipeline(mockConfig);
+
+      (mockProvider.buildRequest as Mock).mockImplementation((req) => ({
+        ...req,
+        enable_thinking: true,
+      }));
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([
+        { role: 'user', content: 'What is 2+2?' },
+      ]);
+
+      const requiredThinkingError = Object.assign(
+        new Error(
+          'The value of the enable_thinking parameter is restricted to True.',
+        ),
+        { status: 400 },
+      );
+      const retryError = new Error('retry failed');
+      (mockClient.chat.completions.create as Mock)
+        .mockRejectedValueOnce(requiredThinkingError)
+        .mockRejectedValueOnce(retryError);
+
+      const request: GenerateContentParameters = {
+        model: 'qwen3.8-max-preview',
+        contents: [{ parts: [{ text: 'What is 2+2?' }], role: 'user' }],
+        config: { thinkingConfig: { includeThoughts: false } },
+      };
+
+      await expect(pipeline.execute(request, 'forked_query')).rejects.toBe(
+        retryError,
+      );
+      expect(mockClient.chat.completions.create).toHaveBeenCalledTimes(2);
+      expect(mockErrorHandler.handle).toHaveBeenCalledWith(
+        retryError,
+        expect.any(Object),
+        request,
+      );
+    });
+
+    it('does not retry an unrelated 400 error', async () => {
+      mockContentGeneratorConfig = {
+        ...mockContentGeneratorConfig,
+        baseUrl:
+          'https://token-plan.cn-beijing.maas.aliyuncs.com/compatible-mode/v1',
+        model: 'qwen3.8-max-preview',
+      } as ContentGeneratorConfig;
+      mockConfig = {
+        ...mockConfig,
+        contentGeneratorConfig: mockContentGeneratorConfig,
+      };
+      pipeline = new ContentGenerationPipeline(mockConfig);
+
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([
+        { role: 'user', content: 'Hello' },
+      ]);
+      const error = Object.assign(new Error('Invalid request parameter.'), {
+        status: 400,
+      });
+      (mockClient.chat.completions.create as Mock).mockRejectedValue(error);
+
+      await expect(
+        pipeline.execute(
+          {
+            model: 'qwen3.8-max-preview',
+            contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+            config: { thinkingConfig: { includeThoughts: false } },
+          },
+          'forked_query',
+        ),
+      ).rejects.toBe(error);
+      expect(mockClient.chat.completions.create).toHaveBeenCalledTimes(1);
+    });
+
     it('should strip reasoning key from extra_body when thinking is disabled', async () => {
       // Arrange — provider injects reasoning via extra_body
       (mockProvider.buildRequest as Mock).mockImplementation((req) => ({
@@ -1672,6 +1834,73 @@ describe('ContentGenerationPipeline', () => {
   });
 
   describe('executeStream', () => {
+    it('retries stream creation when the provider requires thinking', async () => {
+      mockContentGeneratorConfig = {
+        ...mockContentGeneratorConfig,
+        baseUrl:
+          'https://token-plan.cn-beijing.maas.aliyuncs.com/compatible-mode/v1',
+        model: 'qwen3.8-max-preview',
+        extra_body: { enable_thinking: true },
+      } as ContentGeneratorConfig;
+      mockConfig = {
+        ...mockConfig,
+        contentGeneratorConfig: mockContentGeneratorConfig,
+      };
+      pipeline = new ContentGenerationPipeline(mockConfig);
+
+      (mockProvider.buildRequest as Mock).mockImplementation((req) => ({
+        ...req,
+        enable_thinking: true,
+      }));
+      (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([]);
+
+      const requiredThinkingError = Object.assign(
+        new Error(
+          'The value of the enable_thinking parameter is restricted to True.',
+        ),
+        { status: 400 },
+      );
+      const stream = {
+        async *[Symbol.asyncIterator]() {
+          // Empty response is sufficient: this test covers stream creation.
+        },
+      };
+      const failedCreate = Object.assign(Promise.resolve(stream), {
+        withResponse: () => Promise.reject(requiredThinkingError),
+      });
+      const successfulCreate = Object.assign(Promise.resolve(stream), {
+        withResponse: () =>
+          Promise.resolve({
+            data: stream,
+            response: new Response(null, {
+              headers: { 'content-type': 'text/event-stream' },
+            }),
+            request_id: 'retry-success',
+          }),
+      });
+      (mockClient.chat.completions.create as Mock)
+        .mockReturnValueOnce(failedCreate)
+        .mockReturnValueOnce(successfulCreate);
+
+      const result = await pipeline.executeStream(
+        {
+          model: 'qwen3.8-max-preview',
+          contents: [{ parts: [{ text: 'Quick question' }], role: 'user' }],
+          config: { thinkingConfig: { includeThoughts: false } },
+        },
+        'forked_query',
+      );
+      for await (const _ of result) {
+        // Drain the retried stream.
+      }
+
+      const calls = (mockClient.chat.completions.create as Mock).mock.calls;
+      expect(calls).toHaveLength(2);
+      expect(calls[0][0].enable_thinking).toBe(false);
+      expect(calls[1][0].enable_thinking).toBe(true);
+      expect(mockErrorHandler.handle).not.toHaveBeenCalled();
+    });
+
     it('should successfully execute streaming request', async () => {
       // Arrange
       const request: GenerateContentParameters = {

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

@@ -31,9 +31,21 @@ import { getToolCallPreparations } from '../tool-call-preparation.js';
 import { InvalidStreamError } from '../invalid-stream-error.js';
 import { logProtocolTagSanitized } from '../../telemetry/loggers.js';
 import { ProtocolTagSanitizedEvent } from '../../telemetry/types.js';
+import { getErrorMessage, getErrorStatus } from '../../utils/errors.js';
+import { getRateLimitErrorDetails } from '../../utils/rateLimit.js';
 
 const debugLogger = createDebugLogger('OPENAI_PIPELINE');
 
+function isRequiredThinkingError(error: unknown): boolean {
+  if (getErrorStatus(error) !== 400) return false;
+  const providerMessage = getRateLimitErrorDetails(error).providerMessage;
+  const message = `${getErrorMessage(error)} ${providerMessage ?? ''}`;
+  return (
+    message.includes('enable_thinking') &&
+    /(?:restricted to|must be) true/i.test(message)
+  );
+}
+
 /**
  * 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
@@ -281,6 +293,7 @@ export type { PipelineConfig } from './types.js';
 export class ContentGenerationPipeline {
   client: OpenAI;
   private contentGeneratorConfig: ContentGeneratorConfig;
+  private readonly requiredThinkingModels = new Set<string>();
   // Resolved once (config field > env > default) so the env read + any
   // invalid-value warning happen per pipeline, not per streaming request.
   private readonly streamIdleTimeoutMs: number;
@@ -826,10 +839,7 @@ export class ContentGenerationPipeline {
     const isDashScope = DashScopeOpenAICompatibleProvider.isDashScopeProvider(
       this.contentGeneratorConfig,
     );
-    const configModel = (this.contentGeneratorConfig.model ?? '').toLowerCase();
-    const thinkingMandatory =
-      this.contentGeneratorConfig.thinkingMandatory === true &&
-      model === configModel;
+    const thinkingMandatory = this.requiresThinking(model);
     const reasoningDisabled =
       request.config?.thinkingConfig?.includeThoughts === false ||
       this.contentGeneratorConfig.reasoning === false;
@@ -925,6 +935,16 @@ export class ContentGenerationPipeline {
     return providerRequest;
   }
 
+  private requiresThinking(model: string): boolean {
+    const normalizedModel = model.toLowerCase();
+    return (
+      this.requiredThinkingModels.has(normalizedModel) ||
+      (this.contentGeneratorConfig.thinkingMandatory === true &&
+        normalizedModel ===
+          (this.contentGeneratorConfig.model ?? '').toLowerCase())
+    );
+  }
+
   private buildGenerateContentConfig(
     request: GenerateContentParameters,
   ): Record<string, unknown> {
@@ -1066,9 +1086,9 @@ export class ContentGenerationPipeline {
     ) => Promise<T>,
   ): Promise<T> {
     const context = this.createRequestContext(request, isStreaming);
-
-    try {
-      const openaiRequest = await this.buildRequest(
+    let openaiRequest: OpenAI.Chat.ChatCompletionCreateParams | undefined;
+    const executeAttempt = async () => {
+      openaiRequest = await this.buildRequest(
         request,
         userPromptId,
         context,
@@ -1081,9 +1101,28 @@ export class ContentGenerationPipeline {
       openaiRequestCaptureContext.getStore()?.(openaiRequest);
       runtimeDiagnostics.recordOpenAIWireRequest(openaiRequest);
 
-      const result = await executor(openaiRequest, context);
-      return result;
+      return executor(openaiRequest, context);
+    };
+
+    try {
+      return await executeAttempt();
     } catch (error) {
+      const model = context.model.toLowerCase();
+      if (
+        (openaiRequest as Record<string, unknown> | undefined)?.[
+          'enable_thinking'
+        ] === false &&
+        request.config?.abortSignal?.aborted !== true &&
+        isRequiredThinkingError(error)
+      ) {
+        this.requiredThinkingModels.add(model);
+        debugLogger.warn('Retrying with required thinking enabled', { model });
+        try {
+          return await executeAttempt();
+        } catch (retryError) {
+          return await this.handleError(retryError, context, request);
+        }
+      }
       // Use shared error handling logic
       return await this.handleError(error, context, request);
     }