diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 9ca188e8ef5..4d601a748b5 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -67,6 +67,26 @@ describe('OpenAIContentConverter', () => { }; } + function withQwen3TaggedThinkingStreamParser(): RequestContext { + return { + ...withStreamParser(), + model: 'qwen3.8-max', + responseParsingOptions: { + contentOnlyThinkingTagLeaks: true, + taggedThinkingTagsAfterReasoning: true, + }, + }; + } + + function openAIStreamChunk( + delta: Record, + finishReason: string | null = null, + ): OpenAI.Chat.ChatCompletionChunk { + return { + choices: [{ index: 0, delta, finish_reason: finishReason }], + } as unknown as OpenAI.Chat.ChatCompletionChunk; + } + function hasOpenAIToolCalls( message: OpenAI.Chat.ChatCompletionMessageParam, ): message is OpenAI.Chat.ChatCompletionAssistantMessageParam & { @@ -5846,6 +5866,172 @@ describe('OpenAIContentConverter', () => { { text: 'still thinking', thought: true }, ]); }); + + it('should stream ordinary Qwen3 reasoning and content immediately', () => { + const context = withQwen3TaggedThinkingStreamParser(); + + const reasoningChunk = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ reasoning_content: 'step 1' }), + context, + ); + const contentChunk = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: 'answer' }), + context, + ); + + expect(reasoningChunk.candidates?.[0]?.content?.parts).toEqual([ + { text: 'step 1', thought: true }, + ]); + expect(contentChunk.candidates?.[0]?.content?.parts).toEqual([ + { text: 'answer' }, + ]); + }); + + it('should parse a balanced Qwen3 thinking block after reasoning', () => { + const context = withQwen3TaggedThinkingStreamParser(); + + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ reasoning_content: 'step 1' }), + context, + ); + const openingChunk = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: 'step 2answer' }, 'stop'), + context, + ); + + expect(openingChunk.candidates?.[0]?.content?.parts).toEqual([]); + expect(finalChunk.candidates?.[0]?.content?.parts).toEqual([ + { text: 'step 2', thought: true }, + { text: 'answer' }, + ]); + }); + + it('should reject an unclosed Qwen3 thinking block after reasoning', () => { + const context = withQwen3TaggedThinkingStreamParser(); + + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ reasoning_content: 'step 1' }), + context, + ); + + expect(() => + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: 'step 2' }, 'stop'), + context, + ), + ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' })); + }); + + it('should suppress a replayed short Qwen3 thinking block', () => { + const context = withQwen3TaggedThinkingStreamParser(); + const block = 'x'; + + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ reasoning_content: 'step 1' }), + context, + ); + const firstBlock = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block }), + context, + ); + const replayedBlock = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block }), + context, + ); + + expect(firstBlock.candidates?.[0]?.content?.parts).toEqual([ + { text: 'x', thought: true }, + ]); + expect(replayedBlock.candidates?.[0]?.content?.parts).toEqual([]); + }); + + it('should suppress a replayed short Qwen3 thinking opener', () => { + const context = withQwen3TaggedThinkingStreamParser(); + const opener = ''; + + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ reasoning_content: 'step 1' }), + context, + ); + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: opener }), + context, + ); + const replayedOpener = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: opener }), + context, + ); + const finalChunk = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: 'xanswer' }, 'stop'), + context, + ); + + expect(replayedOpener.candidates?.[0]?.content?.parts).toEqual([]); + expect(finalChunk.candidates?.[0]?.content?.parts).toEqual([ + { text: 'x', thought: true }, + { text: 'answer' }, + ]); + }); + + it('should suppress a replayed Qwen3 snapshot beyond the detection window', () => { + const context = withQwen3TaggedThinkingStreamParser(); + const block = `${'x'.repeat(1200)}answer`; + + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ reasoning_content: 'step 1' }), + context, + ); + for (let offset = 0; offset < block.length; offset += 100) { + converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block.slice(offset, offset + 100) }), + context, + ); + } + + const replay = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block }), + context, + ); + const rewind = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block.slice(0, -1) }), + context, + ); + const extension = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: `${block}!` }), + context, + ); + + expect(replay.candidates?.[0]?.content?.parts).toEqual([]); + expect(rewind.candidates?.[0]?.content?.parts).toEqual([]); + expect(extension.candidates?.[0]?.content?.parts).toEqual([ + { text: '!' }, + ]); + }); + + it('should preserve repeated short blocks for eager tagged parsing', () => { + const context = withTaggedThinkingStreamParser(); + const block = 'x'; + + const firstBlock = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block }), + context, + ); + const secondBlock = converter.convertOpenAIChunkToLlm( + openAIStreamChunk({ content: block }), + context, + ); + + expect(firstBlock.candidates?.[0]?.content?.parts).toEqual([ + { text: 'x', thought: true }, + ]); + expect(secondBlock.candidates?.[0]?.content?.parts).toEqual([ + { text: 'x', thought: true }, + ]); + }); }); describe('convertLlmToolsToOpenAI', () => { diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index 225d3d42209..a9d2e70f7bb 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -28,7 +28,10 @@ import { TOKEN_ESTIMATE_UNITS_PER_TOKEN, } from '../../utils/request-tokenizer/textTokenizer.js'; import type { RequestContext, StreamingTextDeltaState } from './types.js'; -import { parseTaggedThinkingText } from './taggedThinkingParser.js'; +import { + parseTaggedThinkingText, + TaggedThinkingParser, +} from './taggedThinkingParser.js'; import { convertSchema, relaxSchemaForFunctionCalling, @@ -1096,7 +1099,10 @@ function convertOpenAITextToParts( requestContext: RequestContext, final = true, ): Part[] { - if (!requestContext.responseParsingOptions?.taggedThinkingTags) { + if ( + !requestContext.responseParsingOptions?.taggedThinkingTags && + !requestContext.taggedThinkingParser + ) { return text ? [{ text }] : []; } @@ -1369,17 +1375,51 @@ export function convertOpenAIChunkToLlm( // Handle text content if (typeof choice.delta?.content === 'string') { - const normalizedContent = normalizeStreamingTextDelta( - choice.delta.content, - (requestContext.textDeltaState ??= { - emittedText: '', - emittedLength: 0, - cumulativeMode: false, - }), - ); - // Skip empty-string push mid-stream; still call on finish_reason to - // flush any buffered tagged-thinking content. - if (normalizedContent || choice.finish_reason) { + const rawContent = choice.delta.content; + const replayState = requestContext.textDeltaState; + const replayedTaggedThinkingSnapshot = + requestContext.responseParsingOptions + ?.taggedThinkingTagsAfterReasoning === true && + requestContext.taggedThinkingParser !== undefined && + replayState !== undefined && + (replayState.emittedText === rawContent || + (rawContent.length === replayState.emittedLength && + rawContent.startsWith(replayState.emittedText))) && + THINKING_TAG_PATTERN.test(rawContent); + if (replayedTaggedThinkingSnapshot) { + replayState.emittedText = rawContent; + replayState.emittedLength = rawContent.length; + replayState.cumulativeMode = true; + } + const normalizedContent = replayedTaggedThinkingSnapshot + ? '' + : normalizeStreamingTextDelta( + rawContent, + (requestContext.textDeltaState ??= { + emittedText: '', + emittedLength: 0, + cumulativeMode: false, + }), + ); + const taggedThinkingCandidate = + (requestContext.pendingThinkingTagCandidate?.text ?? '') + + normalizedContent; + if ( + requestContext.responseParsingOptions + ?.taggedThinkingTagsAfterReasoning && + (requestContext.hasStructuredReasoningContent || reasoningText) && + LEADING_THINKING_TAG_PATTERN.test(taggedThinkingCandidate) && + !taggedThinkingCandidate.trimStart().startsWith(' { expect(results).toHaveLength(2); expect(results[0]).toBe(mockLlmResponse1); expect(results[1]).toBe(mockLlmResponse2); + expect(mockProvider.getResponseParsingOptions).toHaveBeenCalledWith( + 'test-model', + ); const [, firstChunkContext] = ( mockConverter.convertOpenAIChunkToLlm as Mock ).mock.calls[0]; diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index fdde28cbdfc..c8f6220481c 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.ts @@ -644,7 +644,11 @@ export class ContentGenerationPipeline { ]; yield response; } - } else if (context.pendingThinkingTagCandidate) { + } else if ( + context.pendingThinkingTagCandidate || + (context.responseParsingOptions?.taggedThinkingTagsAfterReasoning && + context.taggedThinkingParser?.hasUnclosedThought()) + ) { throw new InvalidStreamError( 'Model response leaked thinking tags.', 'PROTOCOL_TAG_LEAK', @@ -1306,7 +1310,7 @@ export class ContentGenerationPipeline { ? new StreamingToolCallParser() : undefined; const responseParsingOptions = - this.config.provider.getResponseParsingOptions?.(); + this.config.provider.getResponseParsingOptions?.(effectiveModel); const taggedThinkingParser = isStreaming && responseParsingOptions?.taggedThinkingTags ? new TaggedThinkingParser() diff --git a/packages/core/src/core/openaiContentGenerator/provider/default.test.ts b/packages/core/src/core/openaiContentGenerator/provider/default.test.ts index 86a68afb00a..8e375b0dac7 100644 --- a/packages/core/src/core/openaiContentGenerator/provider/default.test.ts +++ b/packages/core/src/core/openaiContentGenerator/provider/default.test.ts @@ -104,11 +104,18 @@ describe('DefaultOpenAICompatibleProvider', () => { }); describe('getResponseParsingOptions', () => { - it('enables leak handling without treating balanced tags as protocol', () => { - expect(provider.getResponseParsingOptions()).toEqual({ + it('keeps balanced tags visible for generic models', () => { + expect(provider.getResponseParsingOptions('gpt-4o')).toEqual({ contentOnlyThinkingTagLeaks: true, }); }); + + it('parses tagged thinking for Qwen3 models', () => { + expect(provider.getResponseParsingOptions('qwen3.8-max')).toEqual({ + contentOnlyThinkingTagLeaks: true, + taggedThinkingTagsAfterReasoning: true, + }); + }); }); describe('buildHeaders', () => { diff --git a/packages/core/src/core/openaiContentGenerator/provider/default.ts b/packages/core/src/core/openaiContentGenerator/provider/default.ts index 76eb7a839ec..1d49046beaa 100644 --- a/packages/core/src/core/openaiContentGenerator/provider/default.ts +++ b/packages/core/src/core/openaiContentGenerator/provider/default.ts @@ -43,7 +43,7 @@ type AssistantMessageWithReasoningFields = reasoning?: string | null; }; -function shouldMirrorReasoningContentForQwen3(model: string): boolean { +function isQwen3Model(model: string): boolean { return model.toLowerCase().includes('qwen3'); } @@ -205,7 +205,7 @@ export class DefaultOpenAICompatibleProvider const requestWithTokenLimits = this.clampConfiguredReasoningEffort( this.applyOutputTokenLimit(request), ); - const messages = shouldMirrorReasoningContentForQwen3(request.model) + const messages = isQwen3Model(request.model) ? requestWithTokenLimits.messages.map(mirrorReasoningContentToReasoning) : requestWithTokenLimits.messages; @@ -220,13 +220,16 @@ export class DefaultOpenAICompatibleProvider return {}; } - getResponseParsingOptions(): OpenAIResponseParsingOptions { + getResponseParsingOptions(model?: string): OpenAIResponseParsingOptions { // Hybrid-thinking models occasionally bypass the reasoning channel and // emit their thinking as literal / tags inside content // (observed in production on qwen3-class models, issue #6666). - // Honored on the streaming path only; non-streaming responses are - // not classified. - return { contentOnlyThinkingTagLeaks: true }; + return { + contentOnlyThinkingTagLeaks: true, + ...(model && isQwen3Model(model) + ? { taggedThinkingTagsAfterReasoning: true } + : {}), + }; } /** diff --git a/packages/core/src/core/openaiContentGenerator/provider/types.ts b/packages/core/src/core/openaiContentGenerator/provider/types.ts index f70f9fcd13f..85332a44636 100644 --- a/packages/core/src/core/openaiContentGenerator/provider/types.ts +++ b/packages/core/src/core/openaiContentGenerator/provider/types.ts @@ -31,7 +31,7 @@ export interface OpenAICompatibleProvider { userPromptId: string, ): OpenAI.Chat.ChatCompletionCreateParams; getDefaultGenerationConfig(): GenerateContentConfig; - getResponseParsingOptions?(): OpenAIResponseParsingOptions; + getResponseParsingOptions?(model?: string): OpenAIResponseParsingOptions; getRequestContextOverrides?(): OpenAIRequestContextOverrides; } diff --git a/packages/core/src/core/openaiContentGenerator/responseParsingOptions.ts b/packages/core/src/core/openaiContentGenerator/responseParsingOptions.ts index 03b6c39c4da..2768123428b 100644 --- a/packages/core/src/core/openaiContentGenerator/responseParsingOptions.ts +++ b/packages/core/src/core/openaiContentGenerator/responseParsingOptions.ts @@ -6,5 +6,6 @@ export interface OpenAIResponseParsingOptions { taggedThinkingTags?: boolean; + taggedThinkingTagsAfterReasoning?: boolean; contentOnlyThinkingTagLeaks?: boolean; } diff --git a/packages/core/src/core/openaiContentGenerator/taggedThinkingParser.ts b/packages/core/src/core/openaiContentGenerator/taggedThinkingParser.ts index b3fc6db83d6..c7d167e7637 100644 --- a/packages/core/src/core/openaiContentGenerator/taggedThinkingParser.ts +++ b/packages/core/src/core/openaiContentGenerator/taggedThinkingParser.ts @@ -65,6 +65,10 @@ export class TaggedThinkingParser { private mode: ParserMode = 'text'; private buffer = ''; + hasUnclosedThought(): boolean { + return this.mode === 'thought'; + } + parse(chunk: string, final = false): Part[] { this.buffer += chunk;