fix(providers): close the streaming cache-telemetry gaps; honor the remaining fallback chain on streamed retries

Three P2 hygiene slices from the prompt-cache audit:

- AnthropicProvider samples the cache-leak detector on the streaming path
  (the RC9 blind spot: only generateCompletion sampled, leaving the
  streamed narrator/companion surfaces invisible to the zero-read and
  unmarked tripwires that were built after the July regressions).
- streamText's fallback recursion passes the REMAINING chain slice like
  generateText (undefined made the recursion rebuild the default chain,
  re-trying the failed primary and ignoring explicit chains) and the
  auto-built chain is now policy-aware (mature streamed turns get the
  uncensored prefix instead of refuse-happy availability legs).
- OpenAIProvider normalizes prompt_tokens_details.cached_tokens (and the
  Responses API's input_tokens_details) into cacheReadInputTokens, the
  same field the Anthropic and OpenRouter providers report — automatic
  caching on the gpt fallback legs stops being invisible.

Author: jddunn

src/api/streamText.ts

@@ -15,7 +15,7 @@ import { hostPolicyToRouteParams, mergeRequiredCapabilities } from './runtime/ho
 import { adaptTools } from './runtime/toolAdapter.js';
 import { runEmulatedToolLoop, type ToolMode } from './runtime/tool-emulation/index.js';
 import {
-  buildFallbackChain,
+  buildPolicyAwareFallbackChain,
   createPlan,
   isRetryableError,
   resolveChainOfThought,
@@ -969,9 +969,13 @@ export function streamText(opts: GenerateTextOptions): StreamTextResult {
       // call targeting the next available fallback.  All parts from the
       // fallback stream are yielded transparently to the consumer.
       // Resolve fallback chain: caller-supplied wins, undefined triggers
-      // auto-build from env keys, empty array explicitly opts out.
+      // auto-build from env keys, empty array explicitly opts out. The
+      // auto-build is POLICY-AWARE like generateText's (mature tiers get
+      // the uncensored prefix) — streaming previously used the plain
+      // availability chain, so a mature streamed turn that lost its
+      // primary fell onto refuse-happy legs first.
       const effectiveFallbacks = opts.fallbackProviders === undefined
-        ? buildFallbackChain(recordedProviderId)
+        ? buildPolicyAwareFallbackChain(opts.policyTier, recordedProviderId)
        : opts.fallbackProviders;
 
       if (effectiveFallbacks.length && isRetryableError(error)) {
@@ -1015,7 +1019,15 @@ export function streamText(opts: GenerateTextOptions): StreamTextResult {
               model: fb.model,
               apiKey: undefined,
               baseUrl: undefined,
-              fallbackProviders: undefined,
+              // Preserve the REMAINING chain (entries AFTER the current fb;
+              // `attempt` is 1-indexed so slice(attempt) drops fb and all
+              // already-tried entries), mirroring generateText. Passing
+              // `undefined` here made the recursion REBUILD the default
+              // chain — it could re-try the already-failed primary and
+              // ignored explicit frontier-only chains. The final entry
+              // passes [] -> explicit opt-out -> the recursion throws
+              // instead of looping.
+              fallbackProviders: effectiveFallbacks.slice(attempt),
               onFallback: undefined,
             });
 

src/core/llm/providers/__tests__/openai-streaming-usage.test.ts

@@ -176,3 +176,113 @@ describe('OpenAIProvider streaming usage', () => {
     expect(usageChunk!.usage!.totalTokens).toBe(46);
   });
 });
+
+describe('OpenAIProvider cached-token normalization (automatic prompt caching)', () => {
+  let provider: OpenAIProvider;
+
+  beforeEach(async () => {
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(
+        JSON.stringify({
+          object: 'list',
+          data: [{ id: 'gpt-4o', object: 'model', created: 1, owned_by: 'openai' }],
+        }),
+        { status: 200, headers: { 'content-type': 'application/json' } },
+      ),
+    );
+    provider = new OpenAIProvider();
+    await provider.initialize({ apiKey: 'sk-test', maxRetries: 1 });
+    vi.spyOn(globalThis, 'fetch').mockClear();
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it('normalizes prompt_tokens_details.cached_tokens on non-streaming completions', async () => {
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(
+        JSON.stringify({
+          id: 'chatcmpl-cached',
+          object: 'chat.completion',
+          created: 1,
+          model: 'gpt-4o',
+          choices: [
+            {
+              index: 0,
+              message: { role: 'assistant', content: 'hello' },
+              finish_reason: 'stop',
+              logprobs: null,
+            },
+          ],
+          usage: {
+            prompt_tokens: 1200,
+            completion_tokens: 40,
+            total_tokens: 1240,
+            prompt_tokens_details: { cached_tokens: 1024 },
+          },
+        }),
+        { status: 200, headers: { 'content-type': 'application/json' } },
+      ),
+    );
+
+    const result = await provider.generateCompletion(
+      'gpt-4o',
+      [{ role: 'user', content: 'hi' }],
+      {},
+    );
+    // Cached tokens surface on the same normalized field the Anthropic and
+    // OpenRouter providers use; prompt_tokens stays INCLUSIVE of them
+    // (OpenAI accounting), so promptTokens is unchanged.
+    expect(result.usage?.cacheReadInputTokens).toBe(1024);
+    expect(result.usage?.promptTokens).toBe(1200);
+  });
+
+  it('normalizes cached_tokens on the trailing streaming usage-only chunk', async () => {
+    const usageOnly = {
+      ...makeUsageOnlyChunk(500, 20),
+      usage: {
+        prompt_tokens: 500,
+        completion_tokens: 20,
+        total_tokens: 520,
+        prompt_tokens_details: { cached_tokens: 384 },
+      },
+    };
+    const sse = [
+      `data: ${JSON.stringify({
+        id: 'chatcmpl-2',
+        object: 'chat.completion.chunk',
+        created: 1,
+        model: 'gpt-4o',
+        choices: [
+          { index: 0, delta: { role: 'assistant', content: 'hi' }, finish_reason: null },
+        ],
+      })}\n\n`,
+      `data: ${JSON.stringify({
+        id: 'chatcmpl-2',
+        object: 'chat.completion.chunk',
+        created: 1,
+        model: 'gpt-4o',
+        choices: [{ index: 0, delta: {}, finish_reason: 'stop' }],
+      })}\n\n`,
+      `data: ${JSON.stringify(usageOnly)}\n\n`,
+      'data: [DONE]\n\n',
+    ].join('');
+
+    vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(sse, { status: 200, headers: { 'content-type': 'text/event-stream' } }),
+    );
+
+    const chunks: { isFinal?: boolean; usage?: { cacheReadInputTokens?: number } }[] = [];
+    for await (const chunk of provider.generateCompletionStream(
+      'gpt-4o',
+      [{ role: 'user', content: 'hi' }],
+      {},
+    )) {
+      chunks.push(chunk as { isFinal?: boolean; usage?: { cacheReadInputTokens?: number } });
+    }
+
+    const usageChunk = chunks.find((c) => c.usage && c.isFinal);
+    expect(usageChunk?.usage?.cacheReadInputTokens).toBe(384);
+  });
+});

src/core/llm/providers/implementations/AnthropicProvider.ts

@@ -638,7 +638,7 @@ export class AnthropicProvider implements IProvider {
   private recordCacheLeakSample(
     modelId: string,
     payload: Record<string, unknown>,
-    apiResponse: AnthropicMessagesResponse,
+    apiResponse: Pick<AnthropicMessagesResponse, 'usage'>,
   ): void {
     try {
       const system = payload.system as
@@ -1139,6 +1139,24 @@ export class AnthropicProvider implements IProvider {
               }),
             };
 
+            // Sample the leak detector on the STREAMING path too. This was
+            // the RC9 blind spot: only generateCompletion sampled, so the
+            // streamed conversation surfaces (narrator, companion) — where
+            // the 2026-07 history-caching regressions actually lived — were
+            // invisible to the zero-read / unmarked tripwires.
+            this.recordCacheLeakSample(modelId, payload, {
+              usage: {
+                input_tokens: inputTokens,
+                output_tokens: outputTokens,
+                ...(cacheCreationTokens !== undefined && {
+                  cache_creation_input_tokens: cacheCreationTokens,
+                }),
+                ...(cacheReadTokens !== undefined && {
+                  cache_read_input_tokens: cacheReadTokens,
+                }),
+              },
+            });
+
             yield {
               id: responseId,
               object: 'chat.completion.chunk',

src/core/llm/providers/implementations/OpenAIProvider.ts

@@ -55,6 +55,14 @@ namespace OpenAIAPITypes {
     prompt_tokens: number;
     completion_tokens: number;
     total_tokens: number;
+    /**
+     * OpenAI automatic prompt caching (gpt-4o and newer): prompt tokens
+     * served from cache at the discounted rate. Normalized into
+     * {@link ModelUsage.cacheReadInputTokens} so the gpt-5.5 / gpt-4o
+     * fallback legs' caching is visible platform-wide (OpenRouterProvider
+     * already normalizes the same field).
+     */
+    prompt_tokens_details?: { cached_tokens?: number };
   }
   /** Complete tool call as returned on a non-streaming message. */
   export interface ToolCall {
@@ -144,6 +152,8 @@ namespace OpenAIAPITypes {
     input_tokens: number;
     output_tokens: number;
     total_tokens: number;
+    /** Responses-API spelling of the cached-prompt-token detail. */
+    input_tokens_details?: { cached_tokens?: number };
   }
   export interface ResponsesOutputContentPart {
     type: string; // 'output_text' | …
@@ -1119,6 +1129,11 @@ export class OpenAIProvider implements IProvider {
               prompt_tokens: apiResponse.usage.input_tokens,
               completion_tokens: apiResponse.usage.output_tokens,
               total_tokens: apiResponse.usage.total_tokens,
+              // Responses API spells the cached detail input_tokens_details;
+              // re-key to the chat spelling the shared mapper reads.
+              ...(apiResponse.usage.input_tokens_details
+                ? { prompt_tokens_details: { cached_tokens: apiResponse.usage.input_tokens_details.cached_tokens } }
+                : {}),
             },
             apiResponse.model ?? modelId
           )
@@ -1307,14 +1322,29 @@ export class OpenAIProvider implements IProvider {
    * @private
    */
   private calculateUsage(
-    usage: { prompt_tokens: number; completion_tokens: number; total_tokens: number },
+    usage: {
+      prompt_tokens: number;
+      completion_tokens: number;
+      total_tokens: number;
+      prompt_tokens_details?: { cached_tokens?: number };
+    },
     modelId: string
   ): ModelUsage {
+    // OpenAI automatic caching reports cached prompt tokens as a detail
+    // field (they remain INCLUDED in prompt_tokens, unlike Anthropic's
+    // exclusive accounting). Normalize into cacheReadInputTokens so the
+    // fallback legs' caching stops being invisible in platform telemetry;
+    // cost stays computed off the full prompt_tokens (the discount is a
+    // billing-side rate, not a token-count change).
+    const cachedTokens = usage.prompt_tokens_details?.cached_tokens;
     return {
       promptTokens: usage.prompt_tokens,
       completionTokens: usage.completion_tokens,
       totalTokens: usage.total_tokens,
       costUSD: this.calculateCost(usage.prompt_tokens, usage.completion_tokens, modelId),
+      ...(typeof cachedTokens === 'number' && cachedTokens >= 0
+        ? { cacheReadInputTokens: cachedTokens }
+        : {}),
     };
   }