fix(core): retry status-less upstream errors instead of ending the turn (#11291)

* fix(core): retry status-less upstream errors instead of ending the turn

A gateway that fails mid-stream — pushing `{"error": {...}}` into an
already-200 SSE response — reaches us as an OpenAI SDK APIError carrying no
HTTP status: the body's `code` and `message`, plus the response's
`x-request-id`. Observed in the wild as `code: 'KeyError'`,
`message: "'id'"`, a server-side defect that ended a four-hour turn on the
first attempt, while socket cuts on the same session were retried without
incident.

The retry classifier had no branch for that shape, so it fell through to
'unknown', and both status-less gates — the default predicate and the stream
path's inline one — matched transport codes only. They now defer to the
classification's verdict, and a status-less error carrying the provider's own
request id classifies as retryable.

The request id is the whole gate. Keying on the provider `code` instead would
sweep in permanent local failures that carry a string code and no status — a
missing API key, an invalid MCP server config, MCP's numeric JSON-RPC codes —
and walk them through the full retry ladder. Only an upstream request id
evidences that the provider accepted and traced the request before failing.
The SDK spells that field `requestID`, which the classifier did not read at
all, so it is now recognized alongside `request_id`/`requestId`.

Retries stay bounded: persistent mode still requires a 429/529 status, model
fallback still requires a status, and mid-stream replay after partial output
still requires a transport classification — so a status-less upstream error
gets the normal seven attempts and can never duplicate user-visible text.

Tests pin the observed shape both as SDK properties and as a provider body
embedded in the message, drive both gates, and guard the two verdicts that
must not move: a code-only local failure and a traced 4xx.

* fix(core): honour the retry verdict where mid-stream errors actually land

Review round on #11291. The status-less upstream class the previous commit
taught to the classifier never reached either gate it widened: both run while
the stream is being established, and the OpenAI SDK builds that error inside
the lazy SSE iterator, after establishment has already resolved. The mid-stream
catch computed the classification and then ignored it — its only retry exits
required a socket code — so the reported turn still died on the first attempt.

The catch now admits the class to both recovery mechanisms beside the transport
one, through a predicate of its own: replay when nothing user-visible has been
delivered, continuation from the delivered text once it has. Both keep the
existing budgets and every no-duplication precondition, and the socket
allow-list is untouched because the Anthropic generator shares it to decide
whether deferred tool-call chunks may be released.

A request id separates upstream from local, not transient from permanent, so
the branch now fails fast on the provider codes that can never succeed on a
resend — moderation, credentials/billing, malformed request — matched case- and
separator-insensitively, while an unrecognised code such as the observed
KeyError stays retryable. An empty request id no longer opens the gate: the
header can be present and empty, and the value is now load-bearing.

The status-less verdict is computed once and shared by both establishment
gates, which had each copied it, and an SSE-delivered instance now reports the
same kind as its siblings. The comments that described the classification as
driving one control decision name the decisions it actually drives.

Tests drive the mid-stream path by throwing from the generator's first next()
instead of rejecting establishment, and cover the continuation path, the
permanent rejection that must still propagate, the transport-over-request-id
ordering, and both empty-id readers. Each was mutation-checked: removing the
production change it pins turns it red.

* test(core): stop leaving the new rejection fixtures unhandled

CI failed the unit job on three unhandled rejections while all 23,749 tests
passed. All three came from tests the previous commit added.

Two were in the retry predicate cases: advancing fake timers between creating
the promise and asserting on it left the rejection with no handler for a tick.
The advance is gone and the assertion attaches immediately, matching the other
non-retry cases in the file — the crispness it bought on a regression is not
worth an unhandled rejection, and the case still reds, by test timeout.

The third drove a stream helper that builds its collector, awaits two timer
steps, and only then hands the collector back, so a stream rejecting in between
is briefly unhandled. That test now drains through the helper that catches
inside the loop, and drops fake timers it never needed: nothing retries on the
path it asserts, so there is no backoff to advance through.

Both still discriminate. With the permanent-code list disabled the retry case
reports three calls instead of one, and the mid-stream case reports a second
generator call in about a second rather than hanging.

* fix(core): keep oversized and malformed rejections out of the new retry gate

Third review round on #11291. Widening the mid-stream recovery gate exposed an
ordering fault of its own: the gate is consulted before the one-shot overflow
recovery below it, so a status-less input-length rejection was re-sent instead
of compacted, and the continuation arm re-sent it strictly larger each time.
Both overflow detectors are now computed above the gate and their hits excluded
from the widened verdict — from that verdict only, so a socket cut keeps the
verdict it had before this branch existed.

A request id separates upstream from local but says nothing about permanence,
and the permanence list read only `code`. The OpenAI SDK puts
`invalid_request_error` on `.type` with `code` null, so the canonical
malformed-request body never matched and a rejection that cannot succeed was
re-sent through the whole ladder. `.type` is now a second input to that guard,
matched against the same anchored list, and kept out of the classification
struct because `providerCode` is compared by exact equality elsewhere.

The list also grows the codes the previous round measured by name — billing
exhaustion, model access, output moderation, the OpenAI malformed-parameter
spelling, and oversized input — while staying anchored on purpose: a sub-code
qualifier left unmatched costs a wasted ladder, whereas over-matching costs a
transient failure that is never retried, which is the defect this branch exists
to remove.

The three stream-recovery logs now carry the classification's reason, provider
code and request id. They previously recorded only a kind and a socket code, so
the class this PR recovers logged a label pointing at sockets and proxies that
were never involved, and discarded the one field a gateway ticket can be filed
against.

Two comments that this branch falsified are corrected: the Anthropic release
gate no longer claims parity with a replay boundary that has since widened, and
the `sse-provider` gloss no longer claims the label means "delivered by the
stream" when it records raw framing surviving into the message.

Tests: a status-less overflow must reach compaction rather than replay, a
permanent `.type` must fail fast while `server_error` stays retryable, the
extended code list, and an oracle that drives the real SDK constructor with
real headers so a dependency bump renaming `requestID` goes red instead of
silently reintroducing the incident. Each was mutation-checked in both
directions where the assertion could be satisfied by accident.

* fix(core): stop resuming answers that already delivered a finish reason

Fourth review round on #11291. The continuation arm the previous round
widened to status-less upstream errors had no completeness precondition:
a gateway error frame that lands after the chunk carrying a terminal
finishReason — the SDK's error scan is position-independent, and the
pipeline keeps pulling the iterator to absorb trailing usage metadata —
was resumed as if the answer had been cut. The resume instruction
asserted a connection drop that never happened, and the fabricated tail
was folded into durable history and the JSONL record. The gate now only
continues an attempt that ended without a finish reason, or with
MAX_TOKENS — the truncated shape the arm exists for.

The remaining round-4 findings were pinning gaps, each closed with a
test and mutation-checked: the payload-overflow half of the R3-1
exclusion (a status-less 413 reason phrase now has a compaction case of
its own), the classifier fields on the continuation-scheduled and
retry-not-taken logs, the Anthropic release gate's deliberate narrowness
for a traced status-less error, the `.type` permanence guard driven
through the real SDK constructor, and the message-scraped half of the
permanent-provider-code read — both the SSE-framed body and the nested
`.error.code` shape.

* fix(core): accept answers that completed before a trailing stream failure

The completeness guard keyed on lastFinishReason could never fire on the
real throw path: the OpenAI pipeline parks the finish chunk for the usage
merge and discarded it when the iterator threw, so a gateway error frame
landing after a finished answer looked identical to a cut mid-answer.

Flush the parked finish response ahead of the propagated error so the
completion signal reaches LlmChat, and decide the post-completion case
explicitly: a stream that fails after a closed finish reason (STOP,
SAFETY, ...) with content already delivered is accepted as the turn's
outcome and persisted, instead of failing the turn or fabricating a
continuation tail into durable history. The veto is now a deny-list of
closed reasons so unmapped values (FINISH_REASON_UNSPECIFIED) fail open
to continuable, and the not-taken log names the terminal finish reason
as the decline cause.

* test(core): annotate closure-captured results array in pipeline test

The array is pushed from inside an async closure, so the evolving-any
inference does not apply and tsc --build rejected it under noImplicitAny
(TS7034/TS7005). Give it the explicit GenerateContentResponse[] type used
elsewhere in the file.

Co-authored-by: Qwen-Coder <[email protected]>

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

* fix(core): cast pendingFinishResponse at error-path tool-call check

The deterministic gate rejected 7f7e010e99: packages/core failed tsc
with TS2339/TS7006 at pipeline.ts:747-749. TypeScript control-flow
analysis narrows pendingFinishResponse to null at the catch block (its
only assignments sit inside the handleChunkMerging callback), so the
new parkedHasToolCall property access read type never. Apply the same
explicit cast this function already uses in the finishYielded merge
above. Runtime behavior is unchanged.

* fix(core): release parked tool-call finish once content was delivered

The error-path flush withheld a parked finish carrying a functionCall
unconditionally, but the transport replay gate it protects is already
shut once user-visible content has been delivered. On the ordinary
agent-turn shape (prose, then a parked tool_calls finish, then a
gateway error frame) the withhold stranded the model's decided tool
call: LlmChat saw delivered prose, no functionCall, and no finish
reason, so the continuation arm resumed over the prose and the turn
was recorded as a fabricated success. Withhold only while nothing
visible was delivered (replay still open); release afterwards so the
delivered call shuts continuation and error-path persistence plus the
scheduler's repair flow take over — the Anthropic footing.

Co-authored-by: Qwen-Coder <[email protected]>

* fix(core): release parked tool-call finish on continuation attempts

* fix(core): stop post-completion gates from falsifying turn outcomes

- The trailing-failure acceptance gate nulled streamError for every
  error class, converting user cancels, pipeline InvalidStreamError
  rejections, and throttling StreamContentErrors into accepted
  completions. Accept only the failure classes the transport recovery
  gates themselves admit (socket cuts and status-less upstream frames).
- On tool-result sends the continuation veto never saw a closed finish
  reason: processStreamResponse defers the reason off yielded chunks and
  a failed attempt never re-emits it. Record the observed close per
  attempt and let the veto and the not-taken log consult it, so a closed
  answer no longer earns a fabricated continuation tail.

* fix(core): classify a trailing stream failure with the send loop's context

The gate that keeps a completed answer from failing on a trailing stream
error decides by classification, but it classified without the auth type
and configured retry codes that every other classification in the send
loop passes. A provider code only the caller's configured
`retryErrorCodes` mark as throttling carries a request id and no status,
so it read as a status-less upstream frame — the one class this gate
accepts — and a throttled turn was certified as a completed one instead
of reaching the rate-limit retry.

* fix(core): raise the pipeline's delivery flags before suspending

Two yield sites recorded what they had delivered after the yield, so a
consumer that throws into the generator while it is suspended there never
ran the assignment. The error-path flush re-tests `finishYielded` before
deciding whether a parked finish still needs delivering, and it therefore
handed the same response object back a second time, folding every part it
carries into the persisted turn twice.

The held-parts flush also yielded without updating `contentYielded` at all.
Held parts can carry non-thought content, which LlmChat counts as delivered
and which shuts its transport replay gate; with the two flags disagreeing,
the flush withheld a parked tool call that no recovery arm could pick up
any more. Held parts that are thought-only still leave the flag false, so
the withhold keeps protecting the gate it exists for.

Both are latent rather than live: nothing in production initiates a throw
into this chain, the two `.throw()` sites in it only forward one, and a
`for await` consumer abandons through `.return()`, which does not run the
catch.

* test(core): pin the Anthropic status-less path to the real SDK shape

The Anthropic release gate's status-less disjunct was justified as putting
the two providers on one footing, but the only fixture behind it hand-built
an OpenAI-shaped error, so nothing showed which Anthropic failures can
actually reach that branch. Drive the SDK's own mid-stream constructor
instead: it short-circuits on the missing status and builds an
APIConnectionError without headers, so the response's request-id never
becomes `request_id`, while the raw frame survives as the message. An id
relayed inside that body still opens the branch, and a permanent Anthropic
`type` in the same body fails fast before the request-id branch can see it.
Scope the gate's comment to that narrower set of producers.

* fix(core): scope the acceptance gate to the turn, fail fast on auth types

Two fails-closed defects from round 11.

The trailing-failure acceptance gate measured completeness with the
attempt's own visible text. On a continuation the text already delivered
lives in the turn-scoped prefix, and the merge that folds it back in runs
below the gate and only once the error is cleared, so an attempt that closed
a continuation without adding prose was refused — and no other arm owns that
failure: replay needs an empty prefix, continuation is vetoed by the same
close, and the rate-limit, overflow and invalid-stream arms do not match a
status-less frame. The turn threw, and the prose the caller watched stream
reached neither history nor the JSONL record. The gate now reads the
turn-scoped text and shares the validation block's own completeness
bindings, so it cannot accept a turn that block would then reject — which is
what keeps the post-tool-result quiet shape, and an attempt that delivered
nothing of its own, on the paths they have today.

The permanent provider-code list caught only `invalid_request_error` out of
the pinned Anthropic error-type union, so a gateway relaying
`authentication_error`, `permission_error`, `not_found_error` or
`billing_error` into an already-200 stream classified as retryable and walked
the whole ladder for a verdict fixed on arrival. Those are the types the SDK
maps to 401/403/404, all of which the HTTP-status branch already fails fast
on; `request_too_large` joins the oversized-payload codes already here. The
union's transient members stay out and are now pinned in both directions.

* test(core): pin the quiet tool-result close the acceptance gate declines

The gate's `lacksVisibleToolResultProgress` term keeps a tool-result
continuation that closed without new visible text off the acceptance path.
Whether that strands a completed answer was raised on review and is not
answerable from the code alone, so pin the measurement: with no trailing
error on the closing attempt, the same shape retries under #7039 and the
prose an earlier attempt delivered is discarded anyway, because the fresh
restart drops the continuation prefix. The discard is therefore the
quiet-tool-result policy's and not the gate's, and the asymmetry that made
the non-continuation case a defect — there the identical attempt without
the error is accepted and persisted — does not hold on this shape.

* docs(core): replace two measured-false warrants in the retry verdict

Both comments justified code by claiming what a mutation would do, and both
claims were wrong in the direction that matters — they overstated how much
depended on the code they described.

The shared verdict's doc said deleting its rate-limit term would silently
stop retrying throttling on every non-streaming path. It would not:
`classifyRetryError` runs the same predicate with the same extra codes, so
every throttle still classifies retryable without the term. Measured, the
term's only effect is on DashScope allocated-quota exhaustion surfacing as
HTTP 429, where it keeps the verdict retryable so the persistent loop's own
fail-fast bound sets the attempt count (3) instead of the first attempt
being the last. `defaultShouldRetry`'s comment repeated the same warrant.

The message-scraped provider-code test said its fixture was the only pin on
that reader and that removing it would leave the suite green. Seven other
cases in the file read the same half and fail with it.

Comment-only: no behaviour changes, and the mutations behind each claim
were reverted and the suites re-run green.

* fix(core): give two unowned trailing-failure shapes an arm that recovers

Round 15 filed two fails-closed regressions, both shapes where the
closed-finish veto and the acceptance gate together left a trailing failure
with no owner, so a turn whose prose the caller had already watched stream
died on a transport artefact and reached neither durable layer.

A continuation attempt that closes with no parts of its own — a bare finish
chunk, which is what a model returns when asked to resume an answer it
considers complete — now stays continuable. The veto exists to stop a
fabricated tail, and fabricating one needs something to fabricate onto, so
scoping it to an attempt that closed with output of its own keeps that
protection and stops it stranding the prefix. The not-taken log reads the
same binding, so it no longer attributes a stop to a finish reason that was
not the operative cause.

On a tool-result send the gate's progress term is now turn-scoped, like its
completeness sibling already was. A quiet post-tool-result close has no text
of its own, and declining on the attempt's measure left that shape owned by
nothing: the validation block which throws NO_TOOL_RESULT_PROGRESS and rides
the #7039 retry is guarded on `streamError === null`, so it was skipped while
the error stood, and replay was shut by the delivered prose. Accepting hands
the turn to that block, which recovers it on the same third attempt the
identical shape reaches one frame earlier when no trailing error arrives. The
hoisted binding the validation block reads stays attempt-local — that is what
makes the quiet close an invalid stream in the first place.

The round-10 witness for the observed-close mirror is re-pointed to a 5xx
trailing failure, which the gate's allow-list does not admit, so the veto is
again the operative cause; it still reds when the mirror is removed. The
witness that pinned the attempt-local progress term now pins the mechanism
that matters — the delivered prefix surviving into the record — because the
shape it asserted propagation for is now recovered.

* fix(core): close two ways a permanent or undelivered shape was misread

Round 16 filed two certifies-falsely/fails-closed regressions.

The permanence guard read `type` from only one of the two sources it reads
`code` from. `getRateLimitErrorDetails` collapses a scraped body's `code`
and `type` into a single `providerCode`, so on the message-embedded route a
permanent `type` was dropped whenever a sibling `code` survived — a gateway
relaying `type: 'content_filter'` beside its own `code` was marked retryable,
which is the exact rejection the permanence list exists to fail fast on. The
details struct now carries `providerType` beside `providerCode` rather than
folding the two, because `providerCode` is compared by exact equality for
allocated-quota exhaustion and pinned as that carrier.

The pipeline's error-path flush releases a parked tool-call finish on its own
view of what was delivered, which counts chunks LlmChat's protocol-tag
suppression still withholds — a leading-JSON first chunk, for one. When the
two disagreed, the released functionCall was the first thing the send loop had
seen, so counting it flipped the delivered flags and shut a replay gate that
was still open: a cut the replay arm could have recovered killed the turn on
one attempt. The flush now tags a released tool call, and the send loop — the
only place that knows what reached the caller — drops one that has no
delivered content and no continuation prefix behind it, leaving replay live.
Tagging only the tool-call release keeps a flushed STOP finish on the path
the acceptance gate's `closedFinishReason` depends on.

* fix(core): close a gate regression and an unguarded flush on cancellation

R17-1: The acceptance gate's progress term was turn-scoped (reading
`completedText`, which merges the prefix from earlier attempts), so a
tool-result continuation attempt closing with only a thought part plus
STOP and a trailing error was accepted -- the validation block then threw
NO_TOOL_RESULT_PROGRESS and the invalid-stream arm's fresh restart lost
the prefix.  R15-1's veto scoping (`attemptClosedWithOwnOutput` requires
`streamYieldedContentChunk`) made the continuation arm available for
thought-only shapes, invalidating the premise that "no arm owns that
shape".  Revert the gate to the attempt-local binding: the gate declines,
the continuation arm fires, and the prefix is preserved.

R17-2: The error-path flush in pipeline.ts had no abort guard, so a user
cancellation delivered a parked functionCall into the cancelled turn.
Cancellation persistence then wrote a model[functionCall] turn into
history and the JSONL record for a call the turn driver never dispatched.
Add `request.config?.abortSignal?.aborted !== true`, spelled exactly as
the three sibling synthesis branches in the same catch spell it.

---------

Co-authored-by: qwen-code-dev-bot <[email protected]>
Co-authored-by: qwen-code-ci-bot <[email protected]>
Co-authored-by: Qwen-Coder <[email protected]>

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

@@ -3959,46 +3959,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([
       {

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,
@@ -1627,10 +1630,28 @@ export class AnthropicContentGenerator implements ContentGenerator {
     if (upstreamStreamFailed) {
       const upstreamErrorClassification =
         classifyRetryError(upstreamStreamError);
-      // Match LlmChat's replay boundary: only known mid-SSE socket cuts
-      // may release an already closed batch before the error is propagated.
+      // 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.
+      // The status-less arm reaches this provider only through an id inside
+      // the error body: the SDK builds a mid-stream failure as an
+      // `APIConnectionError` without headers, so the `request-id` response
+      // header never reaches `request_id` the way the OpenAI SDK stamps its
+      // `x-request-id`. Same policy as the OpenAI path, narrower set of
+      // producers — a gateway relaying its own id in the frame, rather than
+      // the SDK handing one over from the response.
+      // 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/contentGenerator.ts

@@ -56,6 +56,16 @@ export type PromptCacheSharingParameters = GenerateContentParameters & {
    * final message is deliberately excluded from cache breakpoints.
    */
   promptCacheSharing?: boolean;
+  /**
+   * Local control signal, never sent on the wire: true when a streaming send
+   * is a transport-continuation attempt resuming an answer whose prefix was
+   * already delivered (LlmChat's transportContinuationPrefix). The OpenAI
+   * pipeline seeds its per-stream delivered-content flag from it, because
+   * with a continuation in flight the turn's replay gate is already shut by
+   * the accumulated prefix — so a parked tool-call finish must be released
+   * rather than withheld for a replay that can no longer happen.
+   */
+  continuationInFlight?: boolean;
 };
 
 /**

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

@@ -49,9 +49,15 @@ import {
 import {
   classifyRetryError,
   isFallbackEligible,
+  isRetryableUpstreamError,
+  type RetryErrorClassificationContext,
 } from '../utils/retryErrorClassification.js';
 import type { Config } from '../config/config.js';
-import type { ContentGenerator, InputModalities } from './contentGenerator.js';
+import type {
+  ContentGenerator,
+  InputModalities,
+  PromptCacheSharingParameters,
+} from './contentGenerator.js';
 import {
   clampOutputTokensToWindow,
   defaultOutputCeiling,
@@ -130,7 +136,11 @@ import {
   getCustomSystemPrompt,
   getManualPlanExitSystemReminder,
 } from './prompts.js';
-import { isRetryableStreamTransportError } from './stream-transport-retry.js';
+import {
+  isFlushedToolCallPark,
+  isRetryableStatuslessUpstreamError,
+  isRetryableStreamTransportError,
+} from './stream-transport-retry.js';
 import {
   collectToolCallIdsFromHistory,
   getFunctionCallFingerprint,
@@ -151,6 +161,23 @@ const debugLogger = createDebugLogger('QWEN_CODE_CHAT');
 // must stay in sync.
 const GEMINI_EMPTY_CONTENT_PLACEHOLDER = '(empty content)';
 
+/**
+ * Finish reasons that positively mean the model's answer is closed: a
+ * completed or definitively blocked response leaves nothing a continuation
+ * could resume. Deliberately a deny-list, never an allow-list — values that
+ * carry no completeness information (converter fall-throughs such as
+ * FINISH_REASON_UNSPECIFIED, or enum members a future @google/genai version
+ * adds) must fail open to continuable.
+ */
+const CLOSED_FINISH_REASONS: ReadonlySet<string> = new Set([
+  FinishReason.STOP,
+  FinishReason.SAFETY,
+  FinishReason.RECITATION,
+  FinishReason.BLOCKLIST,
+  FinishReason.PROHIBITED_CONTENT,
+  FinishReason.SPII,
+]);
+
 function hasCandidateOutput(response: GenerateContentResponse): boolean {
   return Boolean(
     response.candidates?.some(
@@ -655,14 +682,16 @@ const STREAM_RETRY_CONFIG = {
   maxRetries: 2,
   initialDelayMs: 1000,
   /**
-   * Budget for *continuation* recovery after a socket-level cut that already
-   * delivered output (issue #7832). This is a different mechanism from the
-   * `maxRetries` replay above and therefore has its own budget: a replay
-   * re-sends the request from scratch and is only legal before any chunk
-   * reached callers, while a continuation keeps the delivered output and asks
-   * the model to resume from it. A single long generation can be cut more than
-   * once by the same gateway idle timeout, so this is sized like
-   * {@link MAX_OUTPUT_RECOVERY_ATTEMPTS} rather than like the replay budget.
+   * Budget for *continuation* recovery after a mid-stream cut that already
+   * delivered output (issue #7832) — a socket-level failure, or a status-less
+   * upstream failure the provider traced with its own request id. This is a
+   * different mechanism from the `maxRetries` replay above and therefore has
+   * its own budget: a replay re-sends the request from scratch and is only
+   * legal before any chunk reached callers, while a continuation keeps the
+   * delivered output and asks the model to resume from it. A single long
+   * generation can be cut more than once by the same gateway idle timeout, so
+   * this is sized like {@link MAX_OUTPUT_RECOVERY_ATTEMPTS} rather than like
+   * the replay budget.
    */
   maxContinuationRetries: 3,
 };
@@ -2226,6 +2255,16 @@ export class LlmChat {
     | Parameters<ChatRecordingService['recordAssistantTurn']>[0]
     | null = null;
 
+  /**
+   * The first closed finish reason the in-flight `processStreamResponse`
+   * observed, if any. On a tool-result continuation the finish reason is
+   * deferred off the yielded chunks and re-emitted only on success, so on
+   * a failed attempt the send loop's `lastFinishReason` never sees the
+   * close; this side channel is what the continuation veto consults
+   * instead. Reset per attempt alongside `lastFinishReason`.
+   */
+  private lastObservedClosedFinishReason: string | undefined;
+
   private readonly imagePayloadStore = new InMemoryImagePayloadStore();
 
   /**
@@ -3584,7 +3623,33 @@ export class LlmChat {
             // The processor now owns cancellation persistence for this prefix.
             pendingTransportPrefix = [];
             lastFinishReason = undefined;
+            self.lastObservedClosedFinishReason = undefined;
             for await (const chunk of stream) {
+              // A parked tool-call finish the pipeline released on its error
+              // path, reaching an attempt this loop counts nothing delivered
+              // for and with no continuation in flight. The release was decided
+              // from the pipeline's own view of what it yielded, which includes
+              // chunks the protocol-tag suppression in processStreamResponse
+              // withheld — a leading-JSON first chunk, for one. Counting it
+              // here would flip the delivered flags and shut the replay gate
+              // that is in fact still open, killing on one attempt a turn the
+              // replay arm could recover; forwarding it would dispatch a tool
+              // call over output the caller never saw. Drop it and let replay
+              // re-send. The two content terms mirror the replay gate's, so
+              // this fires only where replay is still the live option on
+              // content grounds; the error-path partial turn this attempt
+              // persisted is popped by that arm. Where replay does not end up
+              // firing — its budget spent, or a failure class it does not own
+              // — the turn fails as it would have anyway, and all the drop
+              // costs is that a tool call belonging to an attempt the caller
+              // never saw is not dispatched for it.
+              if (
+                isFlushedToolCallPark(chunk) &&
+                !streamYieldedContentChunk &&
+                transportContinuationText.trim().length === 0
+              ) {
+                continue;
+              }
               if (hasCandidateOutput(chunk)) {
                 streamYieldedChunk = true;
                 streamYieldedAnyChunk = true;
@@ -3733,6 +3798,17 @@ export class LlmChat {
               });
             }
 
+            // Computed above the recovery gates because a status-less upstream
+            // failure that is really an oversized-payload rejection has to
+            // reach the one-shot compaction below instead of being re-sent:
+            // re-sending cannot shrink a request, and the continuation arm
+            // would re-send it strictly larger. A reverse proxy in front of the
+            // endpoint can reject the serialized request with a bare HTTP 413
+            // (no token wording) even below the token-based compaction
+            // threshold; it recovers through the same one-shot path (#10380).
+            const contextOverflow = getContextLengthExceededInfo(error);
+            const requestPayloadOverflow = getRequestPayloadTooLargeInfo(error);
+
             // HTTP establishment failures already exhausted retryWithBackoff;
             // only server errors raised while reading the stream join replay.
             const isServerStreamError =
@@ -3744,9 +3820,25 @@ export class LlmChat {
               classification.statusCode !== undefined &&
               classification.statusCode >= 500 &&
               classification.statusCode < 600;
-            const isReplayableStreamError =
+            // The classes a cut may be resumed from: a curated socket-level
+            // failure, and a status-less upstream failure the provider traced
+            // with its own request id. The latter is what a gateway error
+            // frame pushed into an already-200 stream produces, and it can
+            // only be decided here: retryWithBackoff resolved when the stream
+            // was established, before a single frame was parsed. The overflow
+            // exclusion narrows only that class, so a socket cut keeps the
+            // verdict it had before this branch existed.
+            const isContinuableStreamCut =
               isRetryableStreamTransportError(classification) ||
-              isServerStreamError;
+              (isRetryableStatuslessUpstreamError(classification) &&
+                !contextOverflow.isExceeded &&
+                !requestPayloadOverflow.isTooLarge);
+            // Replay admits one class more than continuation does: a server
+            // error raised while reading an established stream (#11634) is
+            // worth re-sending from scratch, but is not a cut this branch may
+            // ask the model to resume from.
+            const isReplayableStreamError =
+              isContinuableStreamCut || isServerStreamError;
 
             // Replay transient server errors and curated socket-level failures
             // before any content (non-thought output) has reached callers.
@@ -3789,9 +3881,11 @@ export class LlmChat {
                   yieldedNonContentChunks: streamYieldedChunk,
                   errorKind: classification.kind,
                   transportCode: classification.transportCode,
+                  classificationReason: classification.reason,
+                  providerCode: classification.providerCode,
+                  requestId: classification.requestId,
                   ...(isServerStreamError && {
                     statusCode: classification.statusCode,
-                    providerCode: classification.providerCode,
                   }),
                 },
               );
@@ -3827,8 +3921,33 @@ export class LlmChat {
             // produces a sequence providers reject (the same constraint the
             // MAX_TOKENS recovery loop enforces via its `hasFunctionCall`
             // check), and the scheduler's repair path already covers it.
+            // A closed finish reason on an attempt that produced output of its
+            // own means that answer already completed (or was definitively
+            // blocked) — the failure landed while the SDK was absorbing
+            // trailing metadata, so there is nothing to resume and a
+            // continuation would only fabricate a tail into durable history.
+            // MAX_TOKENS stays continuable: it marks a *truncated* answer, the
+            // exact shape this arm exists for.
+            // The yielded finish reason governs when one reached the
+            // caller; when the tool-result deferral stripped it from the
+            // yielded chunks, the close survives only in what
+            // processStreamResponse observed.
+            const attemptFinishReason =
+              lastFinishReason ?? self.lastObservedClosedFinishReason;
+            // Scoped to an attempt that closed *with output of its own*. The
+            // fabricated tail this veto exists to prevent needs something to
+            // fabricate onto: an attempt that contributed no visible part —
+            // a bare finish chunk, which is what a model returns when asked to
+            // resume an answer it considers complete — leaves the turn with
+            // nothing persisted and every other arm shut, so refusing the
+            // continuation there strands prose an earlier attempt delivered.
+            const attemptClosedWithOwnOutput =
+              attemptFinishReason !== undefined &&
+              CLOSED_FINISH_REASONS.has(attemptFinishReason) &&
+              streamYieldedContentChunk;
             const canContinueAfterTransportCut =
-              isRetryableStreamTransportError(classification) &&
+              isContinuableStreamCut &&
+              !attemptClosedWithOwnOutput &&
               !streamYieldedFunctionCall &&
               transportContinuationText.trim().length > 0 &&
               transportContinuationCount <
@@ -3853,6 +3972,9 @@ export class LlmChat {
                 retryDelayMs: delayMs,
                 errorKind: classification.kind,
                 transportCode: classification.transportCode,
+                classificationReason: classification.reason,
+                providerCode: classification.providerCode,
+                requestId: classification.requestId,
                 deliveredChars: transportContinuationText.length,
               });
               // `isContinuation` keeps the UI's text buffer, so the next
@@ -3867,19 +3989,21 @@ export class LlmChat {
             }
             if (isReplayableStreamError) {
               // Reached only when neither branch above fired: content was
-              // already delivered so replaying would duplicate it, or the
+              // already delivered so replaying would duplicate it, the
               // replay budget is exhausted, or continuation is unavailable
-              // (function-call cut, no text to anchor on, or its own budget
-              // exhausted).
+              // (function-call cut, no text to anchor on, its own budget
+              // exhausted, or a closed finish reason — the attempt's answer
+              // already completed, so there was nothing to resume).
               debugLogger.warn(
                 isServerStreamError
                   ? 'Server stream retry not taken'
                   : 'Transport stream retry not taken',
                 {
                   retryPath: 'stream',
-                  retryDecision:
-                    streamYieldedContentChunk ||
-                    transportContinuationText.trim().length > 0
+                  retryDecision: attemptClosedWithOwnOutput
+                    ? 'skipped_terminal_finish_reason'
+                    : streamYieldedContentChunk ||
+                        transportContinuationText.trim().length > 0
                       ? 'skipped_after_content'
                       : 'exhausted',
                   attempts: streamReplayRetryCount,
@@ -3889,20 +4013,18 @@ export class LlmChat {
                     STREAM_RETRY_CONFIG.maxContinuationRetries,
                   errorKind: classification.kind,
                   transportCode: classification.transportCode,
+                  classificationReason: classification.reason,
+                  providerCode: classification.providerCode,
+                  requestId: classification.requestId,
                   ...(isServerStreamError && {
                     statusCode: classification.statusCode,
-                    providerCode: classification.providerCode,
                   }),
                 },
               );
             }
 
-            const contextOverflow = getContextLengthExceededInfo(error);
-            // A reverse proxy in front of the endpoint can reject the
-            // serialized request with a bare HTTP 413 (no token wording)
-            // even below the token-based compaction threshold; recover it
-            // through this same one-shot reactive path (#10380).
-            const requestPayloadOverflow = getRequestPayloadTooLargeInfo(error);
+            // Both detectors were computed above the recovery gates, so that a
+            // status-less upstream failure could be kept out of them.
             if (
               contextOverflow.isExceeded ||
               requestPayloadOverflow.isTooLarge
@@ -4919,15 +5041,20 @@ export class LlmChat {
   ): Promise<AsyncGenerator<GenerateContentResponse>> {
     const generator =
       overrides?.contentGenerator ?? this.config.getContentGenerator();
-    const apiCall = () =>
-      generator.generateContentStream(
-        {
-          model,
-          contents: requestContents,
-          config: { ...this.generationConfig, ...params.config },
-        },
-        prompt_id,
-      );
+    const apiCall = () => {
+      // A continuation attempt's replay gate is already shut by the
+      // accumulated prefix, so the pipeline must release a parked tool-call
+      // finish rather than withhold it for a replay that cannot happen.
+      const request: PromptCacheSharingParameters = {
+        model,
+        contents: requestContents,
+        config: { ...this.generationConfig, ...params.config },
+        ...(transportContinuationPrefix !== undefined && {
+          continuationInFlight: true,
+        }),
+      };
+      return generator.generateContentStream(request, prompt_id);
+    };
     const cgConfig = this.config.getContentGeneratorConfig();
     const authType = overrides?.retryAuthType ?? cgConfig?.authType;
     const extraRetryErrorCodes =
@@ -4959,21 +5086,12 @@ export class LlmChat {
         if (status === 429) return true;
         if (status && status >= 500 && status < 600) return true;
 
-        // Honor provider-specific rate-limit codes (e.g. DashScope) so a custom
-        // predicate does not silently drop them — the default path checks these
-        // via defaultShouldRetry, but a custom shouldRetryOnError bypasses it.
-        if (isRateLimitError(error, extraRetryErrorCodes)) return true;
-
-        // Transient network errors (ECONNRESET, ETIMEDOUT, etc.) carry no HTTP
-        // status and would otherwise fall through every predicate above.
-        if (
-          classifyRetryError(error, { extraRetryErrorCodes }).kind ===
-          'transport'
-        ) {
-          return true;
-        }
-
-        return false;
+        // Everything an HTTP status cannot decide — provider rate-limit codes
+        // (e.g. DashScope), transport failures (ECONNRESET, ETIMEDOUT, …), and
+        // upstream error bodies the provider traced with its own request id.
+        // Shared with defaultShouldRetry, which a custom predicate like this
+        // one bypasses, so the two paths cannot drift apart.
+        return isRetryableUpstreamError(error, extraRetryErrorCodes);
       },
       authType,
       extraRetryErrorCodes,
@@ -5012,6 +5130,7 @@ export class LlmChat {
       transportContinuationPrefix,
       acceptQuietToolResultCompletion,
       params.config?.abortSignal,
+      { authType, extraRetryErrorCodes },
     );
   }
 
@@ -5549,6 +5668,12 @@ export class LlmChat {
    *   before either durable write, so the JSONL transcript and in-memory
    *   history carry the same merged turn (issue #8094). Undefined on every
    *   non-continuation send.
+   * @param retryClassificationContext - Auth type and configured extra retry
+   *   codes to classify a trailing stream failure with. Must match what the
+   *   send loop classifies with: the acceptance gate below decides by
+   *   classification, and a throttle only the caller's configured codes
+   *   recognise would otherwise be misread as a status-less upstream failure
+   *   and swallowed instead of reaching the rate-limit retry.
    */
   private async *processStreamResponse(
     model: string,
@@ -5558,6 +5683,7 @@ export class LlmChat {
     transportContinuationPrefix?: Part[],
     acceptQuietToolResultCompletion = false,
     abortSignal?: AbortSignal,
+    retryClassificationContext?: RetryErrorClassificationContext,
   ): AsyncGenerator<GenerateContentResponse> {
     // Collect ALL parts from the model response (including thoughts for recording)
     const allModelParts: Part[] = [];
@@ -5577,6 +5703,10 @@ export class LlmChat {
 
     let hasToolCall = false;
     let hasFinishReason = false;
+    // The first closed finish reason seen, if any — tracked so a stream
+    // that fails *after* the model closed its answer can be accepted as
+    // complete below rather than retried.
+    let closedFinishReason: string | undefined;
     const protocolTagDetector = new LeadingProtocolTagLeakDetector();
     let pendingProtocolParts: Part[] = [];
     const takePendingProtocolParts = (): Part[] => {
@@ -5634,6 +5764,16 @@ export class LlmChat {
         hasFinishReason ||=
           chunk?.candidates?.some((candidate) => candidate.finishReason) ??
           false;
+        closedFinishReason ??= chunk?.candidates?.find(
+          (candidate) =>
+            candidate.finishReason !== undefined &&
+            CLOSED_FINISH_REASONS.has(candidate.finishReason),
+        )?.finishReason;
+        // Mirror onto the instance: the tool-result deferral below strips
+        // the reason from the yielded chunk, and a failed attempt never
+        // re-emits it — without this the send loop's continuation veto is
+        // blind to a close on exactly that path.
+        this.lastObservedClosedFinishReason ??= closedFinishReason;
 
         if (isValidResponse(chunk)) {
           const candidate = chunk.candidates?.[0];
@@ -5940,6 +6080,108 @@ export class LlmChat {
       .join('')
       .trim();
 
+    // Completeness is a property of the turn, not of the attempt. On a
+    // continuation the visible text already delivered lives in
+    // `transportContinuationPrefix`, and the merge that folds it back in runs
+    // below — and only once `streamError` is null. Measuring this gate with the
+    // attempt's own `contentText` therefore refused exactly the attempt that
+    // closes a continuation without adding prose (a thought part and STOP), and
+    // no other arm owns that failure: replay needs an empty delivered prefix,
+    // continuation is vetoed by this very close, and the rate-limit, overflow
+    // and invalid-stream arms do not match a status-less frame. The prose the
+    // caller watched stream then reached neither history nor the JSONL record.
+    //
+    // The attempt must still have contributed something of its own. One that
+    // delivered nothing at all is left to the continuation arm, which resumes
+    // from the prefix; accepting it here would route it to the empty-response
+    // validation instead, and the fresh restart that follows re-sends the
+    // original prompt and loses the prose the caller already watched stream.
+    // With no prefix the conjuncts reduce to `contentText`, so every
+    // non-continuation shape is decided exactly as before.
+    //
+    // Guarded on `streamError` because only the gate below reads this, and the
+    // merge it mirrors is a dedup pass over the delivered prefix: without the
+    // guard every successful continuation turn would pay it twice, once here
+    // and once in the merge block. The prefix travels as parts, so it is
+    // reduced to text the same way the cancellation merge above reduces it.
+    const completedText =
+      streamError !== null && transportContinuationPrefix
+        ? mergeDeliveredPrefix(
+            getPlainTextFromParts(transportContinuationPrefix),
+            contentText,
+          )
+        : contentText;
+    // `hasAnyContent` is shared with the stream-validation block below, hoisted
+    // rather than duplicated so the two reads cannot drift.
+    // `lacksVisibleToolResultProgress` is attempt-local: it measures what this
+    // attempt delivered, not what the turn accumulated. The acceptance gate
+    // reads it directly — a continuation attempt that closed with only a
+    // thought part after a tool result has no visible progress of its own, so
+    // the gate declines and the continuation arm owns the shape. The validation
+    // block below reads the same binding for the same reason.
+    const hasAnyContent = contentText || thoughtText;
+    const lacksVisibleToolResultProgress =
+      isToolResultContinuation &&
+      (!contentText || contentText === GEMINI_EMPTY_CONTENT_PLACEHOLDER);
+
+    // A failure that lands after the model already closed its answer —
+    // typically a gateway error frame pushed into an already-200 stream
+    // while the SDK was absorbing trailing usage metadata — must not fail
+    // the turn: the answer is complete, and the trailing error concerns
+    // only a tail the turn no longer needs. Accept the delivered content so
+    // it persists through the success path below; propagating instead
+    // strands a complete answer out of history and the JSONL record, and
+    // the retry arms have nothing to resume (continuing would fabricate a
+    // tail past the finish). Tool-call turns keep the error-path partial
+    // persistence below, which the scheduler's repair flow depends on.
+    if (
+      streamError !== null &&
+      !hasToolCall &&
+      closedFinishReason !== undefined &&
+      completedText &&
+      hasAnyContent &&
+      !lacksVisibleToolResultProgress
+    ) {
+      // Two failure classes can be swallowed here: a curated socket-level cut,
+      // and a status-less upstream frame the provider traced with its own
+      // request id. Both are transport or gateway artefacts that say nothing
+      // about the answer, which is why a closed answer can survive them. The
+      // server-error class the replay gate also admits (#11634) is deliberately
+      // not one of them: that status is the server's own verdict on the
+      // response, and the arm owning it answers with a retry rather than by
+      // accepting the turn. Anything else reached this point precisely because
+      // no recovery arm owns it, so nulling it would certify an outcome the
+      // turn did not have: a user cancel would be converted into a completion,
+      // a throttling StreamContentError would never reach the rate-limit
+      // retry, and the pipeline's own InvalidStreamError would bypass the
+      // invalid-stream retry budget.
+      //
+      // Classified with the send loop's own context. A throttle that only the
+      // configured `retryErrorCodes` recognise carries a request id and no
+      // status, so without that context it reads as a status-less upstream
+      // frame — the one class this gate accepts.
+      const trailingErrorClassification = classifyRetryError(
+        streamError,
+        retryClassificationContext,
+      );
+      if (
+        isRetryableStreamTransportError(trailingErrorClassification) ||
+        isRetryableStatuslessUpstreamError(trailingErrorClassification)
+      ) {
+        debugLogger.warn(
+          'Accepting completed answer despite trailing stream failure.',
+          {
+            finishReason: closedFinishReason,
+            error:
+              streamError instanceof Error
+                ? streamError.message
+                : String(streamError),
+          },
+        );
+        streamError = null;
+      }
+    }
+
     // Deferred until after the throw sites below so a protocol-tag leak
     // or stream-validation failure cannot dispatch a recovered call that
     // the retry path would then execute a second time.
@@ -6080,11 +6322,9 @@ export class LlmChat {
     // tool call, so they retry (#7039) — and once that retry budget is
     // exhausted the quiet completion is accepted rather than failing the
     // run (#9026): some model families legitimately end turns silently
-    // after a tool result.
-    const hasAnyContent = contentText || thoughtText;
-    const lacksVisibleToolResultProgress =
-      isToolResultContinuation &&
-      (!contentText || contentText === GEMINI_EMPTY_CONTENT_PLACEHOLDER);
+    // after a tool result. Both bindings this reads are computed above the
+    // trailing-failure acceptance gate, which shares them (see the comment
+    // there for why sharing them is the point).
     let acceptedQuietToolResultCompletion = false;
     if (
       streamError === null &&

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

@@ -26,7 +26,11 @@ import { OpenAIContentConverter } from './converter.js';
 import { openaiRequestCaptureContext } from './requestCaptureContext.js';
 import { StreamingToolCallParser } from './streamingToolCallParser.js';
 import type { Config } from '../../config/config.js';
-import { AuthType, type ContentGeneratorConfig } from '../contentGenerator.js';
+import {
+  AuthType,
+  type ContentGeneratorConfig,
+  type PromptCacheSharingParameters,
+} from '../contentGenerator.js';
 import type { OpenAICompatibleProvider } from './provider/index.js';
 import { DefaultOpenAICompatibleProvider } from './provider/default.js';
 import { DashScopeOpenAICompatibleProvider } from './provider/dashscope.js';
@@ -3817,7 +3821,7 @@ describe('ContentGenerationPipeline', () => {
       ]);
     });
 
-    it('does not log protocol-tag sanitization before a held finish is yielded', async () => {
+    it('logs protocol-tag sanitization when a held finish is flushed on the error path', async () => {
       const request: GenerateContentParameters = {
         model: 'test-model',
         contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
@@ -3827,16 +3831,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,
         },
@@ -3847,7 +3854,7 @@ describe('ContentGenerationPipeline', () => {
         (_chunk, context) => {
           context.protocolTagSanitized = {
             tagName: 'think',
-            toolCallCount: 1,
+            toolCallCount: 0,
           };
           return finishResponse;
         },
@@ -3861,12 +3868,18 @@ describe('ContentGenerationPipeline', () => {
         'test-prompt-id',
       );
 
+      const results: GenerateContentResponse[] = [];
       await expect(async () => {
-        for await (const _ of resultGenerator) {
+        for await (const result of resultGenerator) {
           // Consume until the stream error after the held finish.
+          results.push(result);
         }
       }).rejects.toThrow(streamError);
-      expect(logProtocolTagSanitized).not.toHaveBeenCalled();
+      // The error-path flush delivers the held finish ahead of the
+      // rejection, so its sanitization is telemetry for a response the
+      // caller really received — suppressing it would lose the event.
+      expect(results).toEqual([finishResponse]);
+      expect(logProtocolTagSanitized).toHaveBeenCalledTimes(1);
     });
 
     it('logs only the accepted finish after duplicate and empty trailing chunks', async () => {
@@ -5454,6 +5467,774 @@ describe('ContentGenerationPipeline', () => {
       }
       expect(totalFunctionCalls).toBe(1);
     });
+
+    it('flushes a parked finish response when the stream fails before the trailing tail', async () => {
+      // A gateway error frame can land where the trailing usage chunk would
+      // have been: the finish chunk is already parked for the usage merge,
+      // and the iterator throws before anything releases it. The caller must
+      // still receive the finish response ahead of the rejection, or
+      // downstream completeness gates cannot tell the answer finished.
+      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 upstreamError = Object.assign(new Error("'id'"), {
+        code: 'KeyError',
+        requestID: 'req-1',
+      });
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield mockChunk1;
+          yield mockChunk2;
+          throw upstreamError;
+        },
+      };
+
+      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 propagated error.
+      const finishResult = await iterator.next();
+      if (finishResult.done) throw new Error('Expected a finish response.');
+      expect(finishResult.value.candidates?.[0]?.finishReason).toBe(
+        FinishReason.STOP,
+      );
+
+      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);
+    });
+
+    it('releases a parked functionCall finish once content was delivered when the stream fails', async () => {
+      // Sibling of the withhold case above: the withhold protects LlmChat's
+      // transport replay gate, which is open only while nothing user-visible
+      // was delivered. Once prose has reached the caller that gate is already
+      // shut, so withholding buys no recovery — it only strands the model's
+      // decided tool call. The parked finish must be released so the
+      // delivered functionCall flips LlmChat's delivered flags and the
+      // error-path persistence plus the scheduler's repair flow take over.
+      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: 'prose-chunk',
+            choices: [
+              {
+                delta: { content: 'Let me read that file. ' },
+                finish_reason: null,
+              },
+            ],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'tool_calls' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          throw streamError;
+        },
+      };
+      const proseResponse = new GenerateContentResponse();
+      proseResponse.candidates = [
+        {
+          content: {
+            parts: [{ text: 'Let me read that file. ' }],
+            role: 'model',
+          },
+          index: 0,
+        },
+      ];
+      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)
+        .mockReturnValueOnce(proseResponse)
+        .mockReturnValueOnce(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);
+      // Both the prose and the released finish reach the caller ahead of the
+      // rejection.
+      expect(results).toEqual([proseResponse, finishResponse]);
+      expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
+    });
+
+    it('withholds a parked functionCall finish when only thought content was delivered', async () => {
+      // The delivered-content flag mirrors LlmChat's notion, which excludes
+      // thought parts: a thought-only prefix persists nothing on the error
+      // path and leaves the transport replay gate open, so the parked
+      // tool-call finish stays withheld exactly as when nothing was
+      // delivered at all.
+      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: 'thought-chunk',
+            choices: [
+              {
+                delta: { content: 'Let me plan this out.' },
+                finish_reason: null,
+              },
+            ],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'tool_calls' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          throw streamError;
+        },
+      };
+      const thoughtResponse = new GenerateContentResponse();
+      thoughtResponse.candidates = [
+        {
+          content: {
+            parts: [{ text: 'Let me plan this out.', thought: true }],
+            role: 'model',
+          },
+          index: 0,
+        },
+      ];
+      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)
+        .mockReturnValueOnce(thoughtResponse)
+        .mockReturnValueOnce(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 thought reached the caller but does not count as delivered
+      // content, so the tool-call finish stays parked.
+      expect(results).toEqual([thoughtResponse]);
+      expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
+    });
+
+    it('releases a parked functionCall finish on a continuation attempt even when only thought content was delivered', async () => {
+      // Sibling of the thought-only withhold case above with the request
+      // marked as a transport-continuation attempt. The replay gate the
+      // withhold protects is turn-scoped (LlmChat's
+      // transportContinuationText), and with a continuation in flight it is
+      // already shut by the accumulated prefix — which a fresh stream's own
+      // yields cannot show. The delivered-content flag is seeded from the
+      // continuation marker, so the decided tool call is released instead of
+      // staying parked into another prose continuation.
+      const request: PromptCacheSharingParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+        continuationInFlight: true,
+      };
+      const streamError = new Error('stream failed after finish');
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'thought-chunk',
+            choices: [
+              {
+                delta: { content: 'Let me plan this out.' },
+                finish_reason: null,
+              },
+            ],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'tool_calls' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          throw streamError;
+        },
+      };
+      const thoughtResponse = new GenerateContentResponse();
+      thoughtResponse.candidates = [
+        {
+          content: {
+            parts: [{ text: 'Let me plan this out.', thought: true }],
+            role: 'model',
+          },
+          index: 0,
+        },
+      ];
+      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)
+        .mockReturnValueOnce(thoughtResponse)
+        .mockReturnValueOnce(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 thought reaches the caller, and so does the parked tool-call
+      // finish: the continuation marker means the replay gate the withhold
+      // would have protected is already shut.
+      expect(results).toEqual([thoughtResponse, finishResponse]);
+      expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
+    });
+
+    it('does not re-deliver a Stage 2d finish when the consumer throws into the generator', async () => {
+      // The Stage 2d flush yields the parked finish and suspends there. A
+      // consumer that throws into the generator at that point lands in the
+      // error-path flush, which re-tests `finishYielded`: had Stage 2d not set
+      // it, the same response object would be delivered a second time, and
+      // every part it carries would be folded into the persisted turn twice.
+      // Latent rather than live: nothing in production initiates a throw into
+      // this chain — the two `.throw()` sites in it only forward one
+      // (`llm-content-generator.ts`, `loggingContentGenerator.ts`), and a
+      // `for await` consumer abandons through `.return()`, which does not run
+      // the catch. This pins the invariant so the first caller that does throw
+      // cannot double-deliver.
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'stop' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+        },
+      };
+      const finishResponse = new GenerateContentResponse();
+      finishResponse.candidates = [
+        {
+          content: { parts: [{ text: 'a complete answer' }], role: 'model' },
+          finishReason: FinishReason.STOP,
+          index: 0,
+        },
+      ];
+
+      (mockConverter.convertLlmRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockConverter.convertOpenAIChunkToLlm as Mock).mockReturnValueOnce(
+        finishResponse,
+      );
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        'test-prompt-id',
+      );
+
+      const first = await resultGenerator.next();
+      expect(first.value).toBe(finishResponse);
+
+      // The error-path flush must not hand the same response back a second
+      // time: the throw propagates to the consumer instead of resolving with
+      // another value.
+      await expect(
+        resultGenerator.throw!(new Error('consumer threw into the generator')),
+      ).rejects.toThrow('consumer threw into the generator');
+      expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
+    });
+
+    it('does not re-deliver an in-loop merged finish when the consumer throws into the generator', async () => {
+      // Sibling of the Stage 2d case above, for the other yield of a parked
+      // finish: a chunk arriving after the finish one is merged into it and
+      // yielded in-loop. A throw landing on that suspension point reaches the
+      // error-path flush with the same parked response still set, so the flag
+      // has to be raised before suspending here too.
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'stop' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          yield {
+            id: 'usage-chunk',
+            choices: [{ delta: {} }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+        },
+      };
+      const finishResponse = new GenerateContentResponse();
+      finishResponse.candidates = [
+        {
+          content: { parts: [{ text: 'a complete answer' }], role: 'model' },
+          finishReason: FinishReason.STOP,
+          index: 0,
+        },
+      ];
+      const usageResponse = new GenerateContentResponse();
+      usageResponse.usageMetadata = {
+        promptTokenCount: 3,
+        candidatesTokenCount: 5,
+        totalTokenCount: 8,
+      };
+
+      (mockConverter.convertLlmRequestToOpenAI as Mock).mockReturnValue([]);
+      (mockConverter.convertOpenAIChunkToLlm as Mock)
+        .mockReturnValueOnce(finishResponse)
+        .mockReturnValueOnce(usageResponse);
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        'test-prompt-id',
+      );
+
+      const first = await resultGenerator.next();
+      // The merged response keeps the parked finish's candidates and carries
+      // the late usage.
+      expect(first.value?.candidates).toBe(finishResponse.candidates);
+      expect(first.value?.usageMetadata?.totalTokenCount).toBe(8);
+
+      await expect(
+        resultGenerator.throw!(new Error('consumer threw into the generator')),
+      ).rejects.toThrow('consumer threw into the generator');
+      expect(mockErrorHandler.handle).toHaveBeenCalledTimes(1);
+    });
+
+    it('releases a parked tool call when the held-parts flush delivered the content', async () => {
+      // The unclosed-thinking-tag flush yields the parts the converter was
+      // holding. When those carry non-thought content, LlmChat counts the
+      // chunk as delivered and shuts its transport replay gate, so
+      // `contentYielded` has to agree at this yield site too — otherwise the
+      // error-path flush withholds a parked tool call that no recovery arm can
+      // pick up any more, stranding the model's decided call into a prose
+      // continuation. Held parts that are thought-only must keep the flag
+      // false — that is the withhold sibling below.
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'held-parts-chunk',
+            choices: [{ delta: { content: ' ' }, finish_reason: null }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+        },
+      };
+      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).mockImplementation(
+        (_chunk, context) => {
+          // A whitespace-only candidate with no closing tag takes the flush
+          // branch, and the parts it holds are plain content, not reasoning.
+          context.pendingThinkingTagCandidate = { text: ' ' };
+          context.pendingUntrustedResponseParts = [
+            { text: 'Let me read that file. ' },
+          ];
+          return finishResponse;
+        },
+      );
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        'test-prompt-id',
+      );
+
+      const first = await resultGenerator.next();
+      expect(first.value?.candidates?.[0]?.content?.parts).toEqual([
+        { text: 'Let me read that file. ' },
+      ]);
+
+      const second = await resultGenerator.throw!(
+        new Error('consumer threw into the generator'),
+      );
+      expect(second.value).toBe(finishResponse);
+    });
+
+    it('keeps a parked tool call withheld when the held-parts flush delivered only thought', async () => {
+      // The other end of the same knob. Held parts can be thought-marked
+      // reasoning, which LlmChat does not count as delivered, so its transport
+      // replay gate is still open and withholding the parked tool call still
+      // buys the recovery it exists for. Updating the flag unconditionally at
+      // that yield site would strand the call in the other direction.
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+      };
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'held-thought-chunk',
+            choices: [{ delta: { content: ' ' }, finish_reason: null }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+        },
+      };
+      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).mockImplementation(
+        (_chunk, context) => {
+          context.pendingThinkingTagCandidate = { text: ' ' };
+          context.pendingUntrustedResponseParts = [
+            { thought: true, text: 'reasoning' },
+          ];
+          return finishResponse;
+        },
+      );
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        'test-prompt-id',
+      );
+
+      const first = await resultGenerator.next();
+      expect(first.value?.candidates?.[0]?.content?.parts).toEqual([
+        { thought: true, text: 'reasoning' },
+      ]);
+
+      // Nothing user-visible was delivered, so the parked finish stays parked
+      // and the throw propagates instead of resolving with it.
+      await expect(
+        resultGenerator.throw!(new Error('consumer threw into the generator')),
+      ).rejects.toThrow('consumer threw into the generator');
+    });
+
+    it('does not flush a parked tool-call finish into a cancelled turn', async () => {
+      // R17-2. The flush synthesises a delivery on the error path, so it needs
+      // the abort guard every other synthesis branch in this catch carries
+      // (the PROTOCOL_TAG_LEAK throw below spells it
+      // `request.config?.abortSignal?.aborted !== true`). Without it a user
+      // cancellation hands over a parked functionCall the caller was never
+      // shown: LlmChat folds it into the turn's parts before the AbortError
+      // lands, and cancellation persistence then writes a model[functionCall]
+      // turn into history and the JSONL record for a call the turn driver
+      // never dispatched — the transcript asserts a tool call in a turn the
+      // user cancelled. Guarded on the signal rather than the error's
+      // identity, because `isAbortError` is not imported here and two branches
+      // of one catch must not disagree about what "aborted" means.
+      const abortController = new AbortController();
+      const abortError = new Error('Aborted');
+      abortError.name = 'AbortError';
+      const request: GenerateContentParameters = {
+        model: 'test-model',
+        contents: [{ parts: [{ text: 'Hello' }], role: 'user' }],
+        config: { abortSignal: abortController.signal },
+      };
+      const mockStream = {
+        async *[Symbol.asyncIterator]() {
+          yield {
+            id: 'prose-chunk',
+            choices: [
+              {
+                delta: { content: 'Let me read that file. ' },
+                finish_reason: null,
+              },
+            ],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          yield {
+            id: 'finish-chunk',
+            choices: [{ delta: {}, finish_reason: 'tool_calls' }],
+          } as OpenAI.Chat.ChatCompletionChunk;
+          // The usage tail never arrives: the user pressed Esc first.
+          abortController.abort();
+          throw abortError;
+        },
+      };
+      const proseResponse = new GenerateContentResponse();
+      proseResponse.candidates = [
+        {
+          content: {
+            parts: [{ text: 'Let me read that file. ' }],
+            role: 'model',
+          },
+          index: 0,
+        },
+      ];
+      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)
+        .mockReturnValueOnce(proseResponse)
+        .mockReturnValueOnce(finishResponse);
+      (mockClient.chat.completions.create as Mock).mockResolvedValue(
+        mockStream,
+      );
+
+      const resultGenerator = await pipeline.executeStream(
+        request,
+        'test-prompt-id',
+      );
+      const results: GenerateContentResponse[] = [];
+      let caught: unknown;
+      try {
+        for await (const result of resultGenerator) results.push(result);
+      } catch (error) {
+        caught = error;
+      }
+
+      // The prose the caller was shown survives; the parked call does not.
+      expect(results).toEqual([proseResponse]);
+      expect(
+        results.some((response) =>
+          response.candidates?.some((candidate) =>
+            candidate.content?.parts?.some((part) => part.functionCall),
+          ),
+        ),
+      ).toBe(false);
+      expect(caught).toBe(abortError);
+    });
   });
 
   describe('buildResponseFormat endpoint gate', () => {

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

@@ -46,6 +46,7 @@ import {
 } from '../stream-guards.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 import { getToolCallPreparations } from '../tool-call-preparation.js';
+import { markFlushedToolCallPark } from '../stream-transport-retry.js';
 import { InvalidStreamError } from '../invalid-stream-error.js';
 import { logProtocolTagSanitized } from '../../telemetry/loggers.js';
 import { ProtocolTagSanitizedEvent } from '../../telemetry/types.js';
@@ -270,6 +271,23 @@ function isSSECompatibleContentType(contentType: string | null): boolean {
   );
 }
 
+/**
+ * True when the response carries user-visible model output: any candidate
+ * part without the `thought` flag (text, functionCall, inlineData, …).
+ * Mirrors LlmChat's delivered-content notion (its
+ * `hasNonThoughtCandidateParts`), which excludes thought parts — a
+ * thought-only prefix must still count as nothing delivered.
+ */
+function hasNonThoughtCandidateParts(
+  response: GenerateContentResponse,
+): boolean {
+  return Boolean(
+    response.candidates?.some((candidate) =>
+      candidate.content?.parts?.some((part) => !part.thought),
+    ),
+  );
+}
+
 /**
  * Thrown when the HTTP 200 response to a streaming request has a content-type
  * incompatible with SSE (e.g. `text/html` from a gateway block page). Carries
@@ -546,7 +564,7 @@ export class ContentGenerationPipeline {
   private async *processStreamWithLogging(
     stream: AsyncIterable<OpenAI.Chat.ChatCompletionChunk>,
     context: RequestContext,
-    request: GenerateContentParameters,
+    request: PromptCacheSharingParameters,
     userPromptId: string,
     telemetryAttempt: GenAiAttemptHandle | undefined,
   ): AsyncGenerator<GenerateContentResponse> {
@@ -559,6 +577,17 @@ export class ContentGenerationPipeline {
     // function-call parts from the finish chunk).
     let pendingFinishResponse: GenerateContentResponse | null = null;
     let finishYielded = false;
+    // Whether any user-visible content (a non-thought part) has been yielded
+    // on this stream. The error-path flush below consults it before
+    // withholding a parked tool-call finish: it must mirror LlmChat's
+    // delivered-content notion, which excludes thought parts. Seeded from the
+    // caller's continuation marker because the replay gate the withhold
+    // protects is turn-scoped (LlmChat's transportContinuationText), which a
+    // fresh attempt's own yields cannot see: with a continuation in flight
+    // that gate is already shut by the accumulated prefix, and withholding
+    // would only strand the model's decided tool call into another prose
+    // continuation.
+    let contentYielded = request.continuationInFlight === true;
     let pendingFinishProtocolTagSanitized:
       | NonNullable<RequestContext['protocolTagSanitized']>
       | undefined;
@@ -681,11 +710,16 @@ export class ContentGenerationPipeline {
               pendingFinishResponse,
               pendingFinishProtocolTagSanitized,
             );
-            yield pendingFinishResponse;
+            // Set before suspending rather than after: a consumer that throws
+            // into this generator at the yield below never runs the statement
+            // that follows it, and the error-path flush re-tests this flag
+            // before deciding whether the response still needs delivering.
             finishYielded = true;
+            yield pendingFinishResponse;
             // Keep pendingFinishResponse alive so late-arriving usage
             // metadata can still be merged (see finishYielded block above).
           } else {
+            contentYielded ||= hasNonThoughtCandidateParts(response);
             logPendingProtocolTagSanitized(response, sanitization);
             yield response;
           }
@@ -708,6 +742,13 @@ export class ContentGenerationPipeline {
               index: 0,
             },
           ];
+          // Held parts are whatever the converter had accumulated — plain
+          // content, thought-marked reasoning, or both — so this goes through
+          // the same predicate as every other yield site. LlmChat counts a
+          // chunk as delivered on that same rule; if the two flags disagree
+          // here, the error-path flush below withholds a parked tool call
+          // whose replay gate is already shut.
+          contentYielded ||= hasNonThoughtCandidateParts(response);
           yield response;
         }
       } else if (
@@ -728,6 +769,8 @@ export class ContentGenerationPipeline {
           pendingFinishResponse,
           pendingFinishProtocolTagSanitized,
         );
+        // Before the yield, for the reason given at the in-loop one above.
+        finishYielded = true;
         yield pendingFinishResponse;
       }
     } catch (error) {
@@ -735,6 +778,67 @@ 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. The Stage 2d flush above
+      // sets `finishYielded` before it suspends, so a consumer that throws
+      // into this generator while it is parked on that yield cannot make this
+      // flush deliver the same response a second time.
+      //
+      // A parked finish carrying a functionCall stays parked only while
+      // nothing user-visible was delivered: 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. Once content has been yielded that gate
+      // is already shut, so withholding buys no recovery — it would strand
+      // the model's decided tool call: LlmChat would see prose, no
+      // functionCall, and no finish reason, so the continuation arm would
+      // resume over the delivered prose while the call never reaches
+      // error-path persistence or the scheduler's repair flow. Releasing it
+      // here puts the cut on the same footing as the Anthropic
+      // deferred-batch release gate.
+      // TypeScript narrows pendingFinishResponse to null here (its only
+      // assignments sit inside the handleChunkMerging callback), so the
+      // property access needs the same explicit cast as the finishYielded
+      // merge above.
+      const parked = pendingFinishResponse as GenerateContentResponse | null;
+      const parkedHasToolCall = parked?.candidates?.some((candidate) =>
+        candidate.content?.parts?.some((part) => part.functionCall),
+      );
+      if (
+        pendingFinishResponse &&
+        !finishYielded &&
+        // A cancellation is not a stream failure to recover from: synthesising
+        // a delivery here hands the consumer a finish it was never shown, and
+        // cancellation persistence keeps whatever the consumer received.
+        // Spelled exactly as the PROTOCOL_TAG_LEAK branch below spells it, so
+        // one catch does not hold two notions of "aborted".
+        request.config?.abortSignal?.aborted !== true &&
+        (!parkedHasToolCall || contentYielded)
+      ) {
+        logPendingProtocolTagSanitized(
+          pendingFinishResponse,
+          pendingFinishProtocolTagSanitized,
+        );
+        // `contentYielded` is this pipeline's view of what was delivered, and
+        // the consumer can still withhold a chunk it was handed — LlmChat's
+        // protocol-tag suppression drops a leading-JSON chunk whole. Tag a
+        // released tool call so the send loop, which knows what actually
+        // reached the caller, can refuse to let it shut a replay gate that is
+        // in fact still open.
+        if (parkedHasToolCall) {
+          markFlushedToolCallPark(pendingFinishResponse);
+        }
+        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) {

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

@@ -4,6 +4,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+import type { GenerateContentResponse } from '@google/genai';
+
 import type { RetryErrorClassification } from '../utils/retryErrorClassification.js';
 
 // Internal stream retry allow-list. Keep this outside llm-chat.ts because
@@ -27,3 +29,52 @@ export function isRetryableStreamTransportError(
     RETRYABLE_STREAM_TRANSPORT_CODES.has(classification.transportCode)
   );
 }
+
+/**
+ * True when the provider traced a failure with its own request id but no HTTP
+ * 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.
+ * 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,
+): boolean {
+  return classification.reason === 'upstream-error-without-status';
+}
+
+const flushedToolCallParks = new WeakMap<GenerateContentResponse, true>();
+
+/**
+ * Tag a parked tool-call finish that the pipeline's error-path flush released.
+ *
+ * The flush decides whether to release from the pipeline's own view of what it
+ * yielded, and that view counts chunks LlmChat's protocol-tag suppression can
+ * still withhold — a first chunk of leading JSON, for one. When the two
+ * disagree the release hands LlmChat a `functionCall` it delivered nothing for,
+ * and receiving one flips the very delivered-flags that shut replay and
+ * continuation, so a cut the replay arm could still have recovered instead
+ * kills the turn on its first attempt. The tag lets the send loop — the only
+ * place that knows what actually reached the caller — decline to count a
+ * release that has no delivered content behind it.
+ *
+ * A WeakMap rather than a field on the response: these objects cross a package
+ * boundary and their type is not this repo's to extend. The same channel shape
+ * is already used for tool-call preparations, whose keys survive from the
+ * converter to `processStreamResponse` unchanged.
+ */
+export function markFlushedToolCallPark(
+  response: GenerateContentResponse,
+): void {
+  flushedToolCallParks.set(response, true);
+}
+
+export function isFlushedToolCallPark(
+  response: GenerateContentResponse,
+): boolean {
+  return flushedToolCallParks.get(response) === true;
+}

packages/core/src/utils/rateLimit.test.ts

@@ -330,6 +330,10 @@ describe('rate-limit retry diagnostics', () => {
 
     expect(getRateLimitErrorDetails(error)).toEqual({
       providerCode: 'invalid_request_error',
+      // The same value again on its own field: `providerCode` is the collapsed
+      // `code ?? type`, so a body carrying both would otherwise hide the type
+      // from the classifier's permanence guard.
+      providerType: 'invalid_request_error',
       providerMessage,
       transport: 'unknown',
     });

packages/core/src/utils/rateLimit.ts

@@ -31,6 +31,15 @@ export interface RetryInfo {
 export interface RateLimitErrorDetails {
   statusCode?: number;
   providerCode?: string;
+  /**
+   * The provider body's `type`, kept separate from `providerCode` rather than
+   * folded into it. `providerCode` is compared by exact equality elsewhere
+   * (`isAllocatedQuotaExceeded`) and is what the collapsed `code ?? type`
+   * spelling has always produced, so it cannot carry both; a body with a
+   * specific `code` beside a class-naming `type` needs the second value on its
+   * own field to be readable at all.
+   */
+  providerType?: string;
   providerMessage?: string;
   requestId?: string;
   transport: 'http' | 'sse' | 'unknown';
@@ -133,6 +142,7 @@ export function getRateLimitErrorDetails(
     ...(payload?.code !== undefined || payload?.type !== undefined
       ? { providerCode: String(payload.code ?? payload.type) }
       : {}),
+    ...(payload?.type !== undefined ? { providerType: payload.type } : {}),
     ...(payload?.message !== undefined
       ? { providerMessage: payload.message }
       : {}),

packages/core/src/utils/retry.test.ts

@@ -447,6 +447,71 @@ describe('retryWithBackoff', () => {
     expect(mockFn).toHaveBeenCalledTimes(2);
   });
 
+  it('should retry a status-less upstream error carrying a provider request id', async () => {
+    // A gateway error pushed into an already-200 SSE stream reaches us as an
+    // APIError with no HTTP status, so only the classification can open the
+    // retry gate. Without it the turn died on the first attempt.
+    let attempts = 0;
+    const mockFn = vi.fn(async () => {
+      attempts++;
+      if (attempts <= 1) {
+        throw Object.assign(new Error("'id'"), {
+          code: 'KeyError',
+          requestID: 'cd7f37f3-d38a-9dec-804f-f70dda5650eb',
+        });
+      }
+      return 'ok';
+    });
+
+    const promise = retryWithBackoff(mockFn, {
+      maxAttempts: 3,
+      initialDelayMs: 10,
+    });
+    await vi.advanceTimersByTimeAsync(1000);
+    const result = await promise;
+
+    expect(result).toBe('ok');
+    expect(mockFn).toHaveBeenCalledTimes(2);
+  });
+
+  it('should not retry a status-less error that carries only a provider code', async () => {
+    // Permanent local failures (missing credentials, invalid MCP config) have a
+    // string `code` but no request id; the request id is what distinguishes an
+    // upstream failure from one of those.
+    const mockFn = vi.fn(async () => {
+      throw Object.assign(new Error('No API key configured'), {
+        code: 'MISSING_API_KEY',
+      });
+    });
+
+    const promise = retryWithBackoff(mockFn, {
+      maxAttempts: 3,
+      initialDelayMs: 10,
+    });
+    await expect(promise).rejects.toThrow('No API key configured');
+    expect(mockFn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should not retry a permanent provider code even when the request is traced', async () => {
+    // Moderation and credential rejections arrive inside an already-200 stream,
+    // so no HTTP status is left to fail fast on. Re-sending the identical
+    // request cannot succeed, and walking the production ladder for it costs
+    // about eighty seconds.
+    const mockFn = vi.fn(async () => {
+      throw Object.assign(new Error('Content filtered'), {
+        code: 'data_inspection_failed',
+        requestID: 'req-1',
+      });
+    });
+
+    const promise = retryWithBackoff(mockFn, {
+      maxAttempts: 3,
+      initialDelayMs: 10,
+    });
+    await expect(promise).rejects.toThrow('Content filtered');
+    expect(mockFn).toHaveBeenCalledTimes(1);
+  });
+
   it('should respect maxDelayMs', async () => {
     const mockFn = createFailingFunction(3);
     const setTimeoutSpy = vi.spyOn(global, 'setTimeout');

packages/core/src/utils/retry.ts

@@ -13,9 +13,11 @@ import {
 } from './quotaErrorDetection.js';
 import { createDebugLogger } from './debugLogger.js';
 import { getErrorStatus } from './errors.js';
-import { isRateLimitError } from './rateLimit.js';
 import { getRetryAfterDelayMs, getRetryDelayMs } from './retryPolicy.js';
-import { classifyRetryError } from './retryErrorClassification.js';
+import {
+  classifyRetryError,
+  isRetryableUpstreamError,
+} from './retryErrorClassification.js';
 import { retryContext } from './retryContext.js';
 import { ResponsesHttpError } from './responses-http-error.js';
 
@@ -102,7 +104,9 @@ export const DEFAULT_RETRY_OPTIONS: RetryOptions = {
 
 /**
  * Default predicate function to determine if a retry should be attempted.
- * Retries on 429 (Too Many Requests) and 5xx server errors.
+ * Retries on 5xx server errors, on provider rate-limit codes, and on anything
+ * else the classifier calls retryable — transport failures and status-less
+ * upstream error bodies, which no HTTP status can decide.
  * @param error The error object.
  * @returns True if the error is a transient error, false otherwise.
  */
@@ -114,19 +118,16 @@ function defaultShouldRetry(
     return error.shouldRetry(extraRetryErrorCodes);
   }
   const status = getErrorStatus(error);
-  // isRateLimitError already covers HTTP 429 (and 503) via RATE_LIMIT_ERROR_CODES,
-  // so an explicit `status === 429` check here would be redundant.
-  if (
-    isRateLimitError(error, extraRetryErrorCodes) ||
-    (status !== undefined && status >= 500 && status < 600)
-  ) {
+  if (status !== undefined && status >= 500 && status < 600) {
     return true;
   }
-  // Transport errors (ECONNRESET, ETIMEDOUT, etc.) carry no HTTP status and
-  // would otherwise fall through every predicate above.
-  return (
-    classifyRetryError(error, { extraRetryErrorCodes }).kind === 'transport'
-  );
+  // HTTP 429/503 and the provider rate-limit codes reach the retry through the
+  // shared verdict, whose classification already returns 'retryable' for them;
+  // the verdict's separate rate-limit term matters only where a higher branch
+  // classified the error fail-fast, and its doc names that case. Sharing the
+  // verdict with LlmChat's inline stream predicate is what keeps a change to
+  // the status-less policy from landing on one path and not the other.
+  return isRetryableUpstreamError(error, extraRetryErrorCodes);
 }
 
 /**
@@ -339,13 +340,15 @@ export async function retryWithBackoff<T>(
     } catch (error) {
       const errorStatus = getErrorStatus(error);
 
-      // Classification drives logging plus one control decision: a 'fail-fast'
+      // Classification drives logging plus three control decisions in this
+      // function: an 'abort' kind rethrows immediately below; a 'fail-fast'
       // verdict keeps a permanent error out of the unbounded persistent loop
-      // (see shouldPersist below). Normal retry control still follows
-      // shouldRetryOnError and the persistent policy. Computed before the Qwen
-      // quota fast-fail so the original error (status, request id, provider
-      // body) is always classified and logged, even when we replace it with a
-      // guidance message.
+      // (see shouldPersist below); and a 'retryable' verdict decides the retry
+      // itself wherever defaultShouldRetry is the operative shouldRetryOnError
+      // — which it is unless the caller supplied its own. Computed before the
+      // Qwen quota fast-fail so the original error (status, request id,
+      // provider body) is always classified and logged, even when we replace it
+      // with a guidance message.
       const retryDiagnostics = classifyRetryError(error, {
         authType,
         extraRetryErrorCodes,

packages/core/src/utils/retryErrorClassification.test.ts

@@ -4,12 +4,14 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import { APIUserAbortError } from 'openai';
+import { APIError as AnthropicAPIError } from '@anthropic-ai/sdk';
+import { APIError, APIUserAbortError } from 'openai';
 import { describe, expect, it } from 'vitest';
 import { AuthType } from '../core/contentGenerator.js';
 import {
   classifyRetryError,
   isFallbackEligible,
+  isRetryableUpstreamError,
 } from './retryErrorClassification.js';
 
 describe('classifyRetryError', () => {
@@ -567,13 +569,477 @@ describe('classifyRetryError', () => {
     });
 
     expect(classifyRetryError(error)).toMatchObject({
-      kind: 'unknown',
-      diagnosis: 'unknown',
+      kind: 'provider',
+      diagnosis: 'retryable',
       providerCode: 'Throttling.Custom',
       providerMessage: 'Provider-specific throttle',
       requestId: 'req-direct-error',
+      reason: 'upstream-error-without-status',
+    });
+  });
+
+  it('classifies a mid-stream upstream error with no HTTP status as retryable', () => {
+    // A gateway that pushes `{"error": {...}}` into an already-200 SSE stream
+    // reaches us as `new APIError(undefined, data.error, undefined,
+    // response.headers)`: no status, the body's `code`/`message`, and the
+    // response's `x-request-id` under the SDK's `requestID` spelling. Observed
+    // in the wild as `code: 'KeyError'`, `message: "'id'"`, which used to fall
+    // through to 'unknown' and kill the turn on the first attempt.
+    const error = Object.assign(new Error("'id'"), {
+      code: 'KeyError',
+      requestID: 'cd7f37f3-d38a-9dec-804f-f70dda5650eb',
+    });
+
+    const classification = classifyRetryError(error);
+    expect(classification).toMatchObject({
+      kind: 'provider',
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+      providerCode: 'KeyError',
+      providerMessage: "'id'",
+      requestId: 'cd7f37f3-d38a-9dec-804f-f70dda5650eb',
+    });
+    expect(classification).not.toHaveProperty('statusCode');
+  });
+
+  it('classifies the SDK error a mid-stream gateway frame actually produces', () => {
+    // The fixtures above hand-build the shape, so a dependency bump that
+    // renames `requestID` would silently reintroduce the incident with the
+    // whole suite green. This drives the real constructor the SDK throws from
+    // inside its SSE iterator, with real headers, as the oracle.
+    const error = new APIError(
+      undefined,
+      { code: 'KeyError', message: "'id'" },
+      undefined,
+      new Headers({ 'x-request-id': 'cd7f37f3-d38a-9dec-804f-f70dda5650eb' }),
+    );
+
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'provider',
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+      providerCode: 'KeyError',
+      requestId: 'cd7f37f3-d38a-9dec-804f-f70dda5650eb',
+    });
+    expect(isRetryableUpstreamError(error)).toBe(true);
+
+    // The same producer with a present-but-empty header: `Headers.get` returns
+    // '' rather than null, and an id the provider never set must not open the
+    // gate.
+    const untraced = new APIError(
+      undefined,
+      { code: 'KeyError', message: "'id'" },
+      undefined,
+      new Headers({ 'x-request-id': '' }),
+    );
+    expect(untraced.requestID).toBe('');
+    expect(classifyRetryError(untraced)).toMatchObject({
+      kind: 'unknown',
+      diagnosis: 'unknown',
       reason: 'unclassified',
     });
+    expect(isRetryableUpstreamError(untraced)).toBe(false);
+  });
+
+  it('classifies the error an Anthropic mid-stream frame actually produces', () => {
+    // The Anthropic SDK raises a mid-stream SSE failure with
+    // `APIError.generate(undefined, ..., sse.data, createResponseHeaders(...))`
+    // (streaming.mjs). `generate` short-circuits on the missing status and
+    // builds an `APIConnectionError` *without* headers, so the `request-id`
+    // the call site passed never reaches `request_id` — a native frame cannot
+    // open the status-less gate on a header id the way the OpenAI SDK's can.
+    // What still can is a gateway relaying its own id inside the frame body,
+    // which is what an Anthropic-compatible `baseUrl` route sees, and the body
+    // reaches the classifier because `generate` keeps the raw frame as the
+    // message.
+    const tracedFrame = JSON.stringify({
+      type: 'error',
+      error: { type: 'api_error', message: 'Internal server error' },
+      request_id: 'gw-trace-1',
+    });
+    const error = AnthropicAPIError.generate(
+      undefined,
+      `SSE Error: ${tracedFrame}`,
+      tracedFrame,
+      // `createResponseHeaders` hands `generate` a plain lower-cased record,
+      // not a `Headers` instance.
+      { 'request-id': 'header-trace-1' },
+    );
+
+    // The header id is dropped even though the call site supplied it; the body
+    // id survives through the message.
+    expect(error.request_id).toBeUndefined();
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'provider',
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+      requestId: 'gw-trace-1',
+    });
+    expect(isRetryableUpstreamError(error)).toBe(true);
+
+    // The same body channel cannot smuggle a permanent rejection into a retry:
+    // Anthropic puts `invalid_request_error` in `type`, the payload reader
+    // folds that into the provider code, and the permanence guard runs before
+    // the request-id branch.
+    const permanentFrame = JSON.stringify({
+      type: 'error',
+      error: {
+        type: 'invalid_request_error',
+        message: 'max_tokens: field required',
+      },
+      request_id: 'gw-trace-2',
+    });
+    const permanent = AnthropicAPIError.generate(
+      undefined,
+      `SSE Error: ${permanentFrame}`,
+      permanentFrame,
+      { 'request-id': 'header-trace-2' },
+    );
+
+    expect(classifyRetryError(permanent)).toMatchObject({
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+    });
+    expect(isRetryableUpstreamError(permanent)).toBe(false);
+
+    // The credential member of the same union, which the SDK maps to 401 when
+    // a status survives: relaying it status-less must not change the verdict,
+    // or a dead key costs the whole ladder before it surfaces.
+    const authFrame = JSON.stringify({
+      type: 'error',
+      error: { type: 'authentication_error', message: 'invalid x-api-key' },
+      request_id: 'gw-trace-3',
+    });
+    const unauthenticated = AnthropicAPIError.generate(
+      undefined,
+      `SSE Error: ${authFrame}`,
+      authFrame,
+      { 'request-id': 'header-trace-3' },
+    );
+
+    expect(classifyRetryError(unauthenticated)).toMatchObject({
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+    });
+    expect(isRetryableUpstreamError(unauthenticated)).toBe(false);
+  });
+
+  it('classifies a status-less provider body embedded in the message as retryable', () => {
+    // The same upstream failure can arrive with the provider's JSON body pasted
+    // into the message rather than on SDK properties. With no `:HTTP_STATUS/`
+    // marker there is no status to classify on, so the request id in the body
+    // is the only evidence that the provider traced the failure. Raw SSE
+    // framing surviving into the message is what earns the `sse-provider` kind
+    // here; the SDK strips that framing, so the case above is plain `provider`.
+    const error = new Error(
+      'id:1\nevent:error\ndata:{"request_id":"req-stream","code":"KeyError","message":"upstream failed"}',
+    );
+
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'sse-provider',
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+      requestId: 'req-stream',
+    });
+  });
+
+  it('fails fast on a permanent provider code scraped from the message', () => {
+    // The permanence guard reads the merged providerCode
+    // (`details.providerCode ?? providerFields.providerCode`), and this fixture
+    // is the message-scraped half of that merge: the error carries no `.code`
+    // property, so the moderation code reaches the guard only through the JSON
+    // in the message. Reading the merge object-only flips it from fail-fast to
+    // retryable. Several other cases here read the same half — the rate-limit
+    // diagnostics and the nested-`.error` sibling among them — so this is not
+    // the only pin on it, just the one that pins it for a permanent code. The
+    // case above does not read that half at all: its request id comes from a
+    // separate reader, so KeyError stays unlisted and retryable either way.
+    const error = new Error(
+      'id:1\nevent:error\ndata:{"request_id":"req-stream","code":"data_inspection_failed","message":"Output data may contain inappropriate content."}',
+    );
+
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'sse-provider',
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+      providerCode: 'data_inspection_failed',
+      requestId: 'req-stream',
+    });
+    expect(isRetryableUpstreamError(error)).toBe(false);
+  });
+
+  it('fails fast on a permanent provider code nested under .error', () => {
+    // `getProviderErrorPayload`'s isApiError fallback reads a nested
+    // `.error.code` when no JSON survives in the message — a second input
+    // shape that reaches the permanence guard only through the scraped half
+    // of the providerCode merge.
+    const error = Object.assign(new Error('moderation rejection'), {
+      error: {
+        code: 'data_inspection_failed',
+        message: 'Output data may contain inappropriate content.',
+      },
+      requestID: 'req-nested',
+    });
+
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'provider',
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+      providerCode: 'data_inspection_failed',
+      requestId: 'req-nested',
+    });
+    expect(isRetryableUpstreamError(error)).toBe(false);
+  });
+
+  it('fails fast on a permanent provider code even when the request is traced', () => {
+    // A request id decides upstream vs. local, not transient vs. permanent.
+    // Moderation, credential/billing and malformed-request rejections arrive
+    // after the 200 on a streaming call, so no status is left to fail fast on —
+    // without this they would walk the whole production ladder for a verdict
+    // that was never going to change.
+    const codes = [
+      'content_filter',
+      'data_inspection_failed',
+      // The same rejection the pipeline re-throws out of the provider's body.
+      'DataInspectionFailed',
+      // DashScope spells output moderation with a prefix.
+      'ResponseDataInspectionFailed',
+      'InvalidApiKey',
+      'Arrearage',
+      // Billing exhaustion that neither quota fast-fail intercepts: one needs a
+      // 429 status plus the free-tier wording, the other a reset time.
+      'insufficient_quota',
+      'Model.AccessDenied',
+      'invalid_request_error',
+      'InvalidParameter',
+      // OpenAI's `.type` spelling for a malformed request.
+      'invalid_parameter_error',
+      // The rest of the pinned Anthropic `ErrorObject` union whose verdict can
+      // never change on a re-send: credentials, entitlement, a model that does
+      // not exist, and billing. A gateway relaying one of them into an
+      // already-200 stream supplies an id and no status, so without these the
+      // request-id branch walks the whole ladder for a verdict that was fixed
+      // on arrival.
+      'authentication_error',
+      'permission_error',
+      'not_found_error',
+      'billing_error',
+      // Recoverable by compaction, never by re-sending the identical payload —
+      // the reasoning that already puts `context_length_exceeded` on this list.
+      'request_too_large',
+      // Recoverable by compaction, never by re-sending the identical payload.
+      'context_length_exceeded',
+    ];
+
+    for (const code of codes) {
+      expect(classifyRetryError({ code, requestID: 'req-1' })).toMatchObject({
+        diagnosis: 'fail-fast',
+        reason: 'permanent-provider-code',
+        providerCode: code,
+      });
+    }
+  });
+
+  it('fails fast on a permanent provider type when the body carries no code', () => {
+    // The canonical OpenAI malformed-request body puts `invalid_request_error`
+    // on `.type` and sets `code` to null, so reading permanence off `code`
+    // alone never fires for it and the request id would open the retry gate on
+    // a rejection that cannot succeed.
+    expect(
+      classifyRetryError({
+        type: 'invalid_request_error',
+        code: null,
+        requestID: 'req-1',
+      }),
+    ).toMatchObject({
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+    });
+  });
+
+  it('fails fast on a permanent provider type when the body also carries a code', () => {
+    // R16-1. On the message-embedded route the provider body is scraped, and
+    // `getRateLimitErrorDetails` collapses that body's `code` and `type` into a
+    // single `providerCode` (`String(payload.code ?? payload.type)`), so a
+    // permanent `type` was dropped whenever a sibling `code` survived. The
+    // object route never had the hole — `getProviderFields` reads `.type`
+    // separately. Moderation is the case the permanence list exists for: a
+    // gateway relaying `type: 'content_filter'` beside its own `code` is still
+    // a rejection that re-sending the identical request cannot change.
+    const moderation = new Error(
+      'event:error\ndata:{"error":{"message":"blocked","type":"content_filter","code":"moderation_blocked"},"request_id":"req-1"}',
+    );
+    expect(classifyRetryError(moderation)).toMatchObject({
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+    });
+    expect(isRetryableUpstreamError(moderation)).toBe(false);
+
+    // The same collapse on OpenAI's malformed-request shape, which carries both
+    // fields: `.type` names the permanent class, `.code` the specific field.
+    const malformed = new Error(
+      'event:error\ndata:{"error":{"type":"invalid_request_error","code":"missing_required_field","message":"x is required"},"request_id":"req-2"}',
+    );
+    expect(classifyRetryError(malformed)).toMatchObject({
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+    });
+    expect(isRetryableUpstreamError(malformed)).toBe(false);
+
+    // The other end of the same knob: reading `type` off the scraped body must
+    // not make every body-named class permanent. A transient one keeps the
+    // verdict the request-id branch gives it.
+    const transient = new Error(
+      'event:error\ndata:{"error":{"message":"upstream died","type":"api_error","code":"upstream_500"},"request_id":"req-3"}',
+    );
+    expect(classifyRetryError(transient)).toMatchObject({
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+    });
+    expect(isRetryableUpstreamError(transient)).toBe(true);
+  });
+
+  it('fails fast on a permanent provider type from the real SDK error', () => {
+    // The hand-built case above pins the guard's reaction to an assumed SDK
+    // output; this drives the real constructor, so a dependency bump that
+    // stops mapping the body's `type` onto the instance property reds it —
+    // the `.type` sibling of the requestID oracle above.
+    const error = new APIError(
+      undefined,
+      { type: 'invalid_request_error', code: null, message: 'x is required' },
+      undefined,
+      new Headers({ 'x-request-id': 'req-1' }),
+    );
+
+    expect(classifyRetryError(error)).toMatchObject({
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+    });
+    expect(isRetryableUpstreamError(error)).toBe(false);
+  });
+
+  it('does not treat every provider type as permanent', () => {
+    // `.type` also carries transient values; matching it against the same
+    // anchored list is what keeps a server-side fault retryable. These are the
+    // transient members of the pinned Anthropic `ErrorObject` union plus
+    // OpenAI's `server_error` — exactly what the permanent spellings must not
+    // swallow. `timeout_error` is the one a broad `.*_error` alternative would
+    // have caught, turning a gateway timeout into a fail-fast.
+    for (const type of ['api_error', 'timeout_error', 'server_error']) {
+      expect(classifyRetryError({ type, requestID: 'req-1' })).toMatchObject({
+        diagnosis: 'retryable',
+        reason: 'upstream-error-without-status',
+      });
+    }
+    // The throttles keep their own arm, which owns the Retry-After-aware delay.
+    for (const type of ['rate_limit_error', 'overloaded_error']) {
+      expect(classifyRetryError({ type, requestID: 'req-1' })).toMatchObject({
+        diagnosis: 'retryable',
+        reason: 'rate-limit',
+      });
+    }
+  });
+
+  it('keeps an unrecognised upstream code retryable', () => {
+    // The point of the branch: the next gateway bug should not have to be
+    // taught to the classifier before it stops killing turns.
+    expect(
+      classifyRetryError({ code: 'KeyError', requestID: 'req-1' }),
+    ).toMatchObject({
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+    });
+  });
+
+  it('treats an empty request id as no request id', () => {
+    // `headers.get('x-request-id')` yields '' for a header that is present but
+    // empty — legal HTTP, and what a proxy emits when the upstream set none. An
+    // error the provider never traced must not open the retry gate.
+    expect(
+      classifyRetryError({ code: 'KeyError', requestID: '' }),
+    ).toMatchObject({
+      kind: 'unknown',
+      diagnosis: 'unknown',
+      reason: 'unclassified',
+    });
+  });
+
+  it('treats an empty request id scraped from the message as no request id', () => {
+    // The same rule on the other reader: the rate-limit details scrape a
+    // provider body out of the message and do not reject an empty id, so the
+    // merge has to fall through it rather than coalesce onto it.
+    const error = new Error(
+      'id:1\nevent:error\ndata:{"request_id":"","code":"KeyError","message":"upstream failed"}',
+    );
+
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'unknown',
+      diagnosis: 'unknown',
+      reason: 'unclassified',
+    });
+  });
+
+  it('keeps permanent local failures without a request id unclassified', () => {
+    // A string `code` alone must not open the status-less retry gate — these
+    // are permanent, and retrying them burns the whole ladder for nothing.
+    const errors = [
+      Object.assign(new Error('No API key configured'), {
+        code: 'MISSING_API_KEY',
+      }),
+      Object.assign(new Error('Invalid MCP server configuration'), {
+        code: 'invalid_config',
+      }),
+      // MCP protocol errors carry a numeric JSON-RPC code.
+      Object.assign(new Error('Internal error'), { code: -32603 }),
+    ];
+
+    for (const error of errors) {
+      expect(classifyRetryError(error)).toMatchObject({
+        kind: 'unknown',
+        diagnosis: 'unknown',
+        reason: 'unclassified',
+      });
+    }
+  });
+
+  it('keeps a definitive HTTP status authoritative over a request id', () => {
+    // A traced 4xx is still a permanent client error: the status block runs
+    // before the status-less branches, so it cannot become retryable.
+    expect(
+      classifyRetryError({
+        status: 400,
+        code: 'invalid_request_error',
+        request_id: 'req-400',
+        message: 'malformed tool call',
+      }),
+    ).toMatchObject({
+      kind: 'http',
+      diagnosis: 'fail-fast',
+      reason: 'client-error',
+      statusCode: 400,
+    });
+  });
+
+  it('keeps a provider-traced socket cut transport-classified over a request id', () => {
+    // The transport branch runs first, and that ordering is load-bearing:
+    // isRetryableStreamTransportError admits mid-stream replay only on
+    // `kind === 'transport'` plus an allow-listed code, so reclassifying a
+    // traced socket cut as a provider error would silently disable replay,
+    // continuation recovery, and Anthropic's release of deferred tool calls
+    // while the whole suite stayed green. `transportCode` is asserted because
+    // that predicate reads it, and no request id is asserted because the
+    // transport return deliberately does not spread the provider fields.
+    const error = Object.assign(new Error('upstream failed'), {
+      requestID: 'req-1',
+      cause: Object.assign(new Error('socket reset'), { code: 'ECONNRESET' }),
+    });
+
+    expect(classifyRetryError(error)).toMatchObject({
+      kind: 'transport',
+      diagnosis: 'retryable',
+      reason: 'transport-error',
+      transportCode: 'ECONNRESET',
+    });
   });
 
   it('does not copy unparsed SSE frames into providerMessage', () => {
@@ -659,4 +1125,24 @@ describe('isFallbackEligible', () => {
       }),
     ).toBe(false);
   });
+
+  it('returns false for a status-less upstream error', () => {
+    // The property under test is "retryable, yet still not fallback-eligible":
+    // with no HTTP status there is no capacity signal, so retries stay on the
+    // primary model. Anchored to the classification as well as the predicate —
+    // asserting `false` alone cannot discriminate, because a status-less error
+    // classified `unknown` is not fallback-eligible either.
+    const classification = classifyRetryError({
+      code: 'KeyError',
+      requestID: 'req-stream',
+    });
+
+    expect(classification).toMatchObject({
+      kind: 'provider',
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+    });
+    expect(classification.statusCode).toBeUndefined();
+    expect(isFallbackEligible(classification)).toBe(false);
+  });
 });

packages/core/src/utils/retryErrorClassification.ts

@@ -41,10 +41,12 @@ export interface RetryErrorClassification {
  * Classifies retry-related failures.
  *
  * The result is primarily diagnostic — it labels the observed error shape for
- * logging. It also feeds a single control decision in `retryWithBackoff`: a
- * `'fail-fast'` diagnosis keeps a permanent error (e.g. allocated-quota
- * exhaustion surfacing as HTTP 429) out of the unbounded persistent loop.
- * Beyond that, it does not drive retry, fail-fast, or fallback control.
+ * logging. It also drives control: `isRetryableUpstreamError` below turns a
+ * `'retryable'` diagnosis into the retry verdict for every error an HTTP status
+ * cannot decide, and both retry gates end there; a `'fail-fast'` diagnosis keeps
+ * a permanent error (e.g. allocated-quota exhaustion surfacing as HTTP 429) out
+ * of the unbounded persistent loop; and `isFallbackEligible` reads the status
+ * recorded here to decide model fallback.
  */
 export function classifyRetryError(
   error: unknown,
@@ -64,7 +66,11 @@ export function classifyRetryError(
   const providerCode = details.providerCode ?? providerFields.providerCode;
   const providerMessage =
     details.providerMessage ?? providerFields.providerMessage;
-  const requestId = details.requestId ?? providerFields.requestId;
+  // `||`, not `??`: an empty string is not an identifier. The SDK stamps
+  // `requestID` from `headers.get('x-request-id')`, which yields '' for a
+  // header that is present but empty, and a request id now decides a retry
+  // verdict below — so '' must fall through the same way undefined does.
+  const requestId = details.requestId || providerFields.requestId;
   const common = {
     ...(statusCode !== undefined ? { statusCode } : {}),
     ...(providerCode !== undefined ? { providerCode } : {}),
@@ -218,6 +224,66 @@ export function classifyRetryError(
     };
   }
 
+  // Mirrors the transport-mapping branches above, which label by
+  // `details.transport`. That field records whether raw SSE framing survived
+  // into `.message`, not whether the error came out of a stream: the SDK strips
+  // the framing from a mid-stream APIError, so the canonical gateway error
+  // frame is labelled 'provider' and only a body pasted into the message reads
+  // 'sse-provider'.
+  const statuslessKind: RetryErrorKind =
+    details.transport === 'sse' ? 'sse-provider' : 'provider';
+
+  // With no status left to read, permanence has to come off the provider body:
+  // a moderation rejection necessarily arrives after the 200, and a gateway can
+  // relay a credential, billing or malformed-request rejection the same way.
+  // Nothing above can fail fast on them, and re-sending the identical request
+  // can never succeed.
+  //
+  // `.type` is read from both sources `code` is read from, because a body can
+  // carry a specific `code` beside a class-naming `type` and either may be the
+  // permanent one. On the object route `getProviderFields` reads the instance
+  // property — that is where the SDK puts `invalid_request_error`, with `code`
+  // null. On the message-scraped route the two arrive separately too:
+  // `details.providerCode` is the collapsed `code ?? type`, so a sibling `code`
+  // hides the `type` unless `details.providerType` is consulted as well.
+  if (
+    isPermanentProviderCode(providerCode) ||
+    isPermanentProviderCode(providerFields.providerType) ||
+    isPermanentProviderCode(details.providerType)
+  ) {
+    return {
+      kind: statuslessKind,
+      diagnosis: 'fail-fast',
+      reason: 'permanent-provider-code',
+      ...common,
+    };
+  }
+
+  // An upstream error body that carries the provider's own request id but no
+  // HTTP status. The OpenAI SDK builds exactly this shape when a gateway pushes
+  // `{"error": {...}}` into an already-200 SSE stream —
+  // `new APIError(undefined, data.error, undefined, response.headers)` — so the
+  // status never reaches us, and a server-side failure (observed in the wild as
+  // `code: 'KeyError'`, `message: "'id'"`) would otherwise fall through to
+  // 'unknown' and kill the turn on the first attempt while a socket cut above
+  // is retried. The id can equally arrive inside a provider JSON body embedded
+  // in the message.
+  //
+  // Only a request id opens this gate, and only for a code the list above does
+  // not already know as permanent. Local failures carry a string `code` but no
+  // request id (`MISSING_API_KEY`, `invalid_config`, MCP's numeric JSON-RPC
+  // codes), so they stay unclassified and are not retried. An unrecognised
+  // upstream code stays retryable on purpose: the next gateway bug should not
+  // have to be taught to this file before it stops killing turns.
+  if (requestId !== undefined) {
+    return {
+      kind: statuslessKind,
+      diagnosis: 'retryable',
+      reason: 'upstream-error-without-status',
+      ...common,
+    };
+  }
+
   return {
     kind: 'unknown',
     diagnosis: 'unknown',
@@ -305,8 +371,44 @@ function isAllocatedQuotaExceeded(providerCode?: string): boolean {
   return providerCode === 'Throttling.AllocationQuota';
 }
 
+// Provider codes meaning "re-sending this exact request can never succeed":
+// content moderation, credentials/billing, a malformed request, and a payload
+// refused for its size. Moderation necessarily happens after the provider has
+// already sent 200, so there is no HTTP status to fail fast on, and a gateway
+// can relay the others the same way. The oversized-payload codes belong here
+// for the same reason: their recovery is compaction, and re-sending the
+// identical payload cannot reach it.
+//
+// Separators are optional and case is ignored because one rejection reaches
+// this classifier in several spellings (`data_inspection_failed`,
+// `DataInspectionFailed`, `ResponseDataInspectionFailed`). The anchors are
+// deliberate: under-matching costs a wasted retry ladder, over-matching costs a
+// transient failure that is never retried — the bug this branch exists to fix —
+// so open-ended sub-code qualifiers such as `InvalidParameter.Range` are left
+// out on purpose.
+//
+// The credential, entitlement, missing-model and billing spellings are `type`
+// values the Anthropic SDK maps to 401/403/404 (400 for billing), every one of
+// which the HTTP-status branch above already fails fast on. A gateway relaying
+// one into an already-200 stream loses the status, and this list is what keeps
+// the verdict the same instead of spending the ladder on it. The union's
+// transient members stay out on purpose: `api_error`, `timeout_error` and
+// OpenAI's `server_error` remain retryable through the request-id branch
+// below, while `rate_limit_error` and `overloaded_error` are caught earlier by
+// the rate-limit arm and its Retry-After-aware delay.
+const PERMANENT_PROVIDER_CODE_PATTERN =
+  /^(?:response[_-]?data[_-]?inspection[_-]?failed|data[_-]?inspection[_-]?failed|content[_-]?filter|invalid[_-]?api[_-]?key|arrearage|insufficient[_-]?quota|model[._-]?access[_-]?denied|invalid[_-]?request[_-]?error|authentication[_-]?error|permission[_-]?error|not[_-]?found[_-]?error|billing[_-]?error|invalid[_-]?parameter(?:[_-]?error)?|context[_-]?length[_-]?exceeded|request[_-]?too[_-]?large)$/i;
+
+function isPermanentProviderCode(providerCode?: string): boolean {
+  return (
+    providerCode !== undefined &&
+    PERMANENT_PROVIDER_CODE_PATTERN.test(providerCode)
+  );
+}
+
 interface ProviderFields {
   providerCode?: string;
+  providerType?: string;
   providerMessage?: string;
   requestId?: string;
 }
@@ -318,9 +420,11 @@ function getProviderFields(error: unknown): ProviderFields {
 
   const source = error as {
     code?: unknown;
+    type?: unknown;
     message?: unknown;
     request_id?: unknown;
     requestId?: unknown;
+    requestID?: unknown;
   };
   const rawCode =
     typeof source.code === 'string' || typeof source.code === 'number'
@@ -336,12 +440,15 @@ function getProviderFields(error: unknown): ProviderFields {
     (error instanceof Error && rawCode?.startsWith('ERR_')) || isHttpStatusEcho
       ? undefined
       : rawCode;
-  const requestId =
-    typeof source.request_id === 'string'
-      ? source.request_id
-      : typeof source.requestId === 'string'
-        ? source.requestId
-        : undefined;
+  // `requestID` is the OpenAI SDK's spelling — it stamps the response's
+  // `x-request-id` header onto every APIError, including the status-less ones
+  // built from a mid-stream error event. That header can be present and empty,
+  // and this value now decides a retry verdict, so '' counts as absent.
+  const requestId = firstNonEmptyString(
+    source.request_id,
+    source.requestId,
+    source.requestID,
+  );
   const providerMessage =
     typeof source.message === 'string' &&
     (!(error instanceof Error) ||
@@ -349,9 +456,15 @@ function getProviderFields(error: unknown): ProviderFields {
       requestId !== undefined)
       ? source.message
       : undefined;
+  // `.type` is where the OpenAI SDK puts `invalid_request_error` — for that
+  // body it sets `code` to null — so permanence cannot be read off `code`
+  // alone. Kept out of the classification struct: it feeds the permanence
+  // guard only, and `providerCode` is compared by exact equality elsewhere.
+  const providerType = firstNonEmptyString(source.type);
 
   return {
     ...(providerCode !== undefined ? { providerCode } : {}),
+    ...(providerType !== undefined ? { providerType } : {}),
     ...(providerMessage !== undefined ? { providerMessage } : {}),
     ...(requestId !== undefined ? { requestId } : {}),
   };
@@ -377,3 +490,46 @@ export function isFallbackEligible(
     classification.diagnosis !== 'unknown'
   );
 }
+
+/**
+ * The retry verdict for everything a caller does not already short-circuit on
+ * an HTTP status of its own: HTTP 429/503 and provider rate-limit codes,
+ * transport failures, 5xx and 529, and status-less upstream bodies the provider
+ * traced with a request id.
+ *
+ * The rate-limit term is not the redundancy it looks like, but neither is it
+ * what keeps throttling retryable: `classifyRetryError` runs the same
+ * `isRateLimitError` with the same `extraRetryErrorCodes`, so a throttle this
+ * term matches already classifies `'retryable'` without it — deleting the term
+ * left the retry, classification and send-loop suites green apart from the one
+ * case below. Its only measurable effect is on an error a branch above the
+ * rate-limit one already owns: DashScope's allocated-quota exhaustion, which
+ * surfaces as HTTP 429 and classifies `fail-fast`. There the term keeps this
+ * verdict retryable, so `retryWithBackoff`'s persistent loop bounds the attempt
+ * count with its own fail-fast check (3 attempts) rather than stopping at the
+ * first. LlmChat's inline predicate short-circuits on its own `status === 429`
+ * line, so none of this applies on that path.
+ *
+ * Both gates end here so the status-less policy is written once. Kept in this
+ * module rather than in `retry.ts` because the package barrel re-exports that
+ * file, and this policy is not public API.
+ */
+export function isRetryableUpstreamError(
+  error: unknown,
+  extraRetryErrorCodes?: readonly number[],
+): boolean {
+  return (
+    isRateLimitError(error, extraRetryErrorCodes) ||
+    classifyRetryError(error, { extraRetryErrorCodes }).diagnosis ===
+      'retryable'
+  );
+}
+
+function firstNonEmptyString(...values: unknown[]): string | undefined {
+  for (const value of values) {
+    if (typeof value === 'string' && value) {
+      return value;
+    }
+  }
+  return undefined;
+}