diff --git a/packages/core/src/core/geminiChat.test.ts b/packages/core/src/core/geminiChat.test.ts index d4faf3b4a14..64bb22dbe23 100644 --- a/packages/core/src/core/geminiChat.test.ts +++ b/packages/core/src/core/geminiChat.test.ts @@ -7272,7 +7272,7 @@ describe('GeminiChat', async () => { } }); - it('does not retry retryable transport stream errors after yielding a chunk', async () => { + it('does not retry retryable transport stream errors after yielding a content chunk', async () => { const transportError = Object.assign(new TypeError('terminated'), { cause: Object.assign(new Error('other side closed'), { code: 'UND_ERR_SOCKET', @@ -7322,6 +7322,132 @@ describe('GeminiChat', async () => { ).toBe(true); }); + it('retries a transport stream error after yielding only thinking chunks', async () => { + // Thinking models stream thought parts within seconds, then can + // spend minutes reasoning — exactly when gateways close long-lived + // SSE connections (#7832). Thought parts are ephemeral (never + // recorded as the assistant's response in history), so the replay + // cannot duplicate user-visible output and must be allowed. + vi.useFakeTimers(); + try { + const transportError = Object.assign(new TypeError('terminated'), { + cause: Object.assign(new Error('other side closed'), { + code: 'UND_ERR_SOCKET', + }), + }); + + vi.mocked(mockContentGenerator.generateContentStream) + .mockResolvedValueOnce( + (async function* () { + yield { + candidates: [ + { + content: { + parts: [ + { text: 'Let me think about this…', thought: true }, + ], + }, + }, + ], + } as unknown as GenerateContentResponse; + throw transportError; + })(), + ) + .mockResolvedValueOnce( + (async function* () { + yield { + candidates: [ + { + content: { + parts: [{ text: 'Recovered after thinking-phase retry' }], + }, + finishReason: 'STOP', + }, + ], + } as unknown as GenerateContentResponse; + })(), + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-transport-retry-after-thinking', + ); + const events = await collectStreamWithFakeTimers(stream, 5_000); + + expect( + mockContentGenerator.generateContentStream, + ).toHaveBeenCalledTimes(2); + expect( + events.filter((event) => event.type === StreamEventType.RETRY), + ).toHaveLength(1); + expect( + events.some( + (event) => + event.type === StreamEventType.CHUNK && + event.value.candidates?.[0]?.content?.parts?.[0]?.text === + 'Recovered after thinking-phase retry', + ), + ).toBe(true); + } finally { + vi.useRealTimers(); + } + }); + + it('does not retry when visible content followed the thinking chunks', async () => { + // The content flag must accumulate across the whole attempt: once a + // non-thought part has flowed — even after any amount of thinking — + // a replay would duplicate visible output and stays blocked. + const transportError = Object.assign(new TypeError('terminated'), { + cause: Object.assign(new Error('other side closed'), { + code: 'UND_ERR_SOCKET', + }), + }); + + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + (async function* () { + yield { + candidates: [ + { + content: { + parts: [{ text: 'Reasoning first…', thought: true }], + }, + }, + ], + } as unknown as GenerateContentResponse; + yield { + candidates: [ + { + content: { + parts: [{ text: 'Visible answer begins' }], + }, + }, + ], + } as unknown as GenerateContentResponse; + throw transportError; + })(), + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-transport-no-retry-after-thinking-then-content', + ); + const events: StreamEvent[] = []; + await expect(async () => { + for await (const event of stream) { + events.push(event); + } + }).rejects.toThrow('terminated'); + + expect(mockContentGenerator.generateContentStream).toHaveBeenCalledTimes( + 1, + ); + expect( + events.filter((event) => event.type === StreamEventType.RETRY), + ).toHaveLength(0); + }); + it('retries a transport stream error after yielding only tool preparation metadata', async () => { vi.useFakeTimers(); try { diff --git a/packages/core/src/core/geminiChat.ts b/packages/core/src/core/geminiChat.ts index 5fcfd0b14fe..974f52a8e76 100644 --- a/packages/core/src/core/geminiChat.ts +++ b/packages/core/src/core/geminiChat.ts @@ -140,6 +140,25 @@ function isToolCallPreparationOnly(response: GenerateContentResponse): boolean { return !hasCandidateOutput && !response.usageMetadata; } +/** + * True when the chunk carries model output beyond ephemeral reasoning: + * any candidate part without the `thought` flag (text, functionCall, + * inlineData, …). Thought parts stream reasoning that is never recorded + * as the assistant's final response in history, so replaying a request + * that has produced only thought parts cannot duplicate user-visible + * output — the distinction the transport stream retry gate relies on + * (#7832). + */ +function hasNonThoughtCandidateParts( + response: GenerateContentResponse, +): boolean { + return Boolean( + response.candidates?.some((candidate) => + candidate.content?.parts?.some((part) => !part.thought), + ), + ); +} + function syncFunctionCallsField( response: GenerateContentResponse, parts: readonly Part[], @@ -2462,6 +2481,7 @@ export class GeminiChat { for (;;) { let streamYieldedChunk = false; + let streamYieldedContentChunk = false; try { if (suppressNextRetryEvent) { suppressNextRetryEvent = false; @@ -2487,6 +2507,9 @@ export class GeminiChat { streamYieldedChunk = true; streamYieldedAnyChunk = true; } + if (hasNonThoughtCandidateParts(chunk)) { + streamYieldedContentChunk = true; + } const fr = chunk.candidates?.[0]?.finishReason; if (fr) lastFinishReason = fr; yield { type: StreamEventType.CHUNK, value: chunk }; @@ -2590,8 +2613,14 @@ export class GeminiChat { }); } - // Replay only curated socket-level failures before any response - // chunk has reached callers. + // Replay only curated socket-level failures before any + // user-visible content has reached callers. Thinking-only + // output does not block the replay: thought parts are + // ephemeral (never recorded as the assistant's response in + // history), so retrying after them cannot duplicate visible + // output — and thinking models can spend minutes in that + // phase, exactly when gateways close long-lived SSE + // connections (#7832). const isRetryableStreamTransportError = classification.kind === 'transport' && classification.transportCode !== undefined && @@ -2600,7 +2629,7 @@ export class GeminiChat { ); if ( isRetryableStreamTransportError && - !streamYieldedChunk && + !streamYieldedContentChunk && transportStreamRetryCount < TRANSPORT_STREAM_RETRY_CONFIG.maxRetries ) { @@ -2615,6 +2644,7 @@ export class GeminiChat { attempt: transportStreamRetryCount, maxRetries: TRANSPORT_STREAM_RETRY_CONFIG.maxRetries, retryDelayMs: delayMs, + yieldedNonContentChunks: streamYieldedChunk, errorKind: classification.kind, transportCode: classification.transportCode, }); @@ -2624,13 +2654,14 @@ export class GeminiChat { continue; } if (isRetryableStreamTransportError) { - // Reached only when the retry above did not fire: either a chunk - // was already yielded (replaying would duplicate output) or the - // retry budget is exhausted. Either way the error propagates. + // Reached only when the retry above did not fire: either + // user-visible content was already yielded (replaying would + // duplicate it) or the retry budget is exhausted. Either way + // the error propagates. debugLogger.warn('Transport stream retry not taken', { retryPath: 'stream', - retryDecision: streamYieldedChunk - ? 'skipped_after_chunk' + retryDecision: streamYieldedContentChunk + ? 'skipped_after_content' : 'exhausted', attempts: transportStreamRetryCount, maxRetries: TRANSPORT_STREAM_RETRY_CONFIG.maxRetries,