fix(core): allow transport stream retry during the thinking-only phase

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 {

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,