diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts index 29e2a408814..219a4baad12 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts @@ -27,6 +27,7 @@ import { StreamingToolCallParser } from './streamingToolCallParser.js'; import type { Config } from '../../config/config.js'; import { AuthType, type ContentGeneratorConfig } from '../contentGenerator.js'; import type { OpenAICompatibleProvider } from './provider/index.js'; +import { DefaultOpenAICompatibleProvider } from './provider/default.js'; import { DEFAULT_STREAM_IDLE_TIMEOUT_MS, MAX_STREAM_IDLE_TIMEOUT_MS, @@ -826,6 +827,308 @@ 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.each([ + { + name: 'preserving unrelated chat_template_kwargs', + extraBody: { + enable_thinking: false, + chat_template_kwargs: { + apply_chat_template: true, + enable_thinking: false, + }, + }, + initialChatTemplateKwargs: { + apply_chat_template: true, + enable_thinking: false, + }, + retryChatTemplateKwargs: { apply_chat_template: true }, + }, + { + name: 'removing empty chat_template_kwargs', + extraBody: { + chat_template_kwargs: { + enable_thinking: false, + }, + }, + initialChatTemplateKwargs: { enable_thinking: false }, + retryChatTemplateKwargs: undefined, + }, + ])( + 'retries without provider-configured thinking opt-outs on non-DashScope endpoints: $name', + async ({ + extraBody, + initialChatTemplateKwargs, + retryChatTemplateKwargs, + }) => { + mockContentGeneratorConfig = { + ...mockContentGeneratorConfig, + baseUrl: 'https://llm.example.com/v1', + model: 'Qwen3.6-27B', + extra_body: extraBody, + } as ContentGeneratorConfig; + const provider = new DefaultOpenAICompatibleProvider( + mockContentGeneratorConfig, + mockCliConfig, + ); + vi.spyOn(provider, 'buildClient').mockReturnValue(mockClient); + pipeline = new ContentGenerationPipeline({ + ...mockConfig, + provider, + contentGeneratorConfig: mockContentGeneratorConfig, + }); + + (mockConverter.convertGeminiRequestToOpenAI as Mock).mockReturnValue([ + { role: 'user', content: 'What is 2+2?' }, + ]); + (mockConverter.convertOpenAIResponseToGemini as Mock).mockReturnValue( + new GenerateContentResponse(), + ); + + const requiredThinkingError = Object.assign( + new Error('enable_thinking must be true for this model'), + { status: 400 }, + ); + (mockClient.chat.completions.create as Mock) + .mockRejectedValueOnce(requiredThinkingError) + .mockResolvedValue({ + id: 'r', + choices: [{ message: { content: '4' }, finish_reason: 'stop' }], + } as OpenAI.Chat.ChatCompletion); + + await pipeline.execute( + { + model: 'Qwen3.6-27B', + contents: [{ parts: [{ text: 'What is 2+2?' }], role: 'user' }], + config: { thinkingConfig: { includeThoughts: false } }, + }, + 'forked_query', + ); + + const calls = (mockClient.chat.completions.create as Mock).mock.calls; + expect(calls).toHaveLength(2); + expect(calls[0][0].chat_template_kwargs).toEqual( + initialChatTemplateKwargs, + ); + expect(calls[0][0].enable_thinking).toBeUndefined(); + if (retryChatTemplateKwargs === undefined) { + expect(calls[1][0].chat_template_kwargs).toBeUndefined(); + } else { + expect(calls[1][0].chat_template_kwargs).toEqual( + retryChatTemplateKwargs, + ); + } + expect(calls[1][0].enable_thinking).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 required-thinking errors after abort', async () => { + (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 }, + ); + (mockClient.chat.completions.create as Mock).mockRejectedValueOnce( + requiredThinkingError, + ); + + const request: GenerateContentParameters = { + model: 'qwen3.8-max-preview', + contents: [{ parts: [{ text: 'What is 2+2?' }], role: 'user' }], + config: { + abortSignal: AbortSignal.abort(), + thinkingConfig: { includeThoughts: false }, + }, + }; + + await expect(pipeline.execute(request, 'forked_query')).rejects.toBe( + requiredThinkingError, + ); + expect(mockClient.chat.completions.create).toHaveBeenCalledTimes(1); + expect(mockErrorHandler.handle).toHaveBeenCalledWith( + requiredThinkingError, + expect.any(Object), + request, + ); + }); + + it.each([ + 'Invalid request parameter.', + 'enable_thinking is not supported for this model', + ])( + 'does not retry a non-required-thinking 400 error: %s', + async (message) => { + 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(message), { + 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 +1975,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 = { diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index eea26e11026..27092568291 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/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\b/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(); // 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; @@ -911,13 +921,25 @@ export class ContentGenerationPipeline { } } - if (thinkingMandatory && isDashScope) { + if (thinkingMandatory) { const typed = providerRequest as unknown as Record; - // DashScope rejects forced tool selection while thinking is enabled. if (typed['enable_thinking'] === false) { delete typed['enable_thinking']; } - if (typed['tool_choice'] === 'required') { + const chatTemplateKwargs = typed['chat_template_kwargs'] as + | Record + | undefined; + if (chatTemplateKwargs?.['enable_thinking'] === false) { + const remaining = { ...chatTemplateKwargs }; + delete remaining['enable_thinking']; + if (Object.keys(remaining).length > 0) { + typed['chat_template_kwargs'] = remaining; + } else { + delete typed['chat_template_kwargs']; + } + } + // DashScope rejects forced tool selection while thinking is enabled. + if (isDashScope && typed['tool_choice'] === 'required') { delete typed['tool_choice']; } } @@ -925,6 +947,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 { @@ -1066,9 +1098,9 @@ export class ContentGenerationPipeline { ) => Promise, ): Promise { 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 +1113,34 @@ 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(); + const wireRequest = openaiRequest as Record | undefined; + const chatTemplateKwargs = wireRequest?.['chat_template_kwargs'] as + | Record + | undefined; + if ( + (wireRequest?.['enable_thinking'] === false || + chatTemplateKwargs?.['enable_thinking'] === false) && + request.config?.abortSignal?.aborted !== true && + isRequiredThinkingError(error) + ) { + this.requiredThinkingModels.add(model); + debugLogger.warn('Retrying with required thinking enabled', { + model, + originalError: getErrorMessage(error), + }); + 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); }