diff --git a/packages/core/src/core/geminiChat.test.ts b/packages/core/src/core/geminiChat.test.ts index cb0df1f80bf..10e5d41aa57 100644 --- a/packages/core/src/core/geminiChat.test.ts +++ b/packages/core/src/core/geminiChat.test.ts @@ -5,6 +5,7 @@ */ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import type OpenAI from 'openai'; import type { Content, GenerateContentConfig, @@ -23,6 +24,8 @@ import { import { RETRYABLE_STREAM_TRANSPORT_CODES } from './stream-transport-retry.js'; import { classifyRetryError } from '../utils/retryErrorClassification.js'; import { StreamContentError } from './openaiContentGenerator/pipeline.js'; +import { OpenAIContentGenerator } from './openaiContentGenerator/openaiContentGenerator.js'; +import type { OpenAICompatibleProvider } from './openaiContentGenerator/provider/index.js'; import type { Config } from '../config/config.js'; import { setSimulate429 } from '../utils/testUtils.js'; import { uiTelemetryService } from '../telemetry/uiTelemetry.js'; @@ -4729,53 +4732,97 @@ describe('GeminiChat', async () => { } }); - it('should keep protocol tag leak retries at the existing budget', async () => { - vi.useFakeTimers(); - try { - vi.mocked( - mockContentGenerator.generateContentStream, - ).mockImplementation(async () => - (async function* () { - yield { - candidates: [ - { - content: { - parts: [ - { - text: 'hiddenleaked', - }, - ], + it.each([ + { + name: 'protocol tag leaks', + errorType: 'PROTOCOL_TAG_LEAK', + delta: { + reasoning_content: 'hidden reasoning', + content: ' leaked visible reasoning', + }, + finishReason: 'stop', + retryCount: 2, + }, + { + name: 'malformed tool calls', + errorType: 'MALFORMED_TOOL_CALL', + delta: { + tool_calls: [ + { + index: 0, + id: 'call_without_name', + type: 'function', + function: { arguments: '{}' }, + }, + ], + }, + finishReason: 'tool_calls', + retryCount: 4, + }, + ] as const)( + 'should retry $name through the OpenAI pipeline', + async (testCase) => { + vi.useFakeTimers(); + try { + const create = vi.fn().mockImplementation(async () => + (async function* () { + yield { + id: 'protocol-tag-leak', + created: 1, + model: 'test-model', + choices: [ + { + index: 0, + delta: testCase.delta, + finish_reason: testCase.finishReason, }, - finishReason: 'STOP', - }, - ], - } as unknown as GenerateContentResponse; - })(), - ); + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk; + })(), + ); + const provider = { + buildClient: () => + ({ chat: { completions: { create } } }) as unknown as OpenAI, + buildRequest: (request: OpenAI.Chat.ChatCompletionCreateParams) => + request, + buildHeaders: () => ({}), + getDefaultGenerationConfig: () => ({}), + } as OpenAICompatibleProvider; + const generator = new OpenAIContentGenerator( + { model: 'test-model', authType: AuthType.USE_OPENAI }, + mockConfig, + provider, + ); + vi.mocked(mockConfig.getContentGenerator).mockReturnValue(generator); + vi.mocked(mockConfig.getContentGeneratorConfig).mockReturnValue({ + model: 'test-model', + authType: AuthType.USE_OPENAI, + }); - const stream = await chat.sendMessageStream( - 'test-model', - { message: 'test' }, - 'prompt-id-protocol-leak-budget', - ); - await expectStreamExhaustion(stream); + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + `prompt-id-${testCase.errorType.toLowerCase()}-budget`, + ); + await expectStreamExhaustion(stream); - expect( - mockContentGenerator.generateContentStream, - ).toHaveBeenCalledTimes(3); - expect(mockLogContentRetry).toHaveBeenCalledTimes(2); - expect(mockLogContentRetryFailure).toHaveBeenCalledWith( - mockConfig, - expect.objectContaining({ - total_attempts: 3, - final_error_type: 'PROTOCOL_TAG_LEAK', - model: 'test-model', - }), - ); - } finally { - vi.useRealTimers(); - } - }); + expect(create).toHaveBeenCalledTimes(testCase.retryCount + 1); + expect(mockLogContentRetry).toHaveBeenCalledTimes( + testCase.retryCount, + ); + expect(mockLogContentRetryFailure).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + total_attempts: testCase.retryCount + 1, + final_error_type: testCase.errorType, + model: 'test-model', + }), + ); + } finally { + vi.useRealTimers(); + } + }, + ); it('keeps invalid stream retry budgets independent across error types', async () => { vi.useFakeTimers(); @@ -6198,17 +6245,6 @@ describe('GeminiChat', async () => { } }); - // NOTE: no test for the InvalidStreamError content-retry branch - // (geminiChat.ts ~line 1399). Verified unreachable for that error - // class: `isTransientStreamError` and `isContentError` are the - // same predicate (`error instanceof InvalidStreamError`), so the - // transient branch above always either `continue`s or `break`s - // before control reaches the content branch. The - // `popPartialIfPushed()` call there is preserved as - // defense-in-depth for a future error class that should diverge - // the predicates; see the comment block at that call site for - // the full analysis. - it('rolls back the chat-recording entry too when the retry succeeds', async () => { // The in-memory rollback test above asserts `this.history` ends // clean after a retry-success. This test asserts the same about @@ -8595,7 +8631,7 @@ describe('GeminiChat', async () => { } function invalidStream( - type: 'NO_FINISH_REASON' | 'PROTOCOL_TAG_LEAK', + type: InvalidStreamError['type'], ): AsyncGenerator { return { [Symbol.asyncIterator]() { @@ -9907,8 +9943,6 @@ describe('GeminiChat', async () => { invalidStream('PROTOCOL_TAG_LEAK'), invalidStream('PROTOCOL_TAG_LEAK'), invalidStream('PROTOCOL_TAG_LEAK'), - invalidStream('PROTOCOL_TAG_LEAK'), - invalidStream('PROTOCOL_TAG_LEAK'), ]; let callIndex = 0; vi.mocked( diff --git a/packages/core/src/core/geminiChat.ts b/packages/core/src/core/geminiChat.ts index 880c6fe4c14..e8c8c8c6613 100644 --- a/packages/core/src/core/geminiChat.ts +++ b/packages/core/src/core/geminiChat.ts @@ -102,6 +102,9 @@ import { collectToolCallIdsFromHistory, normalizeModelToolCallIds, } from './toolCallIdUtils.js'; +import { InvalidStreamError } from './invalid-stream-error.js'; + +export { InvalidStreamError }; const debugLogger = createDebugLogger('QWEN_CODE_CHAT'); @@ -292,16 +295,6 @@ export type StreamEvent = | { type: StreamEventType.COMPRESSED; info: ChatCompressionInfo } | { type: StreamEventType.MODEL_FALLBACK; info: ModelFallbackInfo }; -/** - * Options for retrying due to invalid content from the model. - */ -interface ContentRetryOptions { - /** Total number of attempts to make (1 initial + N retries). */ - maxAttempts: number; - /** The base delay in milliseconds for linear backoff. */ - initialDelayMs: number; -} - interface TryCompressOptions { originalTokenCountOverride?: number; trigger?: CompactTrigger; @@ -332,20 +325,12 @@ interface TryCompressOptions { customInstructions?: string; } -const INVALID_CONTENT_RETRY_OPTIONS: ContentRetryOptions = { - maxAttempts: 2, // 1 initial call + 1 retry - initialDelayMs: 500, -}; - -// Some providers occasionally return transient stream anomalies: either an -// empty stream (usage metadata only, no candidates), a stream that finishes -// normally but contains no usable text, or a stream cut off without a finish -// reason. All are retried with an independent budget (similar to rate-limit -// retries) so they do not consume each other's retry budgets. +// Model-output validation errors (protocol tag leaks, malformed tool calls) +// and transient stream anomalies (empty streams, no usable text, missing +// finish reason) use an independent retry budget so they do not consume each +// other's or HTTP retries' budgets. const INVALID_STREAM_RETRY_CONFIG = { transientMaxRetries: 4, - // Protocol-tag leaks are model-output validation failures, not the - // provider-side empty/truncated streams covered by issue #6670. protocolTagLeakMaxRetries: 2, initialDelayMs: 2000, }; @@ -1041,23 +1026,6 @@ function stripThoughtPartsFromContent(content: Content): Content | null { }; } -/** - * Custom error to signal that a stream completed with invalid content, - * which should trigger a retry. - */ -export class InvalidStreamError extends Error { - readonly type: 'NO_FINISH_REASON' | 'NO_RESPONSE_TEXT' | 'PROTOCOL_TAG_LEAK'; - - constructor( - message: string, - type: 'NO_FINISH_REASON' | 'NO_RESPONSE_TEXT' | 'PROTOCOL_TAG_LEAK', - ) { - super(message); - this.name = 'InvalidStreamError'; - this.type = type; - } -} - const PROTOCOL_TAG_PREFIXES = [ ' 0 || rateLimitRetryCount > 0 || totalInvalidStreamRetryCount() > 0 || transportStreamRetryCount > 0 @@ -2355,8 +2318,6 @@ export class GeminiChat { skipDelay: skip, }, }; - // Don't count rate-limit retries against the content retry limit - attempt--; await delayPromise; continue; } @@ -2400,8 +2361,6 @@ export class GeminiChat { }); yield { type: StreamEventType.RETRY }; suppressNextRetryEvent = true; - // Don't count transport retries against the content retry limit. - attempt--; await delay(delayMs, params.config?.abortSignal).promise; continue; } @@ -2468,9 +2427,6 @@ export class GeminiChat { }; yield { type: StreamEventType.RETRY }; suppressNextRetryEvent = true; - // Do not count reactive compression against the content - // validation retry budget. - attempt--; continue; } @@ -2518,8 +2474,8 @@ export class GeminiChat { break; } - // Invalid stream responses use an independent retry budget and do - // not consume the content retry budget. + // Invalid stream responses use INVALID_STREAM_RETRY_CONFIG, which + // is independent from HTTP retries handled by retryWithBackoff. const isInvalidStreamError = error instanceof InvalidStreamError; const maxInvalidStreamRetries = isInvalidStreamError && error.type === 'PROTOCOL_TAG_LEAK' @@ -2558,46 +2514,9 @@ export class GeminiChat { ), ); yield { type: StreamEventType.RETRY }; - // Don't count transient retries against content retry limit. - attempt--; await delay(delayMs, params.config?.abortSignal).promise; continue; } - // Invalid-stream budget exhausted — stop immediately. - if (isInvalidStreamError) { - break; - } - - // Currently unreachable for `InvalidStreamError`. The - // `isContentError` predicate is identical to - // `isInvalidStreamError` (`error instanceof InvalidStreamError`), - // and the transient branch above already either continued or - // broke for that class. The branch is preserved as - // defense-in-depth: a future error class that should consume - // its own content-retry budget but NOT the transient one - // could be threaded through here without re-deriving the - // popPartialIfPushed sequence. No reachable test path until - // the predicates diverge. - const isContentError = error instanceof InvalidStreamError; - if (isContentError) { - if (attempt < INVALID_CONTENT_RETRY_OPTIONS.maxAttempts - 1) { - self.popPendingPartialAssistantTurn(); - logContentRetry( - self.config, - new ContentRetryEvent( - attempt, - (error as InvalidStreamError).type, - INVALID_CONTENT_RETRY_OPTIONS.initialDelayMs, - model, - ), - ); - await delay( - INVALID_CONTENT_RETRY_OPTIONS.initialDelayMs * (attempt + 1), - params.config?.abortSignal, - ).promise; - continue; - } - } break; } } diff --git a/packages/core/src/core/invalid-stream-error.ts b/packages/core/src/core/invalid-stream-error.ts new file mode 100644 index 00000000000..2452dde6dd7 --- /dev/null +++ b/packages/core/src/core/invalid-stream-error.ts @@ -0,0 +1,22 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Custom error to signal invalid model stream content that should be retried. + */ +export class InvalidStreamError extends Error { + readonly type: + | 'NO_FINISH_REASON' + | 'NO_RESPONSE_TEXT' + | 'PROTOCOL_TAG_LEAK' + | 'MALFORMED_TOOL_CALL'; + + constructor(message: string, type: InvalidStreamError['type']) { + super(message); + this.name = 'InvalidStreamError'; + this.type = type; + } +} diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 560158e35c3..79e9d427106 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -96,6 +96,18 @@ describe('OpenAIContentConverter', () => { } describe('stream-local parser state', () => { + const streamChunk = ( + id: string, + delta: Record, + finishReason: string | null = null, + ) => + ({ + id, + created: 1, + model: 'test', + choices: [{ index: 0, delta, finish_reason: finishReason }], + }) as unknown as OpenAI.Chat.ChatCompletionChunk; + it('creates fresh parser instances', () => { const ctx1 = new StreamingToolCallParser(); const ctx2 = new StreamingToolCallParser(); @@ -351,6 +363,169 @@ describe('OpenAIContentConverter', () => { expect(fn?.args).toEqual({}); expect(fn?.id).toBe('call_noargs'); }); + + it('ignores a phantom slot beside a valid tool call', () => { + const stream = withStreamParser(); + converter.convertOpenAIChunkToGemini( + streamChunk('open', { + tool_calls: [ + { + index: 0, + id: 'call_edit', + function: { name: 'edit', arguments: '{}' }, + }, + { index: 1, function: {} }, + ], + }), + stream, + ); + + const result = converter.convertOpenAIChunkToGemini( + streamChunk('finish', {}, 'tool_calls'), + stream, + ); + + expect(result.candidates?.[0]?.content?.parts).toEqual([ + { + functionCall: { id: 'call_edit', name: 'edit', args: {} }, + }, + ]); + expect(result.candidates?.[0]?.finishReason).toBe(FinishReason.STOP); + }); + + it('rejects a tool-call finish without a completed named call', () => { + const stream = withStreamParser(); + converter.convertOpenAIChunkToGemini( + streamChunk('open', { tool_calls: [{ index: 0, function: {} }] }), + stream, + ); + + expect(() => + converter.convertOpenAIChunkToGemini( + streamChunk('finish', {}, 'tool_calls'), + stream, + ), + ).toThrowError(expect.objectContaining({ type: 'MALFORMED_TOOL_CALL' })); + }); + + it('rejects a tool call that never provides a function name', () => { + const stream = withStreamParser(); + const partial = converter.convertOpenAIChunkToGemini( + streamChunk('open', { + content: 'discard me', + tool_calls: [ + { + index: 0, + id: 'call_without_name', + function: { arguments: '{"path":"a.ts"}' }, + }, + ], + }), + stream, + ); + + expect(partial.candidates?.[0]?.content?.parts).toEqual([]); + expect(() => + converter.convertOpenAIChunkToGemini( + streamChunk('finish', {}, 'stop'), + stream, + ), + ).toThrowError(expect.objectContaining({ type: 'MALFORMED_TOOL_CALL' })); + }); + + it('rejects the recorded cross-channel thinking-tag leak', () => { + const stream = withStreamParser(); + const reasoning = converter.convertOpenAIChunkToGemini( + streamChunk('reasoning', { + reasoning_content: 'Let me check', + }), + stream, + ); + + expect(reasoning.candidates?.[0]?.content?.parts).toEqual([]); + expect(() => + converter.convertOpenAIChunkToGemini( + streamChunk('content', { content: 'the result\n\n' }), + stream, + ), + ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' })); + }); + + it('rejects a closing tag split after a visible line break', () => { + const stream = withStreamParser(); + converter.convertOpenAIChunkToGemini( + streamChunk('reasoning', { + reasoning_content: 'Let me check', + }), + stream, + ); + converter.convertOpenAIChunkToGemini( + streamChunk('content', { content: 'the result\n' }), + stream, + ); + converter.convertOpenAIChunkToGemini( + streamChunk('blank-lines', { content: '\n'.repeat(256) }), + stream, + ); + + expect(() => + converter.convertOpenAIChunkToGemini( + streamChunk('closing-tag', { content: '\n' }), + stream, + ), + ).toThrowError(expect.objectContaining({ type: 'PROTOCOL_TAG_LEAK' })); + }); + + it('preserves a split literal closing tag after ordinary reasoning', () => { + const stream = withStreamParser(); + const reasoning = converter.convertOpenAIChunkToGemini( + streamChunk('reasoning', { reasoning_content: 'Explain the syntax.' }), + stream, + ); + const prefix = converter.convertOpenAIChunkToGemini( + streamChunk('prefix', { content: 'Use ' }), + stream, + ); + const closingTag = converter.convertOpenAIChunkToGemini( + streamChunk('closing-tag', { content: ' to close the tag.' }), + stream, + ); + + expect(reasoning.candidates?.[0]?.content?.parts).toEqual([ + { thought: true, text: 'Explain the syntax.' }, + ]); + expect(prefix.candidates?.[0]?.content?.parts).toEqual([ + { text: 'Use ' }, + ]); + expect(closingTag.candidates?.[0]?.content?.parts).toEqual([ + { text: ' to close the tag.' }, + ]); + }); + + it('releases inline thinking-tag references in both channels', () => { + const stream = withStreamParser(); + const reasoning = converter.convertOpenAIChunkToGemini( + streamChunk('reasoning', { + reasoning_content: 'The format may contain tags.', + }), + stream, + ); + const content = converter.convertOpenAIChunkToGemini( + streamChunk('content', { content: 'Use to close the tag.' }), + stream, + ); + const finish = converter.convertOpenAIChunkToGemini( + streamChunk('finish', {}, 'stop'), + stream, + ); + + expect(reasoning.candidates?.[0]?.content?.parts).toEqual([]); + expect(content.candidates?.[0]?.content?.parts).toEqual([]); + expect(finish.candidates?.[0]?.content?.parts).toEqual([ + { thought: true, text: 'The format may contain tags.' }, + { text: 'Use to close the tag.' }, + ]); + }); }); describe('convertGeminiRequestToOpenAI', () => { diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index 5bfcdb4cac5..b1c7a81f1c6 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -28,6 +28,7 @@ import { convertSchema, type SchemaComplianceMode, } from '../../utils/schemaConverter.js'; +import { InvalidStreamError } from '../invalid-stream-error.js'; const debugLogger = createDebugLogger('CONVERTER'); const SPLIT_TOOL_MEDIA_TEXT = '(attached media from previous tool call)'; @@ -1086,6 +1087,12 @@ function hasThoughtPart(parts: Part[]): boolean { return parts.some((part) => part.thought === true); } +const THINKING_TAG_PATTERN = /<\/?think(?:ing)?\s*>/i; +const CLOSING_THINKING_TAG_PATTERN = /\n[^\S\r\n]*<\/think(?:ing)?[^\S\r\n]*>/i; +const LEADING_CLOSING_THINKING_TAG_PATTERN = + /^[^\S\r\n]*<\/think(?:ing)?[^\S\r\n]*>/i; +const LEADING_THINKING_TAG_PATTERN = /^\s*<\/?think(?:ing)?\s*>/i; + /** * Convert OpenAI response to Gemini format. */ @@ -1225,7 +1232,7 @@ export function convertOpenAIChunkToGemini( } if (choice) { - const parts: Part[] = []; + let parts: Part[] = []; let contentParts: Part[] = []; // Handle reasoning content (thoughts). @@ -1285,6 +1292,12 @@ export function convertOpenAIChunkToGemini( cumulativeMode: false, }), ); + if (normalizedReasoningText) { + requestContext.hasStructuredReasoningContent = true; + if (THINKING_TAG_PATTERN.test(normalizedReasoningText)) { + requestContext.hasThinkingTagInReasoning = true; + } + } if ( normalizedReasoningText && !requestContext.responseParsingOptions?.taggedThinkingTags @@ -1366,6 +1379,66 @@ export function convertOpenAIChunkToGemini( } } + const visibleText = parts + .map((part) => + part.thought !== true && typeof part.text === 'string' ? part.text : '', + ) + .join(''); + const leakedThinkingTag = + requestContext.hasStructuredReasoningContent === true && + ((requestContext.hasVisibleContent !== true && + LEADING_THINKING_TAG_PATTERN.test(visibleText)) || + (requestContext.hasThinkingTagInReasoning === true && + (CLOSING_THINKING_TAG_PATTERN.test(visibleText) || + (requestContext.atVisibleLineStart === true && + LEADING_CLOSING_THINKING_TAG_PATTERN.test(visibleText))))); + if (/\S/.test(visibleText)) { + requestContext.hasVisibleContent = true; + } + if (visibleText && requestContext.hasThinkingTagInReasoning === true) { + const lastLineBreak = visibleText.lastIndexOf('\n'); + const lineSuffix = visibleText.slice(lastLineBreak + 1); + requestContext.atVisibleLineStart = + (lastLineBreak >= 0 || requestContext.atVisibleLineStart === true) && + /^[^\S\r\n]*$/.test(lineSuffix); + } + if (leakedThinkingTag) { + requestContext.pendingUntrustedResponseParts = undefined; + throw new InvalidStreamError( + 'Model response leaked thinking tags.', + 'PROTOCOL_TAG_LEAK', + ); + } + + const toolCallWithoutName = toolCallParser.hasNamelessToolCall(); + const completedToolCalls = choice.finish_reason + ? toolCallParser.getCompletedToolCalls() + : []; + if ( + choice.finish_reason && + (toolCallWithoutName || + (choice.finish_reason === 'tool_calls' && + completedToolCalls.length === 0)) + ) { + requestContext.pendingUntrustedResponseParts = undefined; + throw new InvalidStreamError( + 'Model response contained a malformed tool call.', + 'MALFORMED_TOOL_CALL', + ); + } + + const shouldHoldParts = + !choice.finish_reason && + (toolCallWithoutName || + requestContext.hasThinkingTagInReasoning === true); + if (shouldHoldParts) { + (requestContext.pendingUntrustedResponseParts ??= []).push(...parts); + parts.length = 0; + } else if (requestContext.pendingUntrustedResponseParts) { + parts = requestContext.pendingUntrustedResponseParts.concat(parts); + requestContext.pendingUntrustedResponseParts = undefined; + } + // Only emit function calls when streaming is complete (finish_reason is present) let toolCallsTruncated = false; if (choice.finish_reason) { @@ -1374,8 +1447,6 @@ export function convertOpenAIChunkToGemini( // even when output was cut off mid-JSON due to max_tokens. toolCallsTruncated = toolCallParser.hasIncompleteToolCalls(); - const completedToolCalls = toolCallParser.getCompletedToolCalls(); - for (const toolCall of completedToolCalls) { if (toolCall.name) { parts.push({ diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index 966c4821413..f0048e5b8f9 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.ts @@ -27,6 +27,7 @@ import { QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, } from './constants.js'; import { createDebugLogger } from '../../utils/debugLogger.js'; +import { InvalidStreamError } from '../invalid-stream-error.js'; const debugLogger = createDebugLogger('OPENAI_PIPELINE'); @@ -557,6 +558,10 @@ export class ContentGenerationPipeline { yield pendingFinishResponse; } } catch (error) { + if (error instanceof InvalidStreamError) { + throw error; + } + // Re-throw StreamContentError directly so it can be handled by // the caller's retry logic (e.g., TPM throttling retry in sendMessageStream) if (error instanceof StreamContentError) { @@ -607,10 +612,9 @@ export class ContentGenerationPipeline { if (isFinishChunk) { if (hasPendingFinish) { // Duplicate finish chunk (e.g. from OpenRouter providers that send two - // finish_reason chunks for tool calls). The streaming tool call parser - // was already reset after the first finish chunk, so the second one - // carries no functionCall parts. Merge only usageMetadata and keep the - // candidates (including functionCall parts) from the first finish chunk. + // finish_reason chunks for tool calls). The first finish response owns + // the candidates, including functionCall parts. Merge only usageMetadata + // from later finish chunks. const lastResponse = collectedGeminiResponses[collectedGeminiResponses.length - 1]; if (response.usageMetadata) { diff --git a/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.test.ts b/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.test.ts index e7dc1bec2e7..f9485a2d9d5 100644 --- a/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.test.ts +++ b/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.test.ts @@ -281,6 +281,40 @@ describe('StreamingToolCallParser', () => { }); describe('Tool call metadata handling', () => { + it('tracks real nameless calls but ignores phantom chunks', () => { + parser.addChunk(0, ''); + expect(parser.hasNamelessToolCall()).toBe(false); + + parser.addChunk(0, '', 'call_1'); + expect(parser.hasNamelessToolCall()).toBe(true); + + parser.resetIndex(0); + parser.addChunk(0, '{"path":"a.ts"}'); + expect(parser.hasNamelessToolCall()).toBe(true); + }); + + it.each([ + { index: 1, id: undefined }, + { index: 2, id: 'call_2' }, + ])( + 'accepts and deduplicates a name after complete arguments at index $index', + ({ index, id }) => { + parser.addChunk(index, '{"path":"a.ts"}', id); + parser.addChunk(index, '', id, 'read_file'); + parser.addChunk(index, '', id, 'read_file'); + + expect(parser.hasNamelessToolCall()).toBe(false); + expect(parser.getCompletedToolCalls()).toEqual([ + { + id, + name: 'read_file', + args: { path: 'a.ts' }, + index, + }, + ]); + }, + ); + it('should store and retrieve tool call metadata', () => { parser.addChunk(0, '{"param": "value"}', 'call_123', 'my_function'); @@ -698,11 +732,18 @@ describe('StreamingToolCallParser', () => { // known-ID chunk carrying argument content is a continuation, not a // replay, and must not be swallowed by the replay guard. parser.addChunk(0, '', 'call_1', 'function1'); - const result = parser.addChunk(0, '{"x":1}', 'call_1'); + parser.addChunk(0, '{"text":"hello', 'call_1'); + parser.addChunk(0, ' ', 'call_1'); + const result = parser.addChunk(0, 'world"}', 'call_1'); expect(result.complete).toBe(true); expect(parser.getCompletedToolCalls()).toEqual([ - { id: 'call_1', name: 'function1', args: { x: 1 }, index: 0 }, + { + id: 'call_1', + name: 'function1', + args: { text: 'hello world' }, + index: 0, + }, ]); }); diff --git a/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.ts b/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.ts index 3379e5f5a97..2c962397b89 100644 --- a/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.ts +++ b/packages/core/src/core/openaiContentGenerator/streamingToolCallParser.ts @@ -44,6 +44,7 @@ export class StreamingToolCallParser { private escapes: Map = new Map(); /** Metadata for each tool call index */ private toolCallMeta: Map = new Map(); + private namelessToolCallIndices = new Set(); /** Map from tool call ID to actual index used for storage */ private idToIndexMap: Map = new Map(); /** Counter for generating new indices when collisions occur */ @@ -71,8 +72,17 @@ export class StreamingToolCallParser { id?: string, name?: string, ): ToolCallParseResult { + if (!id && !name && !chunk.trim()) { + const depth = this.depths.get(index) ?? 0; + const inString = this.inStrings.get(index) ?? false; + if (!this.buffers.has(index) || (depth === 0 && !inString)) { + return { complete: false }; + } + } + let actualIndex = index; const isKnownId = Boolean(id && this.idToIndexMap.has(id)); + const isNameOnlyDelta = Boolean(name && chunk.length === 0); // Handle tool call ID mapping for collision detection if (id) { @@ -117,7 +127,7 @@ export class StreamingToolCallParser { // Map this ID to the actual index we're using this.idToIndexMap.set(id, actualIndex); } - } else { + } else if (!isNameOnlyDelta) { // No ID provided - this is a continuation chunk // Try to find which tool call this belongs to based on the index // Look for an existing tool call at this index that's not complete @@ -154,6 +164,18 @@ export class StreamingToolCallParser { const currentBuffer = this.buffers.get(actualIndex)!; const currentDepth = this.depths.get(actualIndex)!; + const meta = this.toolCallMeta.get(actualIndex)!; + if (chunk.length === 0 && (id || name)) { + if (id) meta.id = id; + if (name && !meta.name) meta.name = name; + if (!meta.name && meta.id) { + this.namelessToolCallIndices.add(actualIndex); + } else { + this.namelessToolCallIndices.delete(actualIndex); + } + return { complete: false }; + } + if (isKnownId && currentDepth === 0) { if (currentBuffer.trim()) { try { @@ -165,25 +187,10 @@ export class StreamingToolCallParser { } catch { // Not complete yet; append the incoming chunk below. } - } else if ( - this.toolCallMeta.get(actualIndex)?.name && - name && - !chunk.trim() - ) { - // The call at this index may be a completed no-argument call. An - // incoming chunk that carries a name but no argument content is an - // opener-shaped replay (duplicate ID) and must not mutate the - // surviving call; a chunk with argument content is a continuation - // for a call whose opener streamed empty arguments and must append. - debugLogger.debug( - `Ignoring replayed opener for no-argument toolCall id=${id}`, - ); - return { complete: false }; } } // Update metadata - const meta = this.toolCallMeta.get(actualIndex)!; if (id) meta.id = id; if (name) meta.name = name; @@ -194,6 +201,11 @@ export class StreamingToolCallParser { // Add chunk to buffer const newBuffer = currentBuffer + chunk; this.buffers.set(actualIndex, newBuffer); + if (!meta.name && (meta.id || /\S/.test(newBuffer))) { + this.namelessToolCallIndices.add(actualIndex); + } else { + this.namelessToolCallIndices.delete(actualIndex); + } // Track JSON structure depth - only count brackets/braces outside of strings let depth = currentDepth; @@ -260,6 +272,10 @@ export class StreamingToolCallParser { return this.toolCallMeta.get(index) || {}; } + hasNamelessToolCall(): boolean { + return this.namelessToolCallIndices.size > 0; + } + /** * Gets all completed tool calls that are ready to be emitted * @@ -443,6 +459,7 @@ export class StreamingToolCallParser { this.inStrings.set(index, false); this.escapes.set(index, false); this.toolCallMeta.set(index, {}); + this.namelessToolCallIndices.delete(index); } /** @@ -458,6 +475,7 @@ export class StreamingToolCallParser { this.inStrings.clear(); this.escapes.clear(); this.toolCallMeta.clear(); + this.namelessToolCallIndices.clear(); this.idToIndexMap.clear(); this.nextAvailableIndex = 0; } diff --git a/packages/core/src/core/openaiContentGenerator/types.ts b/packages/core/src/core/openaiContentGenerator/types.ts index 4e363f3df7e..3e638cd4da3 100644 --- a/packages/core/src/core/openaiContentGenerator/types.ts +++ b/packages/core/src/core/openaiContentGenerator/types.ts @@ -79,6 +79,11 @@ export interface RequestContext { * emitted after the reasoning thought if no tagged thought appears. */ pendingContentParts?: Part[]; + pendingUntrustedResponseParts?: Part[]; + hasStructuredReasoningContent?: boolean; + hasThinkingTagInReasoning?: boolean; + hasVisibleContent?: boolean; + atVisibleLineStart?: boolean; } export interface ErrorHandler { diff --git a/packages/core/src/telemetry/types.ts b/packages/core/src/telemetry/types.ts index 624b2122410..dea9cea23c0 100644 --- a/packages/core/src/telemetry/types.ts +++ b/packages/core/src/telemetry/types.ts @@ -676,10 +676,10 @@ export class ContentRetryEvent implements BaseTelemetryEvent { * the `onRetry` callback opt-in) for HTTP 429 / 5xx retries at LLM call sites. * * Distinct from {@link ContentRetryEvent}, which is emitted by `geminiChat`'s - * for-loop for `InvalidStreamError` retries that go through a SEPARATE retry - * budget (`INVALID_CONTENT_RETRY_OPTIONS`, NOT `retryWithBackoff`). A single - * user prompt may fire BOTH event types; sum across event types to count total - * retries per prompt_id. + * for-loop for `InvalidStreamError` retries that use + * `INVALID_STREAM_RETRY_CONFIG`, not `retryWithBackoff`. A single user prompt + * may fire BOTH event types; sum across event types to count total retries per + * prompt_id. */ export class ApiRetryEvent implements BaseTelemetryEvent { 'event.name': 'api_retry';