fix(core): support hybrid thinking streams for Qwen3

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

@@ -571,6 +571,391 @@ describe('OpenAIContentConverter', () => {
       );
     });
 
+    // issue #9348: hybrid-thinking models (qwen3-class via OpenAI-compatible
+    // proxies) stream a first thinking phase through reasoning_content and may
+    // then emit a second, properly balanced <thinking>...</thinking> block
+    // inside content. The balanced block is legitimate — rejecting it
+    // hard-fails the whole turn (retries regenerate the same shape), which
+    // surfaced to users as "[API Error: Model response leaked thinking tags.]".
+    it('demotes a balanced inline thinking block to the thought channel when reasoning is present (issue #9348)', () => {
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      const reasoning = converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const opening = converter.convertOpenAIChunkToGemini(
+        streamChunk('opening', { content: '<thinking>' }),
+        stream,
+      );
+      const closing = converter.convertOpenAIChunkToGemini(
+        streamChunk('closing', { content: 'user asked X</thinking>' }),
+        stream,
+      );
+      const answer = converter.convertOpenAIChunkToGemini(
+        streamChunk('answer', { content: 'Answer here.' }, 'stop'),
+        stream,
+      );
+
+      expect(reasoning.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'Let me think.' },
+      ]);
+      expect(opening.candidates?.[0]?.content?.parts).toEqual([]);
+      expect(closing.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'user asked X' },
+      ]);
+      const answerParts = answer.candidates?.[0]?.content?.parts ?? [];
+      expect(answerParts).toEqual([{ text: 'Answer here.' }]);
+      expect(stream.pendingThinkingTagCandidate).toBeUndefined();
+    });
+
+    it('demotes a balanced inline thinking block split across chunks (issue #9348)', () => {
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'phase one' }),
+        stream,
+      );
+      const first = converter.convertOpenAIChunkToGemini(
+        streamChunk('tag-start', { content: '<thi' }),
+        stream,
+      );
+      const second = converter.convertOpenAIChunkToGemini(
+        streamChunk('tag-body', { content: 'nking>recheck</thi' }),
+        stream,
+      );
+      const third = converter.convertOpenAIChunkToGemini(
+        streamChunk('tag-close', { content: 'nking>Done.' }, 'stop'),
+        stream,
+      );
+
+      expect(first.candidates?.[0]?.content?.parts).toEqual([]);
+      expect(second.candidates?.[0]?.content?.parts).toEqual([]);
+      const parts = third.candidates?.[0]?.content?.parts ?? [];
+      expect(parts).toEqual([
+        { thought: true, text: 'recheck' },
+        { text: 'Done.' },
+      ]);
+      expect(stream.pendingThinkingTagCandidate).toBeUndefined();
+    });
+
+    it('demotes a balanced inline thinking block that arrives in a single content chunk (issue #9348)', () => {
+      // Chunking-dependent shape: the first content delta already carries the
+      // complete balanced block, so no pending tag candidate exists yet when
+      // the inline parser first sees the opening tag.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const block = converter.convertOpenAIChunkToGemini(
+        streamChunk('block', {
+          content: '<thinking>user asked X</thinking>',
+        }),
+        stream,
+      );
+      const answer = converter.convertOpenAIChunkToGemini(
+        streamChunk('answer', { content: 'Answer here.' }, 'stop'),
+        stream,
+      );
+
+      expect(block.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'user asked X' },
+      ]);
+      expect(answer.candidates?.[0]?.content?.parts).toEqual([
+        { text: 'Answer here.' },
+      ]);
+      expect(stream.pendingThinkingTagCandidate).toBeUndefined();
+    });
+
+    it('holds an unclosed inline thinking block that starts in the first content chunk (issue #9348)', () => {
+      // Same initial-delta path, but the block does not balance mid-stream:
+      // it must be held for a closing tag, then rejected at stream end.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const opening = converter.convertOpenAIChunkToGemini(
+        streamChunk('opening', { content: '<thinking>user asked X' }),
+        stream,
+      );
+
+      expect(opening.candidates?.[0]?.content?.parts).toEqual([]);
+      expect(stream.pendingThinkingTagCandidate).toEqual({
+        text: '<thinking>user asked X',
+      });
+      expect(() => finishStream(stream, 'stop')).toThrowError(
+        expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }),
+      );
+    });
+
+    it('still rejects an unclosed inline thinking block after reasoning (issue #9348 regression guard)', () => {
+      // The fix must not weaken real leak detection: an opening tag that is
+      // never balanced by a closing tag remains a PROTOCOL_TAG_LEAK.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('opening', { content: '<thinking>' }),
+        stream,
+      );
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('body', { content: 'user asked X' }),
+        stream,
+      );
+
+      expect(() => finishStream(stream, 'stop')).toThrowError(
+        expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }),
+      );
+    });
+
+    it('demotes an empty balanced inline thinking block instead of failing the turn (issue #9348)', () => {
+      // An empty (whitespace-only) balanced block still balances. Treating it
+      // as "no block balanced" re-held the opener and hard-failed at finish
+      // with the exact PROTOCOL_TAG_LEAK error this fix removes.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const answer = converter.convertOpenAIChunkToGemini(
+        streamChunk(
+          'block',
+          { content: '<thinking></thinking>Answer here.' },
+          'stop',
+        ),
+        stream,
+      );
+
+      expect(answer.candidates?.[0]?.content?.parts).toEqual([
+        { text: 'Answer here.' },
+      ]);
+      expect(stream.pendingThinkingTagCandidate).toBeUndefined();
+    });
+
+    it('demotes a whitespace-only balanced inline thinking block split across chunks (issue #9348)', () => {
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const opening = converter.convertOpenAIChunkToGemini(
+        streamChunk('opening', { content: '<thinking>\n' }),
+        stream,
+      );
+      const answer = converter.convertOpenAIChunkToGemini(
+        streamChunk('closing', { content: '</thinking>Answer here.' }, 'stop'),
+        stream,
+      );
+
+      expect(opening.candidates?.[0]?.content?.parts).toEqual([]);
+      expect(answer.candidates?.[0]?.content?.parts).toEqual([
+        { text: 'Answer here.' },
+      ]);
+    });
+
+    it('demotes consecutive balanced inline thinking blocks (issue #9348)', () => {
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const blocks = converter.convertOpenAIChunkToGemini(
+        streamChunk(
+          'blocks',
+          { content: '<thinking>a</thinking><thinking>b</thinking>Done.' },
+          'stop',
+        ),
+        stream,
+      );
+
+      expect(blocks.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'a' },
+        { thought: true, text: 'b' },
+        { text: 'Done.' },
+      ]);
+    });
+
+    it('holds a trailing tag prefix after a demoted block until it balances (issue #9348)', () => {
+      // The only branch that keeps a trailing partial tag held after a
+      // successful demotion: it must stay held mid-stream and demote once the
+      // second block balances.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const first = converter.convertOpenAIChunkToGemini(
+        streamChunk('first-block', { content: '<thinking>a</thinking><thi' }),
+        stream,
+      );
+
+      expect(first.candidates?.[0]?.content?.parts).toEqual([]);
+      expect(stream.pendingThinkingTagCandidate).toEqual({ text: '<thi' });
+
+      const second = converter.convertOpenAIChunkToGemini(
+        streamChunk(
+          'second-block',
+          { content: 'nking>b</thinking>Done.' },
+          'stop',
+        ),
+        stream,
+      );
+
+      expect(second.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'a' },
+        { thought: true, text: 'b' },
+        { text: 'Done.' },
+      ]);
+    });
+
+    it('rejects an embedded thinking block after a demoted leading block (issue #9348 fail-closed)', () => {
+      // The demotion mechanism exists to stop raw thinking tags from reaching
+      // visible output: a complete tag inside the tail released after a
+      // demotion is embedded/stray and must fail closed like pre-PR.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+
+      expect(() =>
+        converter.convertOpenAIChunkToGemini(
+          streamChunk(
+            'embedded',
+            {
+              content:
+                '<thinking>a</thinking>Answer <thinking>b</thinking> done',
+            },
+            'stop',
+          ),
+          stream,
+        ),
+      ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }));
+    });
+
+    it('rejects embedded thinking tags arriving after visible content was already emitted (issue #9348 fail-closed)', () => {
+      // Chunk-split twin of the interleaved shape: once visible content
+      // exists the candidate machinery no longer applies, so the later chunk
+      // needs its own fail-closed gate.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const first = converter.convertOpenAIChunkToGemini(
+        streamChunk('leading-block', {
+          content: '<thinking>a</thinking>Answer ',
+        }),
+        stream,
+      );
+
+      expect(first.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'a' },
+        { text: 'Answer ' },
+      ]);
+      expect(() =>
+        converter.convertOpenAIChunkToGemini(
+          streamChunk(
+            'embedded',
+            { content: '<thinking>b</thinking> done' },
+            'stop',
+          ),
+          stream,
+        ),
+      ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }));
+    });
+
+    it('rejects a stray closing tag after a demoted leading block (issue #9348 fail-closed)', () => {
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+
+      expect(() =>
+        converter.convertOpenAIChunkToGemini(
+          streamChunk(
+            'stray-closer',
+            { content: '<thinking>a</thinking>b</thinking>' },
+            'stop',
+          ),
+          stream,
+        ),
+      ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }));
+    });
+
+    it('rejects a truncated tag prefix after a balanced block at stream finish (issue #9348 fail-closed)', () => {
+      // '<thi' classifies as pending, never leaked, and the finish-time
+      // leading-tag check needs a complete opener — without a dedicated
+      // fail-closed check the fragment would be released as visible text
+      // while the chunk-split twin throws.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+
+      expect(() =>
+        converter.convertOpenAIChunkToGemini(
+          streamChunk(
+            'truncated-prefix',
+            { content: '<thinking>a</thinking><thi' },
+            'stop',
+          ),
+          stream,
+        ),
+      ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }));
+    });
+
+    it('rejects a split truncated tag prefix after a balanced block at stream finish (issue #9348 fail-closed)', () => {
+      // The tail re-held after a demotion must fail closed at finish instead
+      // of being released as a literal by the undecided-prefix release path.
+      const stream = withStreamParser();
+      stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };
+
+      converter.convertOpenAIChunkToGemini(
+        streamChunk('reasoning', { reasoning_content: 'Let me think.' }),
+        stream,
+      );
+      const first = converter.convertOpenAIChunkToGemini(
+        streamChunk('first-block', { content: '<thinking>a</thinking><thi' }),
+        stream,
+      );
+
+      expect(first.candidates?.[0]?.content?.parts).toEqual([]);
+      expect(() => finishStream(stream, 'stop')).toThrowError(
+        expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' }),
+      );
+    });
+
     it('holds a long confirmed opening tag until its closing tag arrives', () => {
       const stream = withStreamParser();
       stream.responseParsingOptions = { contentOnlyThinkingTagLeaks: true };