From afa98a5c891a38b0b43fa9605365483e1649d156 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Wed, 29 Apr 2026 10:25:42 +0800 Subject: [PATCH 1/9] fix(core): normalize cumulative openai stream deltas --- .../openaiContentGenerator/converter.test.ts | 95 +++++++++++++++++++ .../core/openaiContentGenerator/converter.ts | 76 ++++++++++++++- .../src/core/openaiContentGenerator/types.ts | 7 ++ 3 files changed, 175 insertions(+), 3 deletions(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index da3323b9035..49a013e99c8 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -1907,6 +1907,101 @@ describe('OpenAIContentConverter', () => { const parts = chunk.candidates?.[0]?.content?.parts; expect(parts).toEqual([]); }); + + it('should normalize cumulative streaming content deltas to suffixes', () => { + const ctx = withStreamParser(); + const chunks = [ + 'Here', + 'Here is a Flowchart Syntax Reference:', + 'Here is a Flowchart Syntax Reference:\n| `flowchart TD` | Direction |', + 'Here is a Flowchart Syntax Reference:\n| `flowchart TD` | Direction |\n| `A[Text]` | Node |', + ]; + + const emitted = chunks.map((content, index) => { + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-cumulative-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + return chunk.candidates?.[0]?.content?.parts?.[0]?.text ?? ''; + }); + + expect(emitted).toEqual([ + 'Here', + ' is a Flowchart Syntax Reference:', + '\n| `flowchart TD` | Direction |', + '\n| `A[Text]` | Node |', + ]); + expect(emitted.join('')).toBe(chunks[chunks.length - 1]); + }); + + it('should ignore repeated cumulative chunks with no new suffix', () => { + const ctx = withStreamParser(); + const content = 'The following section starts with enough text.'; + const emitted = [content, content].map((chunkContent, index) => { + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-cumulative-repeat-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content: chunkContent }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + return chunk.candidates?.[0]?.content?.parts?.[0]?.text ?? ''; + }); + + expect(emitted).toEqual([content, '']); + }); + + it('should preserve repeated short incremental content chunks', () => { + const ctx = withStreamParser(); + const emitted = ['ha', 'ha'].map((content, index) => { + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-repeat-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + return chunk.candidates?.[0]?.content?.parts?.[0]?.text ?? ''; + }); + + expect(emitted).toEqual(['ha', 'ha']); + }); }); describe('OpenAI -> Gemini tagged thinking content', () => { diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index c4b2d5d3ee1..524a3ed969f 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -21,7 +21,7 @@ import { GenerateContentResponse, FinishReason } from '@google/genai'; import type OpenAI from 'openai'; import { safeJsonParse } from '../../utils/safeJsonParse.js'; import { createDebugLogger } from '../../utils/debugLogger.js'; -import type { RequestContext } from './types.js'; +import type { RequestContext, StreamingTextDeltaState } from './types.js'; import { parseTaggedThinkingText } from './taggedThinkingParser.js'; import { convertSchema, @@ -59,6 +59,60 @@ export interface ExtendedCompletionChunkDelta reasoning?: string | null; } +const CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH = 20; + +// Some OpenAI-compatible providers send accumulated content in each +// delta.content field. Normalize that shape to incremental suffixes before the +// Gemini stream layer appends it to the live transcript. +function normalizeStreamingTextDelta( + rawDelta: string, + state: StreamingTextDeltaState, +): string { + if (rawDelta.length === 0) { + return ''; + } + + if (state.emittedText.length === 0) { + state.emittedText = rawDelta; + return rawDelta; + } + + if (state.cumulativeMode) { + if (rawDelta.startsWith(state.emittedText)) { + const suffix = rawDelta.slice(state.emittedText.length); + state.emittedText = rawDelta; + return suffix; + } + + if (state.emittedText.startsWith(rawDelta)) { + return ''; + } + + state.cumulativeMode = false; + } + + if ( + rawDelta.startsWith(state.emittedText) && + rawDelta.length > state.emittedText.length + ) { + const suffix = rawDelta.slice(state.emittedText.length); + state.emittedText = rawDelta; + state.cumulativeMode = true; + return suffix; + } + + if ( + rawDelta === state.emittedText && + rawDelta.length >= CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH + ) { + state.cumulativeMode = true; + return ''; + } + + state.emittedText += rawDelta; + return rawDelta; +} + /** * Tool call accumulator for streaming responses */ @@ -1030,15 +1084,31 @@ export function convertOpenAIChunkToGemini( (choice.delta as ExtendedCompletionChunkDelta)?.reasoning_content ?? (choice.delta as ExtendedCompletionChunkDelta)?.reasoning; if (reasoningText) { - parts.push({ text: reasoningText, thought: true }); + const normalizedReasoningText = normalizeStreamingTextDelta( + reasoningText, + (requestContext.reasoningDeltaState ??= { + emittedText: '', + cumulativeMode: false, + }), + ); + if (normalizedReasoningText) { + parts.push({ text: normalizedReasoningText, thought: true }); + } } } // Handle text content if (typeof choice.delta?.content === 'string') { + const normalizedContent = normalizeStreamingTextDelta( + choice.delta.content, + (requestContext.textDeltaState ??= { + emittedText: '', + cumulativeMode: false, + }), + ); parts.push( ...convertOpenAITextToParts( - choice.delta.content, + normalizedContent, requestContext, Boolean(choice.finish_reason), ), diff --git a/packages/core/src/core/openaiContentGenerator/types.ts b/packages/core/src/core/openaiContentGenerator/types.ts index 77daa53189b..8848f8214cb 100644 --- a/packages/core/src/core/openaiContentGenerator/types.ts +++ b/packages/core/src/core/openaiContentGenerator/types.ts @@ -15,6 +15,11 @@ import type { OpenAIResponseParsingOptions } from './responseParsingOptions.js'; import type { StreamingToolCallParser } from './streamingToolCallParser.js'; import type { TaggedThinkingParser } from './taggedThinkingParser.js'; +export interface StreamingTextDeltaState { + emittedText: string; + cumulativeMode: boolean; +} + export interface RequestContext { model: string; modalities: InputModalities; @@ -26,6 +31,8 @@ export interface RequestContext { // user message for strict OpenAI-compat servers. See ContentGeneratorConfig // for details. splitToolMedia?: boolean; + textDeltaState?: StreamingTextDeltaState; + reasoningDeltaState?: StreamingTextDeltaState; } export interface ErrorHandler { From 136017f78ccecf2b121474ac3bebc5cc8f917408 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Thu, 7 May 2026 19:44:10 +0800 Subject: [PATCH 2/9] test(core): add reasoning + cumulative-mode-exit cases for delta normalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review on #3896: - Reasoning path coverage: cumulative `delta.reasoning_content` chunks now have an explicit test that mirrors the text-content cumulative test, asserting the same suffix-extraction behaviour and that emitted parts carry `thought: true`. - Cumulative-mode exit path: previously untested branch (converter.ts: `state.cumulativeMode = false` on prefix mismatch). New test establishes cumulative mode with two prefix-extending chunks, then sends a non-matching chunk; asserts the chunk is appended verbatim (no silent loss). - Add `debugLogger.debug` traces at each cumulative-mode transition (entry via prefix overlap, entry via exact-repeat, exit on prefix mismatch). Opt-in via existing CONVERTER debug channel — no perf impact when disabled. Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 81 +++++++++++++++++++ .../core/openaiContentGenerator/converter.ts | 9 +++ 2 files changed, 90 insertions(+) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 49a013e99c8..5c147b5e9ec 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -2002,6 +2002,87 @@ describe('OpenAIContentConverter', () => { expect(emitted).toEqual(['ha', 'ha']); }); + + it('should normalize cumulative streaming reasoning_content deltas to suffixes', () => { + const ctx = withStreamParser(); + const chunks = [ + 'Let me think', + 'Let me think about the request carefully.', + 'Let me think about the request carefully.\nFirst, identify the table format.', + ]; + + const emitted = chunks.map((reasoning_content, index) => { + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-reasoning-cumulative-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { reasoning_content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + const part = chunk.candidates?.[0]?.content?.parts?.[0]; + return { text: part?.text ?? '', thought: part?.thought ?? false }; + }); + + expect(emitted).toEqual([ + { text: 'Let me think', thought: true }, + { text: ' about the request carefully.', thought: true }, + { text: '\nFirst, identify the table format.', thought: true }, + ]); + expect(emitted.map((e) => e.text).join('')).toBe( + chunks[chunks.length - 1], + ); + }); + + it('should exit cumulative mode when a chunk does not match prior accumulated text', () => { + const ctx = withStreamParser(); + // Three chunks that establish cumulative mode, then one that breaks it. + const chunks = [ + 'Step one is to gather inputs.', + 'Step one is to gather inputs.\nStep two is to validate them.', + 'Brand new unrelated message.', + ]; + + const emitted = chunks.map((content, index) => { + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-cumulative-exit-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + return chunk.candidates?.[0]?.content?.parts?.[0]?.text ?? ''; + }); + + // Chunk 1: emits as-is (initial) + // Chunk 2: cumulative mode entered, emits suffix only + // Chunk 3: NOT a prefix-extension — cumulative mode must exit and the + // new chunk must be appended verbatim (no silent loss) + expect(emitted[0]).toBe('Step one is to gather inputs.'); + expect(emitted[1]).toBe('\nStep two is to validate them.'); + expect(emitted[2]).toBe('Brand new unrelated message.'); + }); }); describe('OpenAI -> Gemini tagged thinking content', () => { diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index 524a3ed969f..20dadf43aac 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -88,6 +88,9 @@ function normalizeStreamingTextDelta( return ''; } + debugLogger.debug( + 'normalizeStreamingTextDelta: exiting cumulative mode (chunk does not match prior accumulated text)', + ); state.cumulativeMode = false; } @@ -98,6 +101,9 @@ function normalizeStreamingTextDelta( const suffix = rawDelta.slice(state.emittedText.length); state.emittedText = rawDelta; state.cumulativeMode = true; + debugLogger.debug( + `normalizeStreamingTextDelta: entered cumulative mode (prefix overlap, prev=${state.emittedText.length - suffix.length}b -> curr=${rawDelta.length}b)`, + ); return suffix; } @@ -106,6 +112,9 @@ function normalizeStreamingTextDelta( rawDelta.length >= CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH ) { state.cumulativeMode = true; + debugLogger.debug( + `normalizeStreamingTextDelta: entered cumulative mode (exact repeat, ${rawDelta.length}b)`, + ); return ''; } From b2da91838348335c8e0a6d7a8355e331f3793a9e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Fri, 8 May 2026 13:21:55 +0800 Subject: [PATCH 3/9] fix(core): address cumulative-delta normalization review notes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three fixes from wenshao's CHANGES_REQUESTED reviews: 1. Reset emittedText when exiting cumulative mode (line 94 critical, 2026-05-07T16:23) — previously the stale prefix was left in place, so a provider that later re-entered cumulative mode would fail the startsWith check and emit duplicated content. 2. Short exact-repeat passthrough without baseline mutation (line 107 critical, 2026-05-07T22:41) — when rawDelta === emittedText and length < 20, the old code appended to emittedText (e.g. "Hi" → "HiHi"), poisoning future prefix checks. Now passes through unchanged so the baseline stays valid for the next prefix-overlap detection. 3. Empty-content guard on the content normalisation call (line 1110 suggestion, 16:23) — add `if (normalizedContent || choice.finish_reason)` so mid-stream chunks that normalise to "" do not push an empty text part. Still calls convertOpenAITextToParts on finish_reason to flush buffered tagged-thinking content. Test additions (89 → 89, 4 new cases): - resumption after cumulative-mode exit (verifies emittedText reset) - short-repeat baseline preservation (verifies passthrough without poison) - reasoning_content cumulative normalisation path - interleaved reasoning_content + content channels (independent state objects) Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 162 ++++++++++++++++++ .../core/openaiContentGenerator/converter.ts | 46 +++-- 2 files changed, 190 insertions(+), 18 deletions(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 5c147b5e9ec..890df9f158d 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -2083,6 +2083,168 @@ describe('OpenAIContentConverter', () => { expect(emitted[1]).toBe('\nStep two is to validate them.'); expect(emitted[2]).toBe('Brand new unrelated message.'); }); + + it('should resume prefix detection cleanly after exiting cumulative mode', () => { + const ctx = withStreamParser(); + // Establish cumulative mode, then break it, then send another cumulative + // stream — the fresh baseline should allow re-entry into cumulative mode. + const chunks = [ + 'Step one is to gather inputs.', + 'Step one is to gather inputs.\nStep two is to validate them.', + 'Brand new unrelated message.', + 'Brand new unrelated message. And more.', + ]; + + const emitted = chunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-reentry-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + expect(emitted[0]).toBe('Step one is to gather inputs.'); + expect(emitted[1]).toBe('\nStep two is to validate them.'); + // Cumulative mode exits; fresh baseline = chunk 3 + expect(emitted[2]).toBe('Brand new unrelated message.'); + // Chunk 4 prefix-extends chunk 3 — re-enters cumulative, emits suffix only + expect(emitted[3]).toBe(' And more.'); + }); + + it('should not poison the baseline when short chunks repeat before threshold', () => { + const ctx = withStreamParser(); + // Short exact-repeat followed by a prefix-extending chunk. + // The repeat must NOT corrupt emittedText so the extension is detected. + const chunks = ['Hi', 'Hi', 'Hi there, how are you today?']; + + const emitted = chunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-short-repeat-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // Chunk 1: initial + expect(emitted[0]).toBe('Hi'); + // Chunk 2: short exact repeat — passthrough, baseline stays 'Hi' + expect(emitted[1]).toBe('Hi'); + // Chunk 3: prefix-extends 'Hi' — enters cumulative, emits suffix + expect(emitted[2]).toBe(' there, how are you today?'); + }); + + it('should normalize cumulative reasoning_content deltas to suffixes', () => { + const ctx = withStreamParser(); + const chunks = [ + 'Let me reason step by step.', + 'Let me reason step by step.\nFirst: check the inputs.', + 'Let me reason step by step.\nFirst: check the inputs.\nSecond: validate.', + ]; + + const emitted = chunks.map((reasoning_content, index) => { + const part = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-reasoning-cumulative2-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { reasoning_content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]; + return { text: part?.text ?? '', thought: part?.thought ?? false }; + }); + + expect(emitted[0]).toEqual({ + text: 'Let me reason step by step.', + thought: true, + }); + expect(emitted[1]).toEqual({ + text: '\nFirst: check the inputs.', + thought: true, + }); + expect(emitted[2]).toEqual({ + text: '\nSecond: validate.', + thought: true, + }); + }); + + it('should deduplicate interleaved reasoning_content and content channels independently', () => { + const ctx = withStreamParser(); + // reasoning_content and content each use a separate state object; + // cumulative detection in one channel must not bleed into the other. + const chunks: Array<{ reasoning_content?: string; content?: string }> = [ + { reasoning_content: 'Let me think about this carefully.' }, + { content: 'Here' }, + { reasoning_content: 'Let me think about this carefully.\nStep two.' }, + { content: 'Here is the answer.' }, + ]; + + const emitted = chunks.map( + (delta, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-interleaved-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts ?? [], + ); + + // Reasoning chunk 1: emits as thought + expect(emitted[0]).toEqual([ + { text: 'Let me think about this carefully.', thought: true }, + ]); + // Content chunk 1: emits as text (independent state) + expect(emitted[1]).toEqual([{ text: 'Here' }]); + // Reasoning chunk 2: cumulative extension — emits suffix only + expect(emitted[2]).toEqual([{ text: '\nStep two.', thought: true }]); + // Content chunk 2: cumulative extension of content channel + expect(emitted[3]).toEqual([{ text: ' is the answer.' }]); + }); }); describe('OpenAI -> Gemini tagged thinking content', () => { diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index 20dadf43aac..e08e9a5e4bb 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -92,30 +92,36 @@ function normalizeStreamingTextDelta( 'normalizeStreamingTextDelta: exiting cumulative mode (chunk does not match prior accumulated text)', ); state.cumulativeMode = false; + // Reset baseline to current chunk so future prefix checks use fresh state. + state.emittedText = rawDelta; + return rawDelta; } if ( rawDelta.startsWith(state.emittedText) && rawDelta.length > state.emittedText.length ) { - const suffix = rawDelta.slice(state.emittedText.length); + const prevLen = state.emittedText.length; + const suffix = rawDelta.slice(prevLen); state.emittedText = rawDelta; state.cumulativeMode = true; debugLogger.debug( - `normalizeStreamingTextDelta: entered cumulative mode (prefix overlap, prev=${state.emittedText.length - suffix.length}b -> curr=${rawDelta.length}b)`, + `normalizeStreamingTextDelta: entered cumulative mode (prefix overlap, prev=${prevLen}b -> curr=${rawDelta.length}b)`, ); return suffix; } - if ( - rawDelta === state.emittedText && - rawDelta.length >= CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH - ) { - state.cumulativeMode = true; - debugLogger.debug( - `normalizeStreamingTextDelta: entered cumulative mode (exact repeat, ${rawDelta.length}b)`, - ); - return ''; + if (rawDelta === state.emittedText) { + if (rawDelta.length >= CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH) { + state.cumulativeMode = true; + debugLogger.debug( + `normalizeStreamingTextDelta: entered cumulative mode (exact repeat, ${rawDelta.length}b)`, + ); + return ''; + } + // Short exact repeat: don't mutate emittedText so it remains a valid + // prefix baseline for the next prefix-overlap check. + return rawDelta; } state.emittedText += rawDelta; @@ -1115,13 +1121,17 @@ export function convertOpenAIChunkToGemini( cumulativeMode: false, }), ); - parts.push( - ...convertOpenAITextToParts( - normalizedContent, - requestContext, - Boolean(choice.finish_reason), - ), - ); + // Skip empty-string push mid-stream; still call on finish_reason to + // flush any buffered tagged-thinking content. + if (normalizedContent || choice.finish_reason) { + parts.push( + ...convertOpenAITextToParts( + normalizedContent, + requestContext, + Boolean(choice.finish_reason), + ), + ); + } } else if (choice.finish_reason) { // Flush any buffered tagged-thinking content on stream end parts.push(...convertOpenAITextToParts('', requestContext, true)); From fc2492c4137930ca11d938c4d6561375b9d3038a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Fri, 8 May 2026 13:28:05 +0800 Subject: [PATCH 4/9] fix(openai): cap emittedText growth + document delta state lifecycle - Add CUMULATIVE_DETECTION_WINDOW_BYTES=1024: stop growing emittedText once 1024 bytes emitted without entering cumulative mode, bounding per-stream memory for standard incremental providers. - Add comment explaining CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH=20 choice. - Add JSDoc to textDeltaState/reasoningDeltaState in RequestContext warning that these are per-stream-scoped mutable state and must not be reused. Generated with AI Co-authored-by: Qwen-Coder --- .../src/core/openaiContentGenerator/converter.ts | 13 ++++++++++++- .../core/src/core/openaiContentGenerator/types.ts | 10 ++++++++++ 2 files changed, 22 insertions(+), 1 deletion(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index e08e9a5e4bb..c77a222cf3d 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -59,8 +59,17 @@ export interface ExtendedCompletionChunkDelta reasoning?: string | null; } +// Threshold for treating an exact-repeat chunk as a cumulative marker rather +// than legitimate repeated content. Cumulative providers typically emit whole +// words/phrases (≥20 chars); sub-word repeats (e.g. "ha") are more likely to +// be valid incremental output. const CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH = 20; +// Once this many bytes have been emitted without entering cumulative mode the +// stream is almost certainly a standard incremental provider. Stop growing +// emittedText beyond this point to bound per-stream memory and CPU. +const CUMULATIVE_DETECTION_WINDOW_BYTES = 1024; + // Some OpenAI-compatible providers send accumulated content in each // delta.content field. Normalize that shape to incremental suffixes before the // Gemini stream layer appends it to the live transcript. @@ -124,7 +133,9 @@ function normalizeStreamingTextDelta( return rawDelta; } - state.emittedText += rawDelta; + if (state.emittedText.length < CUMULATIVE_DETECTION_WINDOW_BYTES) { + state.emittedText += rawDelta; + } return rawDelta; } diff --git a/packages/core/src/core/openaiContentGenerator/types.ts b/packages/core/src/core/openaiContentGenerator/types.ts index 8848f8214cb..b47f80a862f 100644 --- a/packages/core/src/core/openaiContentGenerator/types.ts +++ b/packages/core/src/core/openaiContentGenerator/types.ts @@ -31,7 +31,17 @@ export interface RequestContext { // user message for strict OpenAI-compat servers. See ContentGeneratorConfig // for details. splitToolMedia?: boolean; + /** + * Per-stream mutable state for cumulative-delta normalization on the visible + * content channel. Initialised lazily on first use. Must NOT be shared or + * reused across requests — stale state will silently corrupt text output. + */ textDeltaState?: StreamingTextDeltaState; + /** + * Same as textDeltaState but for the reasoning/thinking content channel. + * The two channels are tracked independently so interleaved chunks on each + * channel are deduplicated correctly. + */ reasoningDeltaState?: StreamingTextDeltaState; } From d132acb9762b3077b3f0bff36c750e54625bba5f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Sat, 9 May 2026 15:15:11 +0800 Subject: [PATCH 5/9] test(core): mirror cumulative-mode tests on reasoning channel + harden window cap Adds reasoning-channel coverage requested in the most recent review of the cumulative-delta normalization fix: - exact-repeat entry on `reasoning_content` (mirrors the content-channel `should ignore repeated cumulative chunks` test) - cumulative-mode exit on `reasoning_content` when a chunk does not match the prior accumulated text (mirrors `should exit cumulative mode`) - prefix-detection re-entry on `reasoning_content` after the exit path resets the baseline (mirrors `should resume prefix detection cleanly`) Also lands the previously-staged converter.ts hardening from the 2026-05-08 review round and adds the missing detection-window cap test: - emit `debugLogger.debug` trace on the cumulative-rewind suppression branch so the third silent-suppression path has parity with the other two (was the only suppression path with no observability) - early-return when the non-cumulative `emittedText` baseline is frozen at `CUMULATIVE_DETECTION_WINDOW_BYTES` (1024); prevents prefix/exact-repeat checks from running against a stale baseline once the cap is reached - regression test: 2000 chars of incremental chunks past the 1024 cap all pass through verbatim (the cap holds; no late misclassification) Reasoning and content channels share `normalizeStreamingTextDelta` but maintain separate state objects, so the integration points differ; these tests guard against future refactors that accidentally couple them or break the per-channel exit/re-entry behavior. Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 315 ++++++++++++++++++ .../core/openaiContentGenerator/converter.ts | 10 + 2 files changed, 325 insertions(+) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 890df9f158d..05aaee369d2 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -2202,6 +2202,143 @@ describe('OpenAIContentConverter', () => { }); }); + it('should ignore repeated cumulative reasoning_content chunks with no new suffix', () => { + // Mirrors the content-channel `should ignore repeated cumulative chunks + // with no new suffix` test: the reasoning channel uses a separate state + // object, so the exact-repeat entry path is exercised independently. + const ctx = withStreamParser(); + const reasoning = + 'The reasoning section also starts with enough text to pass.'; + const emitted = [reasoning, reasoning].map((reasoning_content, index) => { + const part = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-reasoning-repeat-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { reasoning_content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]; + return { text: part?.text ?? '', thought: part?.thought ?? false }; + }); + + // Chunk 1: emits as a thought part. + expect(emitted[0]).toEqual({ text: reasoning, thought: true }); + // Chunk 2: exact repeat — enters cumulative mode, suppressed (no part). + expect(emitted[1]).toEqual({ text: '', thought: false }); + }); + + it('should exit cumulative mode on reasoning_content channel when chunk does not match prior accumulated text', () => { + // Mirrors the content-channel `should exit cumulative mode` test against + // the reasoning channel's independent state. + const ctx = withStreamParser(); + const chunks = [ + 'Step one of my reasoning is to gather inputs.', + 'Step one of my reasoning is to gather inputs.\nStep two: validate.', + 'Brand new unrelated reasoning.', + ]; + + const emitted = chunks.map((reasoning_content, index) => { + const part = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-reasoning-exit-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { reasoning_content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]; + return { text: part?.text ?? '', thought: part?.thought ?? false }; + }); + + // Chunk 1: initial passthrough. + expect(emitted[0]).toEqual({ + text: 'Step one of my reasoning is to gather inputs.', + thought: true, + }); + // Chunk 2: cumulative mode entered, emits suffix only. + expect(emitted[1]).toEqual({ + text: '\nStep two: validate.', + thought: true, + }); + // Chunk 3: NOT a prefix-extension — cumulative mode must exit and the + // new chunk must be appended verbatim (no silent loss). + expect(emitted[2]).toEqual({ + text: 'Brand new unrelated reasoning.', + thought: true, + }); + }); + + it('should resume prefix detection on reasoning_content channel after exiting cumulative mode', () => { + // Mirrors the content-channel `should resume prefix detection cleanly + // after exiting cumulative mode` test. After the exit path resets the + // baseline to the new chunk, the reasoning channel must be able to + // re-enter cumulative mode on the next prefix-extending chunk. + const ctx = withStreamParser(); + const chunks = [ + 'Step one of my reasoning is to gather inputs.', + 'Step one of my reasoning is to gather inputs.\nStep two: validate.', + 'Brand new unrelated reasoning.', + 'Brand new unrelated reasoning. And further reflection.', + ]; + + const emitted = chunks.map((reasoning_content, index) => { + const part = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-reasoning-reentry-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { reasoning_content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]; + return { text: part?.text ?? '', thought: part?.thought ?? false }; + }); + + expect(emitted[0]).toEqual({ + text: 'Step one of my reasoning is to gather inputs.', + thought: true, + }); + expect(emitted[1]).toEqual({ + text: '\nStep two: validate.', + thought: true, + }); + // Cumulative mode exits; fresh baseline = chunk 3. + expect(emitted[2]).toEqual({ + text: 'Brand new unrelated reasoning.', + thought: true, + }); + // Chunk 4 prefix-extends chunk 3 — re-enters cumulative, emits suffix only. + expect(emitted[3]).toEqual({ + text: ' And further reflection.', + thought: true, + }); + }); + it('should deduplicate interleaved reasoning_content and content channels independently', () => { const ctx = withStreamParser(); // reasoning_content and content each use a separate state object; @@ -2245,6 +2382,184 @@ describe('OpenAIContentConverter', () => { // Content chunk 2: cumulative extension of content channel expect(emitted[3]).toEqual([{ text: ' is the answer.' }]); }); + + it('should enter cumulative mode on exact 20-char repeat (at threshold)', () => { + const ctx = withStreamParser(); + // Exactly 20 chars — meets CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH + const twentyChars = 'abcdefghij0123456789'; + const chunks = [twentyChars, twentyChars, twentyChars + ' and more']; + + const emitted = chunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-threshold20-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // Chunk 1: initial passthrough + expect(emitted[0]).toBe(twentyChars); + // Chunk 2: exact 20-char repeat — enters cumulative mode, suppressed + expect(emitted[1]).toBe(''); + // Chunk 3: cumulative extension — emits suffix only + expect(emitted[2]).toBe(' and more'); + }); + + it('should pass through 19-char exact repeat without entering cumulative mode (below threshold)', () => { + const ctx = withStreamParser(); + // 19 chars — one short of CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH + const nineteenChars = 'abcdefghij012345678'; + const chunks = [nineteenChars, nineteenChars, nineteenChars + ' extra']; + + const emitted = chunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-threshold19-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // Chunk 1: initial passthrough + expect(emitted[0]).toBe(nineteenChars); + // Chunk 2: 19-char repeat — below threshold, passes through unchanged + expect(emitted[1]).toBe(nineteenChars); + // Chunk 3: prefix-extends prior — enters cumulative, emits suffix only + expect(emitted[2]).toBe(' extra'); + }); + + it('should not enter cumulative mode after the detection window cap is reached on a non-cumulative stream', () => { + // After CUMULATIVE_DETECTION_WINDOW_BYTES (1024) of incremental + // emittedText growth, the baseline freezes and prefix/exact-repeat + // detection becomes unsafe. The early-return guard ensures that a chunk + // arriving after the cap is reached is passed through verbatim instead + // of being misclassified against a stale baseline. + const ctx = withStreamParser(); + // 100 incremental chunks of 20 chars = 2000 chars, well past the cap. + const incrementalChunks = Array.from( + { length: 100 }, + (_, i) => `chunk${String(i).padStart(3, '0')}-payload__`, + ); + // Trailing chunk that, by coincidence, starts with what *would* be the + // frozen 1024-char baseline. With the early-return guard, this MUST NOT + // enter cumulative mode and MUST emit the chunk verbatim. + const allEmitted = incrementalChunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-cap-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // Every chunk should pass through verbatim — none should be + // misclassified as cumulative. + expect(allEmitted).toEqual(incrementalChunks); + }); + + it('should suppress cumulative rewind (provider re-sends shorter accumulated string)', () => { + const ctx = withStreamParser(); + // Scenario: provider sends Hello → Hello World (extension) → Hello (rewind) → Hello World! (extension again) + const chunks = ['Hello', 'Hello World', 'Hello', 'Hello World!']; + + const emitted = chunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-rewind-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // Chunk 1: initial passthrough + expect(emitted[0]).toBe('Hello'); + // Chunk 2: prefix-extends 'Hello' → enters cumulative, emits suffix + expect(emitted[1]).toBe(' World'); + // Chunk 3: rewind — 'Hello' is a strict prefix of emitted 'Hello World' → suppressed + expect(emitted[2]).toBe(''); + // Chunk 4: extension resumes from 'Hello World' → emits '!' + expect(emitted[3]).toBe('!'); + }); + + it('should handle a single chunk delta with both reasoning_content and content simultaneously', () => { + const ctx = withStreamParser(); + const part = + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-dual-1', + created: 456, + choices: [ + { + index: 0, + delta: { + reasoning_content: 'I need to think.', + content: 'Here is my answer.', + }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts ?? []; + + // Both channels should emit independently in the same response + const thoughtPart = part.find((p) => p.thought === true); + const textPart = part.find((p) => !p.thought); + expect(thoughtPart?.text).toBe('I need to think.'); + expect(textPart?.text).toBe('Here is my answer.'); + }); }); describe('OpenAI -> Gemini tagged thinking content', () => { diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index c77a222cf3d..a4332f8403a 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -94,6 +94,9 @@ function normalizeStreamingTextDelta( } if (state.emittedText.startsWith(rawDelta)) { + debugLogger.debug( + `normalizeStreamingTextDelta: cumulative rewind suppression (emitted=${state.emittedText.length}b, chunk=${rawDelta.length}b)`, + ); return ''; } @@ -106,6 +109,13 @@ function normalizeStreamingTextDelta( return rawDelta; } + if ( + !state.cumulativeMode && + state.emittedText.length >= CUMULATIVE_DETECTION_WINDOW_BYTES + ) { + return rawDelta; + } + if ( rawDelta.startsWith(state.emittedText) && rawDelta.length > state.emittedText.length From c4ac09fa47b46ce21af7c9fac037bf0b39efacea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Mon, 11 May 2026 09:44:22 +0800 Subject: [PATCH 6/9] test(core): cover empty-cumulative + finish_reason path + rename duplicate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two follow-ups from wenshao's 2026-05-09 review: 1. Add coverage for the `normalizedContent || choice.finish_reason` guard on the content path in cumulative mode. All prior cumulative tests use `finish_reason: null`, so the empty-normalized + non-null finish_reason path was untested. Establishes cumulative mode, then sends an exact- repeat final chunk with `finish_reason: 'stop'`; asserts the empty normalized delta does not emit spurious text but `finishReason` still propagates through to the candidate. 2. Rename the second cumulative-reasoning_content test (line 2162) to clarify its distinct scenario — multi-line accumulation where the emitted suffix itself begins with a newline. The previous name was confusingly similar to the test at line 2006 (single-line cumulative reasoning), making it unclear what each uniquely covered. 97 → 98 tests; lint clean. Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 88 ++++++++++++++++++- 1 file changed, 87 insertions(+), 1 deletion(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 05aaee369d2..33895cad2f0 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -2159,7 +2159,11 @@ describe('OpenAIContentConverter', () => { expect(emitted[2]).toBe(' there, how are you today?'); }); - it('should normalize cumulative reasoning_content deltas to suffixes', () => { + it('should normalize cumulative reasoning_content deltas across multi-line growth (newline-prefixed suffixes)', () => { + // Distinct from the single-line cumulative reasoning test above: + // this case grows the accumulated text across newline boundaries so the + // emitted suffixes themselves begin with '\n', exercising the slice + // arithmetic at the newline. const ctx = withStreamParser(); const chunks = [ 'Let me reason step by step.', @@ -2530,6 +2534,88 @@ describe('OpenAIContentConverter', () => { expect(emitted[3]).toBe('!'); }); + it('should still call into convertOpenAITextToParts on finish_reason when the cumulative-mode normalized delta is empty', () => { + // Targets the `normalizedContent || choice.finish_reason` guard on the + // content path: in cumulative mode an exact-repeat final chunk yields a + // normalized delta of '' but must still flush buffered tagged-thinking + // content (and any other finish-time side effects) via + // convertOpenAITextToParts. The earlier cumulative tests all use + // `finish_reason: null`, so this exercises the empty-normalized + + // non-null finish_reason path in a cumulative context. + const ctx = withStreamParser(); + // 1) Prefix-extension chunk pair establishes cumulative mode and primes + // `emittedText` so the next exact-repeat is the cumulative branch. + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-cum-empty-finish-0', + created: 456, + choices: [ + { + index: 0, + delta: { content: 'Answer: forty-two' }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-cum-empty-finish-1', + created: 457, + choices: [ + { + index: 0, + delta: { content: 'Answer: forty-two and more.' }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + // 2) Final chunk: re-sends the accumulated string verbatim along with + // `finish_reason: 'stop'`. The normalized delta is '' (cumulative + // suffix-of-self), but the finish_reason must still drive + // convertOpenAITextToParts. + const finalChunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-cum-empty-finish-2', + created: 458, + choices: [ + { + index: 0, + delta: { content: 'Answer: forty-two and more.' }, + finish_reason: 'stop', + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + + // The cumulative-suppressed empty delta produces no text part, but + // because finish_reason is set, the converter still reaches the parts + // pipeline; on a clean (no buffered tag) state this yields parts: []. + // The crucial invariant: no exception thrown, finishReason propagates, + // and no spurious duplicate text emerges. + expect(finalChunk.candidates?.[0]?.finishReason).toBe('STOP'); + const finalText = + finalChunk.candidates?.[0]?.content?.parts + ?.filter((p) => 'text' in p) + ?.map((p) => (p as { text: string }).text) + ?.join('') ?? ''; + expect(finalText).toBe(''); + }); + it('should handle a single chunk delta with both reasoning_content and content simultaneously', () => { const ctx = withStreamParser(); const part = From a21ac59f0ae7f278b6b2fee8c1664dfb6567850c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Tue, 12 May 2026 14:03:32 +0800 Subject: [PATCH 7/9] fix(core): allow cumulative detection when first chunk exceeds window cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address wenshao's [Critical] review on PR #3896: the CUMULATIVE_DETECTION_WINDOW_BYTES early-return guard was firing before the prefix-overlap and exact-repeat checks, so cumulative providers whose first chunk exceeds the 1024-byte cap had detection permanently disabled, causing the entire first chunk to be duplicated on the second chunk. The cap's original goal — bounding per-stream memory for non-cumulative streams — is already enforced by the concat guard at the fallback path (`emittedText += rawDelta` runs only below the cap). The early-return guard on the detection path was redundant defense-in-depth and harmful to real cumulative streams with large initial chunks. Also applies the [Suggestion] from the same review pass: swap the prefix-overlap conditions so the cheap length check runs before `startsWith`, avoiding the call when `rawDelta` cannot be a strict extension. Adds a regression test covering the cumulative-first-chunk-large case (1500/1700/1750-char chunks) to prevent re-introduction. Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 41 +++++++++++++++++++ .../core/openaiContentGenerator/converter.ts | 11 +---- 2 files changed, 43 insertions(+), 9 deletions(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 33895cad2f0..31efe4c7fc8 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -2498,6 +2498,47 @@ describe('OpenAIContentConverter', () => { expect(allEmitted).toEqual(incrementalChunks); }); + it('should detect cumulative mode even when the first chunk exceeds the detection window cap', () => { + // Regression for https://github.com/QwenLM/qwen-code/pull/3896 review: + // Some cumulative providers ship a large initial chunk (>1024 chars) + // and then accumulate more text on subsequent chunks. The detection + // window cap must not short-circuit prefix-overlap detection before the + // second chunk gets a chance to be classified, otherwise the entire + // first chunk gets duplicated. + const ctx = withStreamParser(); + const firstChunk = 'A'.repeat(1500); // well past CUMULATIVE_DETECTION_WINDOW_BYTES (1024) + const secondChunk = firstChunk + 'B'.repeat(200); // cumulative extension + const thirdChunk = secondChunk + 'C'.repeat(50); // further cumulative extension + + const emitted = [firstChunk, secondChunk, thirdChunk].map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-large-first-${index}`, + created: 789 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // Chunk 1: initial passthrough. + expect(emitted[0]).toBe(firstChunk); + // Chunk 2: prefix-extends → cumulative mode, emits the 200-char suffix only. + expect(emitted[1]).toBe('B'.repeat(200)); + // Chunk 3: continues in cumulative mode, emits only the new 50-char suffix. + expect(emitted[2]).toBe('C'.repeat(50)); + }); + it('should suppress cumulative rewind (provider re-sends shorter accumulated string)', () => { const ctx = withStreamParser(); // Scenario: provider sends Hello → Hello World (extension) → Hello (rewind) → Hello World! (extension again) diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index a4332f8403a..dd1b438ba90 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -110,15 +110,8 @@ function normalizeStreamingTextDelta( } if ( - !state.cumulativeMode && - state.emittedText.length >= CUMULATIVE_DETECTION_WINDOW_BYTES - ) { - return rawDelta; - } - - if ( - rawDelta.startsWith(state.emittedText) && - rawDelta.length > state.emittedText.length + rawDelta.length > state.emittedText.length && + rawDelta.startsWith(state.emittedText) ) { const prevLen = state.emittedText.length; const suffix = rawDelta.slice(prevLen); From cdc7b6b0e5cfcf62fe81b75dc7dde62fbc4f493e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Tue, 12 May 2026 19:18:12 +0800 Subject: [PATCH 8/9] test(core): rename misleading detection-window-cap test The test previously claimed to verify that a trailing chunk arriving after the cap "would have" started with the frozen 1024-byte baseline, but no such chunk existed in the fixture. The 100 distinct incremental chunks never overlapped each other, so prefix/exact-repeat detection never fired regardless of the cap. Rename and re-comment the test to reflect the scenario it actually exercises (incremental passthrough past the cap) and explicitly call out the case it does NOT cover. Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 30 +++++++++++-------- 1 file changed, 18 insertions(+), 12 deletions(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 31efe4c7fc8..d1d1d59b493 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -2457,21 +2457,27 @@ describe('OpenAIContentConverter', () => { expect(emitted[2]).toBe(' extra'); }); - it('should not enter cumulative mode after the detection window cap is reached on a non-cumulative stream', () => { - // After CUMULATIVE_DETECTION_WINDOW_BYTES (1024) of incremental - // emittedText growth, the baseline freezes and prefix/exact-repeat - // detection becomes unsafe. The early-return guard ensures that a chunk - // arriving after the cap is reached is passed through verbatim instead - // of being misclassified against a stale baseline. + it('should pass incremental chunks through verbatim past the detection window cap (none of them overlap)', () => { + // Incremental providers send fresh, non-overlapping chunks. Even after + // emittedText growth exceeds CUMULATIVE_DETECTION_WINDOW_BYTES (1024) + // and the baseline stops growing, every subsequent chunk that lacks + // prefix overlap with the frozen baseline must still be emitted + // verbatim (i.e., it must fall through to the final passthrough + // branch). This guards against any future regression that would, e.g., + // wrongly short-circuit the passthrough path once the cap is reached. + // + // Note: this test does NOT cover the (currently unhandled) case where a + // later chunk happens to start with the frozen baseline — that chunk + // would still trigger prefix-overlap detection against a stale + // baseline. Such a chunk is vanishingly unlikely on a true incremental + // stream (≥1024 bytes of exact-prefix coincidence) but is not + // explicitly defended against here. const ctx = withStreamParser(); - // 100 incremental chunks of 20 chars = 2000 chars, well past the cap. + // 100 distinct incremental chunks of 20 chars = 2000 chars, well past the cap. const incrementalChunks = Array.from( { length: 100 }, (_, i) => `chunk${String(i).padStart(3, '0')}-payload__`, ); - // Trailing chunk that, by coincidence, starts with what *would* be the - // frozen 1024-char baseline. With the early-return guard, this MUST NOT - // enter cumulative mode and MUST emit the chunk verbatim. const allEmitted = incrementalChunks.map( (content, index) => converter.convertOpenAIChunkToGemini( @@ -2493,8 +2499,8 @@ describe('OpenAIContentConverter', () => { ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', ); - // Every chunk should pass through verbatim — none should be - // misclassified as cumulative. + // Every chunk should pass through verbatim — none of them overlap + // with prior emittedText, so prefix/exact-repeat detection never fires. expect(allEmitted).toEqual(incrementalChunks); }); From d1cdc1e6fe69f024e710a130b695496ab3ed3eb8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A7=A6=E5=A5=87?= Date: Wed, 13 May 2026 10:14:46 +0800 Subject: [PATCH 9/9] fix(core): raise exact-repeat threshold and slice from true emitted length Addresses two e2e-verified findings from wenshao's PR #3896 review (2026-05-13 CHANGES_REQUESTED) where local replay surfaced real user-visible regressions in the cumulative-delta heuristic. 1. Bump CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH from 20 to 64. Any legitimately repeated chunk of 20+ chars (e.g. duplicate import lines, repeated short paragraphs, repeated emoji sequences) was silently suppressed by the prior threshold. Cumulative-buffer replays are virtually always hundreds of bytes, so the higher threshold preserves catch-rate while eliminating the silent-loss surface. 2. Track the true user-visible emitted length separately from the 1024-byte capped baseline. For incremental-then-cumulative hybrid streams (200 incremental chunks totalling 1600 bytes followed by a cumulative chunk), the prior implementation sliced the suffix from byte 1024 of the cumulative chunk and re-emitted ~576 bytes the user had already seen. The new emittedLength field captures the true total so the slice starts at the user-visible boundary; the historical short-repeat-then-extend behaviour is preserved by gating the emittedLength-based slice on the baseline actually having frozen at the cap. Also documents the per-stream/per-channel state lifecycle and the "exit cumulative" verbatim-emit semantics on normalizeStreamingTextDelta. Generated with AI Co-authored-by: Qwen-Coder --- .../openaiContentGenerator/converter.test.ts | 160 ++++++++++++++++-- .../core/openaiContentGenerator/converter.ts | 96 +++++++++-- .../src/core/openaiContentGenerator/types.ts | 17 ++ 3 files changed, 239 insertions(+), 34 deletions(-) diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index d1d1d59b493..be4075bf257 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -1950,7 +1950,12 @@ describe('OpenAIContentConverter', () => { it('should ignore repeated cumulative chunks with no new suffix', () => { const ctx = withStreamParser(); - const content = 'The following section starts with enough text.'; + // Must be ≥ CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH (64 chars) so the + // exact-repeat branch enters cumulative mode rather than treating this + // as a short legitimate repeat. Realistic cumulative providers replay + // buffers of hundreds of bytes, so this length is representative. + const content = + 'The following section starts with more than enough text for cumulative-mode detection.'; const emitted = [content, content].map((chunkContent, index) => { const chunk = converter.convertOpenAIChunkToGemini( { @@ -2211,8 +2216,9 @@ describe('OpenAIContentConverter', () => { // with no new suffix` test: the reasoning channel uses a separate state // object, so the exact-repeat entry path is exercised independently. const ctx = withStreamParser(); + // Must be ≥ CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH (64 chars). const reasoning = - 'The reasoning section also starts with enough text to pass.'; + 'The reasoning section also starts with more than enough text to pass detection.'; const emitted = [reasoning, reasoning].map((reasoning_content, index) => { const part = converter.convertOpenAIChunkToGemini( { @@ -2387,18 +2393,21 @@ describe('OpenAIContentConverter', () => { expect(emitted[3]).toEqual([{ text: ' is the answer.' }]); }); - it('should enter cumulative mode on exact 20-char repeat (at threshold)', () => { + it('should enter cumulative mode on exact 64-char repeat (at threshold)', () => { const ctx = withStreamParser(); - // Exactly 20 chars — meets CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH - const twentyChars = 'abcdefghij0123456789'; - const chunks = [twentyChars, twentyChars, twentyChars + ' and more']; + // Exactly 64 chars — meets CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH. + // The threshold sits well above realistic legit-repeat lengths (e.g. a + // duplicate `import { foo } from './module';` is ~31 chars) so that + // legitimate repeats are never silently suppressed. + const atThreshold = 'A'.repeat(64); + const chunks = [atThreshold, atThreshold, atThreshold + ' and more']; const emitted = chunks.map( (content, index) => converter.convertOpenAIChunkToGemini( { object: 'chat.completion.chunk', - id: `chunk-threshold20-${index}`, + id: `chunk-threshold64-${index}`, created: 456 + index, choices: [ { @@ -2415,25 +2424,29 @@ describe('OpenAIContentConverter', () => { ); // Chunk 1: initial passthrough - expect(emitted[0]).toBe(twentyChars); - // Chunk 2: exact 20-char repeat — enters cumulative mode, suppressed + expect(emitted[0]).toBe(atThreshold); + // Chunk 2: exact 64-char repeat — enters cumulative mode, suppressed expect(emitted[1]).toBe(''); // Chunk 3: cumulative extension — emits suffix only expect(emitted[2]).toBe(' and more'); }); - it('should pass through 19-char exact repeat without entering cumulative mode (below threshold)', () => { + it('should pass through 63-char exact repeat without entering cumulative mode (below threshold)', () => { const ctx = withStreamParser(); - // 19 chars — one short of CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH - const nineteenChars = 'abcdefghij012345678'; - const chunks = [nineteenChars, nineteenChars, nineteenChars + ' extra']; + // 63 chars — one short of CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH + const belowThreshold = 'A'.repeat(63); + const chunks = [ + belowThreshold, + belowThreshold, + belowThreshold + ' extra', + ]; const emitted = chunks.map( (content, index) => converter.convertOpenAIChunkToGemini( { object: 'chat.completion.chunk', - id: `chunk-threshold19-${index}`, + id: `chunk-threshold63-${index}`, created: 456 + index, choices: [ { @@ -2450,13 +2463,53 @@ describe('OpenAIContentConverter', () => { ); // Chunk 1: initial passthrough - expect(emitted[0]).toBe(nineteenChars); - // Chunk 2: 19-char repeat — below threshold, passes through unchanged - expect(emitted[1]).toBe(nineteenChars); + expect(emitted[0]).toBe(belowThreshold); + // Chunk 2: 63-char repeat — below threshold, passes through unchanged + expect(emitted[1]).toBe(belowThreshold); // Chunk 3: prefix-extends prior — enters cumulative, emits suffix only expect(emitted[2]).toBe(' extra'); }); + it('should preserve legitimate duplicate import-line chunks (regression: silent data loss)', () => { + // Regression for https://github.com/QwenLM/qwen-code/pull/3896 review + // (wenshao, 2026-05-13 CHANGES_REQUESTED, finding #1). Realistic + // incremental streams emit duplicate import/boilerplate lines and the + // exact-repeat threshold must be high enough that those legitimate + // repeats are NOT silently suppressed. A duplicate ~31-char import is + // the canonical motivating case. + const ctx = withStreamParser(); + const importLine = "import { foo } from './module';"; // 31 chars + const chunks = [importLine, importLine, '\nconst x = 1;']; + + const emitted = chunks.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-import-${index}`, + created: 456 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + // All three chunks must reach the user — no suppression. + expect(emitted[0]).toBe(importLine); + expect(emitted[1]).toBe(importLine); + expect(emitted[2]).toBe('\nconst x = 1;'); + // Sanity: the reassembled stream equals the user-visible total. + expect(emitted.join('')).toBe(importLine + importLine + '\nconst x = 1;'); + }); + it('should pass incremental chunks through verbatim past the detection window cap (none of them overlap)', () => { // Incremental providers send fresh, non-overlapping chunks. Even after // emittedText growth exceeds CUMULATIVE_DETECTION_WINDOW_BYTES (1024) @@ -2545,6 +2598,79 @@ describe('OpenAIContentConverter', () => { expect(emitted[2]).toBe('C'.repeat(50)); }); + it('should not duplicate emitted bytes when an incremental stream transitions into cumulative mode past the window cap', () => { + // Regression for https://github.com/QwenLM/qwen-code/pull/3896 review + // (wenshao, 2026-05-13 CHANGES_REQUESTED, finding #2). Hybrid scenario: + // upstream emits 200 distinct incremental chunks of 8 bytes each (1600 + // bytes of user-visible content, well past the 1024-byte detection- + // window cap), then sends a single cumulative chunk that replays the + // full 1600 bytes and appends new content. The internal baseline froze + // at 1024 bytes; without tracking the true emitted length, the suffix + // would be sliced from byte 1024 of the cumulative chunk and the user + // would see bytes 1024..1600 a second time. The fix tracks emittedLength + // separately so the slice starts from the real user-visible boundary + // (1600). The chunks must be DISTINCT (otherwise the short-exact-repeat + // branch keeps emittedText pinned and the cap is never reached). + const ctx = withStreamParser(); + const incremental = Array.from( + { length: 200 }, + (_, i) => `c${String(i).padStart(3, '0')}=AB_`, // 8 bytes, distinct per chunk + ); + const accumulated = incremental.join(''); // 1600 bytes + const tail = '|CONTINUATION|'; // 14 bytes + const cumulativeChunk = accumulated + tail; // 1614 bytes + + const incrementalEmitted = incremental.map( + (content, index) => + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: `chunk-hybrid-incr-${index}`, + created: 1000 + index, + choices: [ + { + index: 0, + delta: { content }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? '', + ); + + const cumulativeEmitted = + converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-hybrid-cum', + created: 2000, + choices: [ + { + index: 0, + delta: { content: cumulativeChunk }, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ).candidates?.[0]?.content?.parts?.[0]?.text ?? ''; + + // Incremental phase: every chunk passes through verbatim. + expect(incrementalEmitted).toEqual(incremental); + // Cumulative chunk: only the new 14-byte tail must be emitted — not the + // ~576 bytes between the cap (1024) and the true emitted total (1600). + expect(cumulativeEmitted).toBe(tail); + // Sanity: reassembled stream equals the original accumulated text. + const userVisible = incrementalEmitted.join('') + cumulativeEmitted; + expect(userVisible).toBe(cumulativeChunk); + expect(userVisible.length).toBe(1614); + }); + it('should suppress cumulative rewind (provider re-sends shorter accumulated string)', () => { const ctx = withStreamParser(); // Scenario: provider sends Hello → Hello World (extension) → Hello (rewind) → Hello World! (extension again) diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index dd1b438ba90..ea218638d07 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -60,19 +60,48 @@ export interface ExtendedCompletionChunkDelta } // Threshold for treating an exact-repeat chunk as a cumulative marker rather -// than legitimate repeated content. Cumulative providers typically emit whole -// words/phrases (≥20 chars); sub-word repeats (e.g. "ha") are more likely to -// be valid incremental output. -const CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH = 20; +// than legitimate repeated content. Cumulative providers replay the entire +// accumulated buffer (typically hundreds of bytes) on each re-send, while +// legitimate repeats in real output (duplicated import lines, repeated short +// boilerplate like "\n", repeated emoji sequences) are usually +// well under this threshold. 64 sits comfortably above realistic legit-repeat +// lengths while remaining far below any practical cumulative-buffer replay, +// so it preserves catch-rate without silently suppressing legitimate chunks. +const CUMULATIVE_DELTA_EXACT_REPEAT_MIN_LENGTH = 64; // Once this many bytes have been emitted without entering cumulative mode the // stream is almost certainly a standard incremental provider. Stop growing -// emittedText beyond this point to bound per-stream memory and CPU. +// emittedText beyond this point to bound per-stream memory and CPU. The true +// emitted total is preserved separately in `state.emittedLength` so a late +// transition into cumulative mode still slices the correct suffix. const CUMULATIVE_DETECTION_WINDOW_BYTES = 1024; -// Some OpenAI-compatible providers send accumulated content in each -// delta.content field. Normalize that shape to incremental suffixes before the -// Gemini stream layer appends it to the live transcript. +/** + * Some OpenAI-compatible providers (e.g. DashScope) send the entire + * accumulated content in each `delta.content` field instead of incremental + * suffixes. Normalize that shape to incremental suffixes before the Gemini + * stream layer appends it to the live transcript. + * + * State invariants and lifecycle: + * - `state` is per-stream and per-channel — the content and reasoning + * channels are tracked independently to avoid cross-contamination. State + * MUST NOT be shared or reused across requests; stale state will silently + * corrupt text output. + * - In cumulative mode `state.emittedText` retains the full accumulated text + * for the request lifetime (worst case: ~final response size, e.g. ~100KB + * for a long answer). This is single-request scoped and bounded by request + * completion. A future optimization could retain only the last N bytes once + * cumulative mode is firmly established, but is not required today. + * - In non-cumulative mode `state.emittedText` is capped at + * CUMULATIVE_DETECTION_WINDOW_BYTES; `state.emittedLength` tracks the true + * user-visible total separately so a late transition into cumulative mode + * still produces the correct suffix. + * - The "exit cumulative" path is a verbatim-emit path with no overlap + * reconciliation: the diverged chunk is assumed to be fully fresh content. + * Cumulative providers that emit a half-overlapping chunk on exit (not + * observed on DashScope-class providers) would produce visible duplication + * on the overlap. + */ function normalizeStreamingTextDelta( rawDelta: string, state: StreamingTextDeltaState, @@ -83,6 +112,7 @@ function normalizeStreamingTextDelta( if (state.emittedText.length === 0) { state.emittedText = rawDelta; + state.emittedLength = rawDelta.length; return rawDelta; } @@ -90,6 +120,7 @@ function normalizeStreamingTextDelta( if (rawDelta.startsWith(state.emittedText)) { const suffix = rawDelta.slice(state.emittedText.length); state.emittedText = rawDelta; + state.emittedLength = rawDelta.length; return suffix; } @@ -105,7 +136,12 @@ function normalizeStreamingTextDelta( ); state.cumulativeMode = false; // Reset baseline to current chunk so future prefix checks use fresh state. + // Note: this is a verbatim-emit path with no overlap reconciliation — the + // diverged chunk is assumed to be fully fresh content. If a cumulative + // provider were to emit a half-overlapping chunk on exit (rare; not + // observed on DashScope-class providers) the overlap would be visible. state.emittedText = rawDelta; + state.emittedLength += rawDelta.length; return rawDelta; } @@ -113,14 +149,35 @@ function normalizeStreamingTextDelta( rawDelta.length > state.emittedText.length && rawDelta.startsWith(state.emittedText) ) { - const prevLen = state.emittedText.length; - const suffix = rawDelta.slice(prevLen); - state.emittedText = rawDelta; - state.cumulativeMode = true; - debugLogger.debug( - `normalizeStreamingTextDelta: entered cumulative mode (prefix overlap, prev=${prevLen}b -> curr=${rawDelta.length}b)`, - ); - return suffix; + const baselineLen = state.emittedText.length; + // The baseline may have been frozen at CUMULATIVE_DETECTION_WINDOW_BYTES + // during a long incremental phase. If the cap actually kicked in and the + // real emitted total exceeds the (frozen) baseline, slice the suffix from + // the real total so an incremental-then-cumulative hybrid stream doesn't + // re-emit bytes the user already saw between the cap and the true total. + // Outside that hybrid-after-cap case, use the baseline so the historical + // short-repeat-then-extend behaviour is preserved (the baseline is kept + // unmodified across short exact repeats specifically to support that + // case). + const baselineFrozenAtCap = + baselineLen >= CUMULATIVE_DETECTION_WINDOW_BYTES && + state.emittedLength > baselineLen; + const sliceFrom = baselineFrozenAtCap ? state.emittedLength : baselineLen; + if (rawDelta.length > sliceFrom) { + const suffix = rawDelta.slice(sliceFrom); + state.emittedText = rawDelta; + state.emittedLength = rawDelta.length; + state.cumulativeMode = true; + debugLogger.debug( + `normalizeStreamingTextDelta: entered cumulative mode (prefix overlap, baseline=${baselineLen}b sliceFrom=${sliceFrom}b -> curr=${rawDelta.length}b)`, + ); + return suffix; + } + // rawDelta startsWith baseline but isn't strictly longer than sliceFrom. + // Only reachable in the baselineFrozenAtCap branch when the cumulative + // chunk is shorter than the real emitted total (a cumulative-rewind-like + // shape during the transition). Treat as a no-op: don't enter cumulative + // mode here, fall through to the rewind/passthrough branches below. } if (rawDelta === state.emittedText) { @@ -132,13 +189,16 @@ function normalizeStreamingTextDelta( return ''; } // Short exact repeat: don't mutate emittedText so it remains a valid - // prefix baseline for the next prefix-overlap check. + // prefix baseline for the next prefix-overlap check. The chunk is still + // emitted verbatim, so bump emittedLength to track user-visible bytes. + state.emittedLength += rawDelta.length; return rawDelta; } if (state.emittedText.length < CUMULATIVE_DETECTION_WINDOW_BYTES) { state.emittedText += rawDelta; } + state.emittedLength += rawDelta.length; return rawDelta; } @@ -1117,6 +1177,7 @@ export function convertOpenAIChunkToGemini( reasoningText, (requestContext.reasoningDeltaState ??= { emittedText: '', + emittedLength: 0, cumulativeMode: false, }), ); @@ -1132,6 +1193,7 @@ export function convertOpenAIChunkToGemini( choice.delta.content, (requestContext.textDeltaState ??= { emittedText: '', + emittedLength: 0, cumulativeMode: false, }), ); diff --git a/packages/core/src/core/openaiContentGenerator/types.ts b/packages/core/src/core/openaiContentGenerator/types.ts index b47f80a862f..046155d62ce 100644 --- a/packages/core/src/core/openaiContentGenerator/types.ts +++ b/packages/core/src/core/openaiContentGenerator/types.ts @@ -16,7 +16,24 @@ import type { StreamingToolCallParser } from './streamingToolCallParser.js'; import type { TaggedThinkingParser } from './taggedThinkingParser.js'; export interface StreamingTextDeltaState { + /** + * Rolling baseline used for prefix/exact-repeat detection. Once the stream + * has been classified as incremental and the buffer reaches + * CUMULATIVE_DETECTION_WINDOW_BYTES bytes it is frozen at the cap to bound + * memory; the true emitted total is tracked separately in `emittedLength`. + * In cumulative mode this always reflects the full accumulated text. + */ emittedText: string; + /** + * Monotonic count of user-visible bytes already emitted on this channel. + * Diverges from `emittedText.length` only on long incremental streams where + * `emittedText` is capped at CUMULATIVE_DETECTION_WINDOW_BYTES. Used to slice + * the correct suffix when an incremental-then-cumulative hybrid stream + * transitions into cumulative mode after the cap (otherwise the suffix would + * re-include bytes between the cap and the true emitted length, producing + * visible duplication). + */ + emittedLength: number; cumulativeMode: boolean; }