fix(core): align parked-finish flush and Anthropic release gate with LlmChat recovery

packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.test.ts

@@ -3629,46 +3629,69 @@ describe('AnthropicContentGenerator', () => {
       ]);
     });
 
-    it('releases closed valid tool calls before rethrowing an upstream stream error', async () => {
-      const networkError = Object.assign(
-        new Error('SSE connection reset by peer'),
-        { code: 'ECONNRESET' },
-      );
-      anthropicState.createImpl.mockResolvedValue(
-        (async function* interruptedToolUseStream() {
-          yield {
-            type: 'content_block_start',
-            index: 0,
-            content_block: {
-              type: 'tool_use',
-              id: 'call-complete',
-              name: 'read_file',
-              input: {},
-            },
-          };
-          yield {
-            type: 'content_block_delta',
-            index: 0,
-            delta: {
-              type: 'input_json_delta',
-              partial_json: '{"file_path":"a.sql"}',
-            },
-          };
-          yield { type: 'content_block_stop', index: 0 };
-          throw networkError;
-        })(),
-      );
-      const { chunks, error } = await collectGeneratedStream();
+    it.each([
+      {
+        case: 'a retryable socket cut',
+        error: Object.assign(new Error('SSE connection reset by peer'), {
+          code: 'ECONNRESET',
+        }),
+      },
+      {
+        // A gateway error frame pushed into an already-200 stream carries
+        // neither a status nor an allow-listed socket code; the provider's
+        // own request id is what classifies it. LlmChat's mid-stream
+        // boundary admits this class, so the release gate does too: the
+        // delivered functionCall shuts both recovery gates there, and the
+        // error-path persistence plus the scheduler's repair flow take over
+        // — the same footing a socket cut produces. The fixture must carry
+        // no allow-listed socket code at any cause level, or it classifies
+        // as transport and the case no longer exercises the status-less
+        // disjunct.
+        case: 'a status-less upstream error the provider traced with a request id',
+        error: Object.assign(new Error("'id'"), {
+          code: 'KeyError',
+          requestID: 'req-1',
+        }),
+      },
+    ])(
+      'releases closed valid tool calls before rethrowing $case',
+      async ({ error: upstreamError }) => {
+        anthropicState.createImpl.mockResolvedValue(
+          (async function* interruptedToolUseStream() {
+            yield {
+              type: 'content_block_start',
+              index: 0,
+              content_block: {
+                type: 'tool_use',
+                id: 'call-complete',
+                name: 'read_file',
+                input: {},
+              },
+            };
+            yield {
+              type: 'content_block_delta',
+              index: 0,
+              delta: {
+                type: 'input_json_delta',
+                partial_json: '{"file_path":"a.sql"}',
+              },
+            };
+            yield { type: 'content_block_stop', index: 0 };
+            throw upstreamError;
+          })(),
+        );
+        const { chunks, error } = await collectGeneratedStream();
 
-      expect(chunks.flatMap((chunk) => chunk.functionCalls ?? [])).toEqual([
-        {
-          id: 'call-complete',
-          name: 'read_file',
-          args: { file_path: 'a.sql' },
-        },
-      ]);
-      expect(error).toBe(networkError);
-    });
+        expect(chunks.flatMap((chunk) => chunk.functionCalls ?? [])).toEqual([
+          {
+            id: 'call-complete',
+            name: 'read_file',
+            args: { file_path: 'a.sql' },
+          },
+        ]);
+        expect(error).toBe(upstreamError);
+      },
+    );
 
     it.each([
       {
@@ -3687,19 +3710,6 @@ describe('AnthropicContentGenerator', () => {
           code: 'ECONNREFUSED',
         }),
       },
-      {
-        // LlmChat's mid-stream boundary deliberately admits this class while
-        // this release gate stays transport-only: releasing the closed batch
-        // would flip the delivered-output flags and shut both of LlmChat's
-        // recovery gates. The fixture must carry no allow-listed socket code
-        // at any cause level, or it classifies as transport and the batch IS
-        // released — the opposite of what this case pins.
-        case: 'a status-less upstream error the provider traced with a request id',
-        error: Object.assign(new Error("'id'"), {
-          code: 'KeyError',
-          requestID: 'req-1',
-        }),
-      },
     ])(
       'does not release a closed call before rethrowing $case',
       async ({ error }) => {

packages/core/src/core/anthropicContentGenerator/anthropicContentGenerator.ts

@@ -56,7 +56,10 @@ import { parseToolCallArguments } from '../tool-call-arguments.js';
 import { classifyRetryError } from '../../utils/retryErrorClassification.js';
 import { getErrorStatus } from '../../utils/errors.js';
 import { buildSessionAwareFetch } from '../outbound-session-id.js';
-import { isRetryableStreamTransportError } from '../stream-transport-retry.js';
+import {
+  isRetryableStatuslessUpstreamError,
+  isRetryableStreamTransportError,
+} from '../stream-transport-retry.js';
 import {
   reportAnthropicEvent,
   reportAnthropicFollowingRequest,
@@ -1621,17 +1624,21 @@ export class AnthropicContentGenerator implements ContentGenerator {
     if (upstreamStreamFailed) {
       const upstreamErrorClassification =
         classifyRetryError(upstreamStreamError);
-      // Narrower than LlmChat's replay boundary on purpose. LlmChat also
-      // admits a status-less upstream failure the provider traced with its own
-      // request id; releasing a closed batch here would flip what LlmChat then
-      // sees as already delivered (`streamYieldedContentChunk`,
-      // `streamYieldedFunctionCall`) and shut both of its recovery gates, so
-      // for that class the batch stays withheld and the error propagates,
-      // leaving LlmChat free to replay or continue — which is also what this
-      // path did before that class was classified at all. Only known mid-SSE
-      // socket cuts release.
+      // Match LlmChat's replay boundary: known mid-SSE socket cuts and
+      // status-less upstream failures the provider traced with a request id
+      // both release an already closed batch before the error propagates.
+      // Releasing keeps the two providers' functionCall cuts on one footing —
+      // the delivered call flips LlmChat's delivered flags
+      // (`streamYieldedContentChunk`, `streamYieldedFunctionCall`), which
+      // shuts replay and continuation, and the error-path persistence plus
+      // the scheduler's repair flow take over. Withholding would instead
+      // leave a resume over prose as the only recovery once answer text has
+      // been delivered: a withheld batch never sets
+      // `streamYieldedFunctionCall`, so the model would be asked to continue
+      // an answer whose tool call it never saw.
       if (
-        isRetryableStreamTransportError(upstreamErrorClassification) &&
+        (isRetryableStreamTransportError(upstreamErrorClassification) ||
+          isRetryableStatuslessUpstreamError(upstreamErrorClassification)) &&
         deferredToolCalls.length > 0 &&
         !hasEmptyToolCall &&
         !hasMalformedToolCall &&

packages/core/src/core/llm-chat.test.ts

@@ -10913,6 +10913,76 @@ describe('LlmChat', async () => {
         ).toHaveBeenCalledTimes(1);
       });
 
+      it('does not continue a status-less upstream error that delivered a functionCall', async () => {
+        // The continuation gate admits a status-less upstream failure for
+        // prose, but a delivered functionCall excludes it exactly as for a
+        // socket cut. The Anthropic deferred-batch release relies on this
+        // exclusion: it releases a closed tool call ahead of this error
+        // class so the call reaches error-path persistence and the
+        // scheduler's repair flow, instead of the model being asked to
+        // resume prose whose tool decision it never saw.
+        const upstreamError = Object.assign(new Error("'id'"), {
+          code: 'KeyError',
+          requestID: 'cd7f37f3-d38a-9dec-804f-f70dda5650eb',
+        });
+        const toolChunk = {
+          candidates: [
+            {
+              content: {
+                role: 'model',
+                parts: [
+                  {
+                    functionCall: {
+                      id: 'call_1',
+                      name: 'read_file',
+                      args: { path: '/tmp/a.txt' },
+                    },
+                  },
+                ],
+              },
+            },
+          ],
+        } as unknown as GenerateContentResponse;
+
+        vi.mocked(mockContentGenerator.generateContentStream)
+          .mockResolvedValueOnce(
+            (async function* () {
+              yield textChunk('Let me read that file. ');
+              yield toolChunk;
+              throw upstreamError;
+            })(),
+          )
+          // Tripwire: consumed only if the cut is wrongly resumed.
+          .mockResolvedValueOnce(
+            (async function* () {
+              yield textChunk('fabricated tail', 'STOP');
+            })(),
+          );
+
+        const stream = await chat.sendMessageStream(
+          'test-model',
+          { message: 'test' },
+          'prompt-upstream-statusless-functioncall',
+        );
+        const { events, caughtError } = await drainCollecting(stream);
+
+        expect(
+          mockContentGenerator.generateContentStream,
+        ).toHaveBeenCalledTimes(1);
+        expect(
+          events.filter((event) => event.type === StreamEventType.RETRY),
+        ).toHaveLength(0);
+        expect(caughtError).toBe(upstreamError);
+        // The delivered functionCall is persisted on the error path, paired
+        // for the scheduler's repair flow.
+        expect(
+          chat
+            .getHistory()
+            .at(-1)
+            ?.parts?.some((part) => part.functionCall),
+        ).toBe(true);
+      });
+
       it('replays rather than continues when only a thought was delivered', async () => {
         // The reported failure mode: thinking models emit reasoning within
         // seconds, so gating replay on "any chunk yielded" made it

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

@@ -3336,16 +3336,19 @@ describe('ContentGenerationPipeline', () => {
         async *[Symbol.asyncIterator]() {
           yield {
             id: 'finish-chunk',
-            choices: [{ delta: {}, finish_reason: 'tool_calls' }],
+            choices: [{ delta: {}, finish_reason: 'stop' }],
           } as OpenAI.Chat.ChatCompletionChunk;
           throw streamError;
         },
       };
+      // The error-path flush never releases a functionCall-bearing finish
+      // (see the withhold test below), so this fixture's finish carries a
+      // text part: the log pin needs a finish the flush actually delivers.
       const finishResponse = new GenerateContentResponse();
       finishResponse.responseId = 'finish-response';
       finishResponse.candidates = [
         {
-          content: { parts: [{ functionCall: { name: 'read_file' } }] },
+          content: { parts: [{ text: 'sanitized answer' }] },
           finishReason: FinishReason.STOP,
           index: 0,
         },
@@ -3356,7 +3359,7 @@ describe('ContentGenerationPipeline', () => {
         (_chunk, context) => {
           context.protocolTagSanitized = {
             tagName: 'think',
-            toolCallCount: 1,
+            toolCallCount: 0,
           };
           return finishResponse;
         },
@@ -5045,6 +5048,140 @@ describe('ContentGenerationPipeline', () => {
       await expect(iterator.next()).rejects.toThrow("'id'");
       expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
     });
+
+    it('flushes a parked finish response ahead of a stream-guard timeout', async () => {
+      // A provider that emits content plus `finish_reason: 'stop'` and then
+      // goes silent — a hung or drip-fed gateway after the answer — trips
+      // the inactivity watchdog while the finish chunk is still parked for
+      // the usage merge. Downstream completeness gates key on the finish
+      // reason, so the parked response must reach the caller ahead of the
+      // guard's rethrow; a flush placed below the guard branch never runs
+      // for this error class.
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const userPromptId = 'test-prompt-id';
+
+      const mockChunk1 = {
+        id: 'chunk-1',
+        choices: [
+          { delta: { content: 'a complete answer' }, finish_reason: null },
+        ],
+      } as OpenAI.Chat.ChatCompletionChunk;
+      const mockChunk2 = {
+        id: 'chunk-2',
+        choices: [{ delta: { content: '' }, finish_reason: 'stop' }],
+      } as OpenAI.Chat.ChatCompletionChunk;
+
+      const guardError = new StreamInactivityTimeoutError(240_000, 2, 240_500);
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield mockChunk1;
+          yield mockChunk2;
+          throw guardError;
+        },
+      };
+
+      const mockContentResponse = new GenerateContentResponse();
+      mockContentResponse.candidates = [
+        { content: { parts: [{ text: 'a complete answer' }], role: 'model' } },
+      ];
+      const mockFinishResponse = new GenerateContentResponse();
+      mockFinishResponse.candidates = [
+        {
+          content: { parts: [], role: 'model' },
+          finishReason: FinishReason.STOP,
+        },
+      ];
+
+      (mockConverter.convertLlmRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockConverter.convertOpenAIChunkToLlm as Mock)
+        .mockReturnValueOnce(mockContentResponse)
+        .mockReturnValueOnce(mockFinishResponse);
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        userPromptId,
+      );
+      const iterator = resultGenerator[Symbol.asyncIterator]();
+
+      const contentResult = await iterator.next();
+      if (contentResult.done) throw new Error('Expected a content response.');
+      expect(contentResult.value).toBe(mockContentResponse);
+
+      // The parked finish is flushed ahead of the guard's rethrow.
+      const finishResult = await iterator.next();
+      if (finishResult.done) throw new Error('Expected a finish response.');
+      expect(finishResult.value.candidates?.[0]?.finishReason).toBe(
+        FinishReason.STOP,
+      );
+
+      // The caller still rejects with the dedicated timeout error instance —
+      // its type and idle/chunk metadata intact — because the guard branch
+      // bypasses handleError.
+      await expect(iterator.next()).rejects.toBe(guardError);
+      expect(mockErrorHandler.handle).not.toHaveBeenCalled();
+    });
+
+    it('withholds a parked functionCall finish when the stream fails before the trailing tail', async () => {
+      // The converter emits functionCall parts only on the finish chunk, and
+      // streaming always parks that chunk for the trailing usage metadata.
+      // Releasing it ahead of the error would flip LlmChat's delivered flags
+      // (streamYieldedContentChunk, streamYieldedFunctionCall) and shut the
+      // transport replay gate that recovers exactly this cut, while the
+      // post-completion acceptance arm the flush feeds excludes tool calls
+      // anyway — so a tool-call finish stays parked on the error path.
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const streamError = new Error('stream failed after finish');
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'tool_calls' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          throw streamError;
+        },
+      };
+      const finishResponse = new GenerateContentResponse();
+      finishResponse.candidates = [
+        {
+          content: { parts: [{ functionCall: { name: 'read_file' } }] },
+          finishReason: FinishReason.STOP,
+          index: 0,
+        },
+      ];
+
+      (mockConverter.convertLlmRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockConverter.convertOpenAIChunkToLlm as Mock).mockReturnValue(
+        finishResponse,
+      );
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        'test-prompt-id',
+      );
+
+      const results: GenerateContentResponse[] = [];
+      await expect(async () => {
+        for await (const result of resultGenerator) {
+          results.push(result);
+        }
+      }).rejects.toThrow(streamError);
+      // The tool-call finish stays parked: the caller sees no chunk, only
+      // the rejection.
+      expect(results).toEqual([]);
+      expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
+    });
   });
 
   describe('buildResponseFormat endpoint gate', () => {

packages/core/src/core/openaiContentGenerator/pipeline.ts

@@ -727,6 +727,36 @@ export class ContentGenerationPipeline {
         throw error;
       }
 
+      // A finish chunk parked for the usage merge must not be lost when the
+      // iterator throws before the trailing usage chunk arrives (e.g. a
+      // gateway error frame landing where that tail would have been):
+      // downstream completeness gates key on the finish reason to tell a
+      // completed answer from a cut one. The flush sits below the
+      // InvalidStreamError rethrow — a protocol-tag-leak stream must not
+      // deliver one — and above the guard and StreamContentError rethrows so
+      // every recoverable error class still sees it. `handleError` never
+      // returns, so the Stage 2d flush above cannot double-yield this
+      // response.
+      //
+      // A parked finish carrying a functionCall stays parked: the converter
+      // emits functionCall parts only on the finish chunk, and releasing one
+      // here would flip LlmChat's delivered flags (streamYieldedContentChunk,
+      // streamYieldedFunctionCall) and shut the transport replay gate that
+      // recovers exactly this cut, while the post-completion acceptance arm
+      // this flush feeds excludes tool calls anyway.
+      const parkedHasToolCall = pendingFinishResponse?.candidates?.some(
+        (candidate) =>
+          candidate.content?.parts?.some((part) => part.functionCall),
+      );
+      if (pendingFinishResponse && !finishYielded && !parkedHasToolCall) {
+        logPendingProtocolTagSanitized(
+          pendingFinishResponse,
+          pendingFinishProtocolTagSanitized,
+        );
+        yield pendingFinishResponse;
+        finishYielded = true;
+      }
+
       // 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) {
@@ -778,21 +808,6 @@ export class ContentGenerationPipeline {
         );
       }
 
-      // A finish chunk parked for the usage merge must not be lost when the
-      // iterator throws before the trailing usage chunk arrives (e.g. a
-      // gateway error frame landing where that tail would have been):
-      // downstream completeness gates key on the finish reason to tell a
-      // completed answer from a cut one. `handleError` never returns, so the
-      // Stage 2d flush above cannot double-yield this response.
-      if (pendingFinishResponse && !finishYielded) {
-        logPendingProtocolTagSanitized(
-          pendingFinishResponse,
-          pendingFinishProtocolTagSanitized,
-        );
-        yield pendingFinishResponse;
-        finishYielded = true;
-      }
-
       // Use shared error handling logic
       await this.handleError(error, context, request);
     }

packages/core/src/core/stream-transport-retry.ts

@@ -33,12 +33,11 @@ export function isRetryableStreamTransportError(
  * status survived — a gateway error frame pushed into an already-200 SSE
  * stream, which the OpenAI SDK surfaces with neither a status nor a socket
  * code. This needs its own predicate instead of a wider
- * `RETRYABLE_STREAM_TRANSPORT_CODES`: that set is socket-only by contract, and
- * the Anthropic generator shares the predicate over it to decide whether
- * deferred tool-call chunks may be released. That release gate deliberately
- * stays transport-only, so it is narrower than LlmChat's replay boundary, which
- * admits this class too; the comment at its call site in the Anthropic
- * generator records why releasing there would disable the recovery.
+ * `RETRYABLE_STREAM_TRANSPORT_CODES`: that set is socket-only by contract.
+ * LlmChat's replay/continuation boundary admits this class alongside socket
+ * cuts, and the Anthropic generator's deferred tool-call release gate reads
+ * both predicates so a closed batch reaches LlmChat on the same footing on
+ * either provider.
  */
 export function isRetryableStatuslessUpstreamError(
   classification: RetryErrorClassification,