fix(usage): price Claude fallbacks under the models that ran them

apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts

@@ -23,6 +23,7 @@ import * as TestClock from "effect/testing/TestClock";
 
 import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
 import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
+import * as EventSink from "../orchestration-v2/EventSink.ts";
 import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
 import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
 import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
@@ -198,17 +199,23 @@ it.effect.each([
         const orchestrator = yield* Orchestrator.OrchestratorV2;
         const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
         const service = yield* OrchestratorMcpService.OrchestratorMcpService;
+        const eventSink = yield* EventSink.EventSinkV2;
         const sweep = Effect.andThen(
           Effect.suspend(() => sweeps.get("usage-limit-recovery")!),
           worker.drain(),
         );
         const parentThreadId = ThreadId.make("thread:task-cancel-limit-parent");
         const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
-          orchestrator.streamDomainEvents.pipe(
-            Stream.filter(predicate),
-            Stream.take(1),
-            Stream.runDrain,
-            Effect.forkScoped,
+          eventSink.latestSequence().pipe(
+            Effect.flatMap((afterSequence) =>
+              eventSink.stream({ afterSequence }).pipe(
+                Stream.map((stored) => stored.event),
+                Stream.filter(predicate),
+                Stream.take(1),
+                Stream.runDrain,
+                Effect.forkScoped,
+              ),
+            ),
           );
         const turnRunning = () =>
           watch(

apps/server/src/provider/Drivers/claudeUsage.test.ts

@@ -220,6 +220,113 @@ describe("Claude advisor iterations", () => {
   });
 });
 
+describe("Claude fallback iterations", () => {
+  // Shaped like Claude Code's records of a server-side fallback: after a mid-stream decline the
+  // top-level usage names the requested model and merges cache writes across both attempts.
+  const fallbackTurn = (declinedOutputTokens: number, fallbackModel: string | undefined) => ({
+    type: "assistant",
+    timestamp: "2026-11-02T03:04:05.000Z",
+    sessionId: "5a128faa-8253-489e-b935-6c08e8e670c0",
+    requestId: "req_fallback",
+    message: {
+      id: "msg_fallback",
+      model: declinedOutputTokens > 0 ? "claude-opus-5-5" : "claude-opus-4-8",
+      usage: {
+        input_tokens: 6,
+        cache_read_input_tokens: 0,
+        cache_creation_input_tokens: declinedOutputTokens > 0 ? 30000 : 10000,
+        output_tokens: 900,
+        iterations: [
+          {
+            type: "message",
+            model: "claude-opus-5-5",
+            input_tokens: 3,
+            cache_read_input_tokens: 0,
+            cache_creation_input_tokens: 20000,
+            output_tokens: declinedOutputTokens,
+          },
+          {
+            type: "fallback_message",
+            model: fallbackModel,
+            input_tokens: 6,
+            cache_read_input_tokens: 0,
+            cache_creation_input_tokens: 10000,
+            output_tokens: 900,
+          },
+        ],
+      },
+    },
+  });
+  const served = {
+    provider: "claude",
+    timestampMs: Date.parse("2026-11-02T03:04:05.000Z"),
+    model: "claude-opus-4-8",
+    sessionId: "5a128faa-8253-489e-b935-6c08e8e670c0",
+    totals: {
+      uncachedInputTokens: 6,
+      cachedInputTokens: 0,
+      cacheCreationTokens: 10000,
+      outputTokens: 900,
+      reasoningTokens: 0,
+    },
+    reportedCostUsd: null,
+    speed: "standard",
+    dedupeKey: "msg_fallback:req_fallback:fallback",
+    supersedesDedupeKey: "msg_fallback:req_fallback",
+  };
+  const unattributed = {
+    ...served,
+    model: "claude-opus-5-5",
+    totals: { ...served.totals, cacheCreationTokens: 30000 },
+    dedupeKey: "msg_fallback:req_fallback",
+    supersedesDedupeKey: undefined,
+  };
+
+  it.each([
+    {
+      declined: "before any output",
+      declinedOutputTokens: 0,
+      fallbackModel: "claude-opus-4-8",
+      expected: [served],
+    },
+    {
+      declined: "mid-stream",
+      declinedOutputTokens: 250,
+      fallbackModel: "claude-opus-4-8",
+      expected: [
+        served,
+        {
+          ...served,
+          model: "claude-opus-5-5",
+          totals: {
+            uncachedInputTokens: 3,
+            cachedInputTokens: 0,
+            cacheCreationTokens: 20000,
+            outputTokens: 250,
+            reasoningTokens: 0,
+          },
+          dedupeKey: "msg_fallback:req_fallback:attempt:0",
+          supersedesDedupeKey: undefined,
+        },
+      ],
+    },
+    ...[undefined, "", "<synthetic>"].map((fallbackModel) => ({
+      declined: `mid-stream with unavailable serving model ${String(fallbackModel)}`,
+      declinedOutputTokens: 250,
+      fallbackModel,
+      expected: [unattributed],
+    })),
+  ])(
+    "attributes usage when the request declined $declined",
+    ({ declinedOutputTokens, fallbackModel, expected }) => {
+      const turn = fallbackTurn(declinedOutputTokens, fallbackModel);
+
+      expect(claudeUsageFormat.parseLine(JSON.stringify(turn), undefined)).toEqual(expected);
+      expect(claudeUsageFormat.parseProjected(turn, undefined)).toEqual(expected);
+    },
+  );
+});
+
 describe("Claude cache write pricing", () => {
   it("prices one-hour cache writes at the one-hour rate and the rest at the five-minute rate", () => {
     const model = "claude-opus-4-5-20251101";

apps/server/src/provider/Drivers/claudeUsage.ts

@@ -63,61 +63,86 @@ function parseClaudeRecord(parsed: unknown): readonly UsageRecord[] {
     messageId === null && requestId === null ? null : `${messageId ?? ""}:${requestId ?? ""}`;
 
   const cost = record["costUSD"];
+  const iterations: readonly unknown[] = Array.isArray(usageRecord["iterations"])
+    ? usageRecord["iterations"]
+    : [];
+  // After a server-side fallback the top-level usage can name the declined model and merge
+  // attempts; the serving `fallback_message` iteration holds what its own model billed.
+  const fallbackIndex = iterations.findLastIndex(
+    (iteration) => iterationRecord(iteration)?.["type"] === "fallback_message",
+  );
+  const fallback = iterationRecord(iterations[fallbackIndex]);
+  const fallbackModel = fallback === null ? null : iterationModel(fallback);
+  const serving = fallbackModel === null ? null : fallback;
+  const servingIndex = serving === null ? -1 : fallbackIndex;
 
   const executor: UsageRecord = {
     provider: "claude",
     timestampMs,
-    model,
+    model: fallbackModel ?? model,
     sessionId: typeof record["sessionId"] === "string" ? record["sessionId"] : "",
-    totals: {
-      uncachedInputTokens: tokenCount(usageRecord["input_tokens"]),
-      cachedInputTokens: tokenCount(usageRecord["cache_read_input_tokens"]),
-      cacheCreationTokens: tokenCount(usageRecord["cache_creation_input_tokens"]),
-      outputTokens: tokenCount(usageRecord["output_tokens"]),
-      // Anthropic folds thinking tokens into output and does not break them out.
-      reasoningTokens: 0,
-    },
-    ...cacheCreation1h(usageRecord),
+    totals: usageTotals(serving ?? usageRecord),
+    ...cacheCreation1h(serving ?? usageRecord),
     reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) ? cost : null,
     speed: usageRecord["speed"] === "fast" ? "fast" : "standard",
-    dedupeKey,
+    dedupeKey: serving === null || dedupeKey === null ? dedupeKey : `${dedupeKey}:fallback`,
+    ...(serving === null || dedupeKey === null ? {} : { supersedesDedupeKey: dedupeKey }),
   };
 
-  // Top-level `usage` is the executor's alone. Advisor calls run on another model at its own rates
-  // and appear only in `iterations`, so each one is its own record under that model.
+  // Advisor calls and declined attempts bill at their own model's rates. Anthropic bills a
+  // declined attempt that produced output; most declines before any output are free.
   const records = [executor];
-  const iterations: readonly unknown[] = Array.isArray(usageRecord["iterations"])
-    ? usageRecord["iterations"]
-    : [];
   for (const [index, iteration] of iterations.entries()) {
-    if (typeof iteration !== "object" || iteration === null) continue;
-    const advisor = iteration as Record<string, unknown>;
-    if (advisor["type"] !== "advisor_message") continue;
-
-    const advisorModel = typeof advisor["model"] === "string" ? advisor["model"] : "";
-    if (advisorModel.length === 0 || advisorModel === CLAUDE_SYNTHETIC_MODEL) continue;
+    const entry = iterationRecord(iteration);
+    if (entry === null) continue;
+    const kind =
+      entry["type"] === "advisor_message"
+        ? "advisor"
+        : entry["type"] === "message" &&
+            index < servingIndex &&
+            tokenCount(entry["output_tokens"]) > 0
+          ? "attempt"
+          : null;
+    const entryModel = iterationModel(entry);
+    if (kind === null || entryModel === null) continue;
 
     records.push({
       provider: "claude",
       timestampMs,
-      model: advisorModel,
+      model: entryModel,
       sessionId: executor.sessionId,
-      totals: {
-        uncachedInputTokens: tokenCount(advisor["input_tokens"]),
-        cachedInputTokens: tokenCount(advisor["cache_read_input_tokens"]),
-        cacheCreationTokens: tokenCount(advisor["cache_creation_input_tokens"]),
-        outputTokens: tokenCount(advisor["output_tokens"]),
-        reasoningTokens: 0,
-      },
-      ...cacheCreation1h(advisor),
+      totals: usageTotals(entry),
+      ...cacheCreation1h(entry),
       reportedCostUsd: null,
       speed: executor.speed,
-      dedupeKey: dedupeKey === null ? null : `${dedupeKey}:advisor:${index}`,
+      dedupeKey: dedupeKey === null ? null : `${dedupeKey}:${kind}:${index}`,
     });
   }
   return records;
 }
 
+function iterationRecord(value: unknown): Record<string, unknown> | null {
+  return typeof value === "object" && value !== null ? (value as Record<string, unknown>) : null;
+}
+
+function iterationModel(iteration: Record<string, unknown>): string | null {
+  const model = iteration["model"];
+  return typeof model === "string" && model.length > 0 && model !== CLAUDE_SYNTHETIC_MODEL
+    ? model
+    : null;
+}
+
+function usageTotals(usage: Record<string, unknown>): UsageRecord["totals"] {
+  return {
+    uncachedInputTokens: tokenCount(usage["input_tokens"]),
+    cachedInputTokens: tokenCount(usage["cache_read_input_tokens"]),
+    cacheCreationTokens: tokenCount(usage["cache_creation_input_tokens"]),
+    outputTokens: tokenCount(usage["output_tokens"]),
+    // Anthropic folds thinking tokens into output and does not break them out.
+    reasoningTokens: 0,
+  };
+}
+
 export const claudeUsageFormat: TranscriptUsageFormat<void> = {
   selectFields: {
     type: true,

apps/server/src/usage/UsageService.test.ts

@@ -3207,7 +3207,7 @@ describe("UsageService", () => {
             60 + 20 * (version === 4 ? 1 : 2),
           );
           yield* upgraded.awaitPersisted;
-          assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v17.json");
+          assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v18.json");
           assert.deepStrictEqual(
             (yield* Effect.promise(() => NodeFSP.readdir(stateDir)))
               .filter((name) => name.startsWith("usage-scan-cache"))
@@ -3437,42 +3437,70 @@ describe("UsageService", () => {
     }),
   );
 
-  it.effect("credits the same copy of a duplicate after its transcripts are deleted", () =>
-    Effect.gen(function* () {
-      const { transcript, settings, home } = yield* setup;
-      const dir = NodePath.dirname(transcript);
-      const [first = "", second = ""] = yield* Effect.promise(async () => {
-        await NodeFSP.writeFile(NodePath.join(dir, "a.jsonl"), "");
-        await NodeFSP.writeFile(NodePath.join(dir, "b.jsonl"), "");
-        return (await NodeFSP.readdir(dir)).map((name) => NodePath.join(dir, name));
-      });
-      const forked = (line: string) => line.replace('"session-1"', '"session-2"');
-      yield* Effect.promise(async () => {
-        await NodeFSP.writeFile(
-          first,
+  it.effect.each([
+    { fallback: false, reverse: false, keepOther: true },
+    { fallback: true, reverse: false, keepOther: false },
+    { fallback: true, reverse: true, keepOther: false },
+    { fallback: true, reverse: false, keepOther: true },
+  ])(
+    "credits surviving sessions after their transcripts are deleted (%j)",
+    ({ fallback, reverse, keepOther }) =>
+      Effect.gen(function* () {
+        const { transcript, settings, home } = yield* setup;
+        const dir = NodePath.dirname(transcript);
+        const [first = "", second = ""] = yield* Effect.promise(async () => {
+          await NodeFSP.writeFile(NodePath.join(dir, "a.jsonl"), "");
+          await NodeFSP.writeFile(NodePath.join(dir, "b.jsonl"), "");
+          return (await NodeFSP.readdir(dir)).map((name) => NodePath.join(dir, name));
+        });
+        const forked = (line: string) => line.replace('"session-1"', '"session-2"');
+        const completed = fallback
+          ? encodeUnknownJsonString({
+              type: "assistant",
+              timestamp: "2026-08-01T10:00:00Z",
+              requestId: "req_1",
+              sessionId: "session-1",
+              message: {
+                id: "msg_1",
+                model: "claude-fable-5",
+                usage: {
+                  ...COMPLETE_USAGE,
+                  iterations: [
+                    { type: "message", model: "claude-fable-5", output_tokens: 0 },
+                    { type: "fallback_message", model: "claude-opus-5-5", ...COMPLETE_USAGE },
+                  ],
+                },
+              },
+            }) + "\n"
+          : claudeLine(1, 5);
+        const partial =
           claudeLine(1, 5).replace(
             '"message":',
             '"padding":' + encodeUnknownJsonString("x".repeat(9 * 1024 * 1024)) + ',"message":',
-          ),
+          ) + (fallback && keepOther ? claudeLine(3, 11) : "");
+        const final = forked(completed) + (!fallback ? forked(claudeLine(2, 7)) : "");
+        yield* Effect.promise(async () => {
+          await NodeFSP.writeFile(first, reverse ? final : partial);
+          await NodeFSP.writeFile(second, reverse ? partial : final);
+        });
+        yield* Effect.gen(function* () {
+          const service = yield* UsageService.make;
+          const live = yield* service.readSummary(WINDOW);
+          assert.strictEqual(live.sources[0]?.distinctSessions, keepOther ? 2 : 1);
+          assert.strictEqual(totalOutputTokens(live), fallback ? 264 + (keepOther ? 11 : 0) : 12);
+
+          yield* Effect.promise(() => Promise.all([NodeFSP.rm(first), NodeFSP.rm(second)]));
+          const saved = yield* service.readSummary(WINDOW);
+          assert.deepStrictEqual(saved.buckets, live.buckets);
+          assert.deepStrictEqual(saved.sources, live.sources);
+          yield* service.awaitPersisted;
+          const restored = yield* (yield* UsageService.make).readSummary(WINDOW);
+          assert.deepStrictEqual(restored.buckets, live.buckets);
+          assert.deepStrictEqual(restored.sources, live.sources);
+        }).pipe(
+          Effect.provide(layerService({ prefix: "usage-service-copy-order-test", home, settings })),
         );
-        await NodeFSP.writeFile(second, forked(claudeLine(1, 5)) + forked(claudeLine(2, 7)));
-      });
-      yield* Effect.gen(function* () {
-        const service = yield* UsageService.make;
-        const live = yield* service.readSummary(WINDOW);
-        assert.strictEqual(live.buckets[0]?.sessions, 2);
-
-        yield* Effect.promise(() => Promise.all([NodeFSP.rm(first), NodeFSP.rm(second)]));
-        const saved = yield* service.readSummary(WINDOW);
-        assert.deepStrictEqual(saved.buckets, live.buckets);
-        assert.deepStrictEqual(saved.sources, live.sources);
-        yield* service.awaitPersisted;
-        const restored = yield* (yield* UsageService.make).readSummary(WINDOW);
-        assert.deepStrictEqual(restored.buckets, live.buckets);
-      }).pipe(
-        Effect.provide(layerService({ prefix: "usage-service-copy-order-test", home, settings })),
-      );
-    }),
+      }),
   );
 
   it.effect("credits the same copy of a duplicate from the last scan as an exact read does", () =>