From bbc718fe0e85da52b7ecdafb7e239b20c22e6052 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 21:35:04 +1100 Subject: [PATCH 1/6] fix(usage): price Claude fallbacks under the models that ran them --- .../src/provider/Drivers/claudeUsage.test.ts | 86 +++++++++++++++++++ .../src/provider/Drivers/claudeUsage.ts | 79 ++++++++++------- apps/server/src/usage/UsageService.test.ts | 2 +- apps/server/src/usage/usageScanCache.test.ts | 6 +- apps/server/src/usage/usageScanCache.ts | 9 +- 5 files changed, 145 insertions(+), 37 deletions(-) diff --git a/apps/server/src/provider/Drivers/claudeUsage.test.ts b/apps/server/src/provider/Drivers/claudeUsage.test.ts index 6f457f4f4cf2f..bf11035e46eb8 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.test.ts +++ b/apps/server/src/provider/Drivers/claudeUsage.test.ts @@ -216,3 +216,89 @@ describe("Claude advisor iterations", () => { expect(parse(unusable).map((record) => record.model)).toEqual(["claude-opus-5-5"]); }); }); + +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) => ({ + 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: "claude-opus-4-8", + 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", + }; + + it.each([ + { declined: "before any output", declinedOutputTokens: 0, expected: [served] }, + { + declined: "mid-stream", + declinedOutputTokens: 250, + 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", + }, + ], + }, + ])( + "prices the serving fallback under its own model when the request declined $declined", + ({ declinedOutputTokens, expected }) => { + const turn = fallbackTurn(declinedOutputTokens); + + expect(claudeUsageFormat.parseLine(JSON.stringify(turn), undefined)).toEqual(expected); + expect(claudeUsageFormat.parseProjected(turn, undefined)).toEqual(expected); + }, + ); +}); diff --git a/apps/server/src/provider/Drivers/claudeUsage.ts b/apps/server/src/provider/Drivers/claudeUsage.ts index acb697d2941de..7ec60f3249d2f 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.ts +++ b/apps/server/src/provider/Drivers/claudeUsage.ts @@ -56,59 +56,80 @@ 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 servingIndex = iterations.findLastIndex( + (iteration) => iterationRecord(iteration)?.["type"] === "fallback_message", + ); + const serving = iterationRecord(iterations[servingIndex]); const executor: UsageRecord = { provider: "claude", timestampMs, - model, + model: (serving === null ? null : iterationModel(serving)) ?? 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, - }, + totals: usageTotals(serving ?? usageRecord), reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) ? cost : null, speed: usageRecord["speed"] === "fast" ? "fast" : "standard", 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; - 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, - }, + totals: usageTotals(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 | null { + return typeof value === "object" && value !== null ? (value as Record) : null; +} + +function iterationModel(iteration: Record): string | null { + const model = iteration["model"]; + return typeof model === "string" && model.length > 0 && model !== CLAUDE_SYNTHETIC_MODEL + ? model + : null; +} + +function usageTotals(usage: Record): 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 = { selectFields: { type: true, diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 1c03c05e13727..426d32bc02bd4 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2782,7 +2782,7 @@ describe("UsageService", () => { 60 + 20 * (version === 4 ? 1 : 2), ); yield* upgraded.awaitPersisted; - assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v12.json"); + assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v13.json"); assert.deepStrictEqual( (yield* Effect.promise(() => NodeFSP.readdir(stateDir))) .filter((name) => name.startsWith("usage-scan-cache")) diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index 2866a295da8fe..9f2fca6fc758b 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -244,11 +244,11 @@ describe("scan cache round trip", () => { }, ); - it("re-parses a Claude transcript cached before partial copies were merged", () => { + it.each([11, 12])("re-parses v%s Claude transcripts for corrected attempt usage", (version) => { const original = cacheWith([["/claude.jsonl", 100, [record()]]]); - const v11 = { ...parse(makeScanCacheWriter()(original, {})), version: 11 }; + const legacy = { ...parse(makeScanCacheWriter()(original, {})), version }; - const restored = decodeScanCache(JSON.parse(JSON.stringify(v11)), TEST_FORMAT_MAP); + const restored = decodeScanCache(JSON.parse(JSON.stringify(legacy)), TEST_FORMAT_MAP); expect(restored.get("/claude.jsonl")).toEqual({ ...original.get("/claude.jsonl"), size: -1, diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 4cd959c75aa25..80d2262e539aa 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -10,7 +10,7 @@ import { CLAUDE_SYNTHETIC_MODEL } from "../provider/Drivers/claudeUsage.ts"; import { mergeUsageCopies } from "./usageAggregation.ts"; import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts"; -const USAGE_SCAN_CACHE_VERSION = 12 as const; +const USAGE_SCAN_CACHE_VERSION = 13 as const; const SPEED_COMPATIBLE_SINCE_VERSION = 4; /** @@ -18,8 +18,9 @@ const SPEED_COMPATIBLE_SINCE_VERSION = 4; * replace it, dropping saved usage for deleted transcripts. Separate files keep * the newer one; older files are removed once it is saved. */ -export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v12.json"; +export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v13.json"; export const LEGACY_SCAN_CACHE_FILE_NAMES = [ + "usage-scan-cache-v12.json", "usage-scan-cache-v11.json", "usage-scan-cache-v10.json", "usage-scan-cache-v9.json", @@ -291,9 +292,9 @@ export function decodeScanCache( } // Keep history for deleted rollouts, but re-parse incompatible reducer state. // v9 and v11 repair skipped Codex requests; clean v8+ entries stay served while unchanged. - // v10 adds Claude advisor calls; before v12 a repeated Claude key kept its first, possibly partial, copy. + // Claude history before v13 lacks per-attempt fallback usage. const legacy = - (entry.p === "claude" && version < 12) || + (entry.p === "claude" && version < 13) || (format.state !== undefined && (version < 6 || (entry.p === "codex" && (version < 8 || (version < 11 && malformedRecords > 0))))); From f8b6a36b2db77b424f96d4bbdb91b91b3623aa3a Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 22:09:43 +1100 Subject: [PATCH 2/6] fix(usage): reconcile completed Claude fallback copies --- .../src/provider/Drivers/claudeUsage.test.ts | 33 +++++-- .../src/provider/Drivers/claudeUsage.ts | 12 ++- apps/server/src/usage/UsageService.test.ts | 2 +- .../server/src/usage/usageAggregation.test.ts | 99 +++++++++++++++++++ apps/server/src/usage/usageAggregation.ts | 12 +++ apps/server/src/usage/usageScanCache.test.ts | 23 +++-- apps/server/src/usage/usageScanCache.ts | 28 +++--- packages/provider-core/src/server/usage.ts | 2 + 8 files changed, 179 insertions(+), 32 deletions(-) diff --git a/apps/server/src/provider/Drivers/claudeUsage.test.ts b/apps/server/src/provider/Drivers/claudeUsage.test.ts index bf11035e46eb8..bee40e41acdbb 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.test.ts +++ b/apps/server/src/provider/Drivers/claudeUsage.test.ts @@ -220,7 +220,7 @@ 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) => ({ + const fallbackTurn = (declinedOutputTokens: number, fallbackModel: string | undefined) => ({ type: "assistant", timestamp: "2026-11-02T03:04:05.000Z", sessionId: "5a128faa-8253-489e-b935-6c08e8e670c0", @@ -244,7 +244,7 @@ describe("Claude fallback iterations", () => { }, { type: "fallback_message", - model: "claude-opus-4-8", + model: fallbackModel, input_tokens: 6, cache_read_input_tokens: 0, cache_creation_input_tokens: 10000, @@ -268,14 +268,28 @@ describe("Claude fallback iterations", () => { }, 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, expected: [served] }, + { + 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, { @@ -289,13 +303,20 @@ describe("Claude fallback iterations", () => { reasoningTokens: 0, }, dedupeKey: "msg_fallback:req_fallback:attempt:0", + supersedesDedupeKey: undefined, }, ], }, + ...[undefined, "", ""].map((fallbackModel) => ({ + declined: `mid-stream with unavailable serving model ${String(fallbackModel)}`, + declinedOutputTokens: 250, + fallbackModel, + expected: [unattributed], + })), ])( - "prices the serving fallback under its own model when the request declined $declined", - ({ declinedOutputTokens, expected }) => { - const turn = fallbackTurn(declinedOutputTokens); + "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); diff --git a/apps/server/src/provider/Drivers/claudeUsage.ts b/apps/server/src/provider/Drivers/claudeUsage.ts index 7ec60f3249d2f..1d3f261533134 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.ts +++ b/apps/server/src/provider/Drivers/claudeUsage.ts @@ -61,20 +61,24 @@ function parseClaudeRecord(parsed: unknown): readonly UsageRecord[] { : []; // 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 servingIndex = iterations.findLastIndex( + const fallbackIndex = iterations.findLastIndex( (iteration) => iterationRecord(iteration)?.["type"] === "fallback_message", ); - const serving = iterationRecord(iterations[servingIndex]); + 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: (serving === null ? null : iterationModel(serving)) ?? model, + model: fallbackModel ?? model, sessionId: typeof record["sessionId"] === "string" ? record["sessionId"] : "", totals: usageTotals(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 }), }; // Advisor calls and declined attempts bill at their own model's rates. Anthropic bills a diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 426d32bc02bd4..d8bc01b57620b 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2782,7 +2782,7 @@ describe("UsageService", () => { 60 + 20 * (version === 4 ? 1 : 2), ); yield* upgraded.awaitPersisted; - assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v13.json"); + assert.strictEqual(SCAN_CACHE_FILE_NAME, "usage-scan-cache-v14.json"); assert.deepStrictEqual( (yield* Effect.promise(() => NodeFSP.readdir(stateDir))) .filter((name) => name.startsWith("usage-scan-cache")) diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index 7e4e0ea689cdf..1f8b319e6ec55 100644 --- a/apps/server/src/usage/usageAggregation.test.ts +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it } from "@effect/vitest"; import { resolveModelAliases, UsageAggregator } from "./usageAggregation.ts"; +import { decodeScanCache, dedupeWithinFile, makeScanCacheWriter } from "./usageScanCache.ts"; +import { TEST_FORMAT_MAP } from "./usageTestFormats.ts"; +import { parseClaudeLine } from "../provider/Drivers/claudeUsage.ts"; import type { RateTable } from "./usagePricing.ts"; import type { UsageRecord } from "@t3tools/provider-core/server/usage"; @@ -149,6 +152,102 @@ describe("UsageAggregator", () => { expect(result.buckets).toHaveLength(0); }); + it.each([false, true])( + "replaces a fallback's stream-start copy across cached files (reverse=%s)", + (reverse) => { + const attempt = { + type: "message", + model: "claude-opus-5-5", + input_tokens: 3, + cache_creation_input_tokens: 20000, + output_tokens: 250, + }; + const fallback = { + type: "fallback_message", + model: "claude-fable-5", + input_tokens: 6, + cache_creation_input_tokens: 10000, + output_tokens: 900, + }; + const line = (usage: unknown) => + JSON.stringify({ + type: "assistant", + timestamp: "2026-08-07T04:05:13.944Z", + sessionId: "session-a", + requestId: "req_fallback", + message: { id: "msg_fallback", model: attempt.model, usage }, + }); + const early = parseClaudeLine(line({ ...attempt, output_tokens: 3 })); + const complete = parseClaudeLine( + line({ ...fallback, cache_creation_input_tokens: 30000, iterations: [attempt, fallback] }), + ); + const original = new Map( + [ + ...new Map([ + ["/partial.jsonl", early], + ["/complete.jsonl", dedupeWithinFile([...early, ...complete, ...complete])], + [ + "/deleted.jsonl", + [{ ...early[0]!, totals: { ...early[0]!.totals, cacheCreationTokens: 30000 } }], + ], + ]), + ].map( + ([path, records]) => + [ + path, + { + provider: "claude" as const, + size: 100, + mtimeMs: 100, + records, + tailRecords: [], + malformedRecords: 0, + position: { resumeOffset: 100, guardLength: 0, guardHash: 0, state: null }, + }, + ] as const, + ), + ); + const cache = decodeScanCache( + JSON.parse([...makeScanCacheWriter()(original, {})].join("")), + TEST_FORMAT_MAP, + ); + const copies = [...cache.values()]; + if (reverse) copies.reverse(); + for (const model of [undefined, attempt.model, fallback.model]) { + const aggregator = new UsageAggregator({ + timeZone: "UTC", + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + ...(model === undefined ? {} : { modelFilter: { include: [{ model }] } }), + }); + for (const file of copies) for (const record of file.records) aggregator.add(record); + const buckets = aggregator.finish().buckets; + const expected = + model === undefined + ? [attempt, fallback] + : [model === attempt.model ? attempt : fallback]; + expect( + buckets.map((bucket) => [ + bucket.model, + bucket.totals.cacheCreationTokens, + bucket.totals.outputTokens, + bucket.records, + ]), + ).toEqual( + expected + .toSorted((a, b) => a.model.localeCompare(b.model)) + .map((entry) => [ + entry.model, + entry.cache_creation_input_tokens, + entry.output_tokens, + 1, + ]), + ); + } + }, + ); + it("buckets by the day in the requested time zone", () => { const utc = aggregate([record()], "UTC"); const losAngeles = aggregate([record()], "America/Los_Angeles"); diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts index 0bdf1b8cc2612..d72833390cfa8 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -173,6 +173,7 @@ interface Contribution { export class UsageAggregator { readonly #buckets = new Map(); readonly #seen = new Map(); + readonly #supersededKeys = new Set(); readonly #toDay: (timestampMs: number) => string; readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null; readonly #options: AggregateOptions; @@ -204,6 +205,15 @@ export class UsageAggregator { } add(input: UsageRecord, sourcePath?: string): boolean { + // A completed fallback replaces provisional snapshots even when its serving model is filtered out. + if (input.supersedesDedupeKey !== undefined) { + this.#supersededKeys.add(input.supersedesDedupeKey); + if (this.#seen.delete(input.supersedesDedupeKey)) this.#duplicatesDropped += 1; + } + if (input.dedupeKey !== null && this.#supersededKeys.has(input.dedupeKey)) { + this.#duplicatesDropped += 1; + return false; + } if (!keepsUsageRecord(this.#options.modelFilter, input)) return false; const record = this.#mapModel(input); if (record.dedupeKey !== null) { @@ -344,8 +354,10 @@ export class UsageAggregator { if (bucket !== null) this.#accumulate(bucket, record); } this.#seen.clear(); + this.#supersededKeys.clear(); const buckets: UsageBucket[] = []; for (const [key, bucket] of this.#buckets) { + if (bucket.records === 0) continue; const [day = "", hourStart = "", provider = "", model = "", sourcePath = ""] = key.split("\u0000"); const category = bucket.categoryCostUsd; diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index 9f2fca6fc758b..0ed936053d9a3 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -244,17 +244,20 @@ describe("scan cache round trip", () => { }, ); - it.each([11, 12])("re-parses v%s Claude transcripts for corrected attempt usage", (version) => { - const original = cacheWith([["/claude.jsonl", 100, [record()]]]); - const legacy = { ...parse(makeScanCacheWriter()(original, {})), version }; + it.each([11, 12, 13])( + "re-parses v%s Claude transcripts for corrected attempt usage", + (version) => { + const original = cacheWith([["/claude.jsonl", 100, [record()]]]); + const legacy = { ...parse(makeScanCacheWriter()(original, {})), version }; - const restored = decodeScanCache(JSON.parse(JSON.stringify(legacy)), TEST_FORMAT_MAP); - expect(restored.get("/claude.jsonl")).toEqual({ - ...original.get("/claude.jsonl"), - size: -1, - position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, - }); - }); + const restored = decodeScanCache(JSON.parse(JSON.stringify(legacy)), TEST_FORMAT_MAP); + expect(restored.get("/claude.jsonl")).toEqual({ + ...original.get("/claude.jsonl"), + size: -1, + position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, + }); + }, + ); it("rewrites only changed entries and still restores the whole cache", () => { const write = makeScanCacheWriter(); diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 80d2262e539aa..e6ee651b9950e 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -10,7 +10,7 @@ import { CLAUDE_SYNTHETIC_MODEL } from "../provider/Drivers/claudeUsage.ts"; import { mergeUsageCopies } from "./usageAggregation.ts"; import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts"; -const USAGE_SCAN_CACHE_VERSION = 13 as const; +const USAGE_SCAN_CACHE_VERSION = 14 as const; const SPEED_COMPATIBLE_SINCE_VERSION = 4; /** @@ -18,8 +18,9 @@ const SPEED_COMPATIBLE_SINCE_VERSION = 4; * replace it, dropping saved usage for deleted transcripts. Separate files keep * the newer one; older files are removed once it is saved. */ -export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v13.json"; +export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v14.json"; export const LEGACY_SCAN_CACHE_FILE_NAMES = [ + "usage-scan-cache-v13.json", "usage-scan-cache-v12.json", "usage-scan-cache-v11.json", "usage-scan-cache-v10.json", @@ -60,7 +61,8 @@ type SerializedRecord = readonly [ reportedCostUsd: number | null, speed: number, rateModelIndex?: number | null, - unresolvedServiceTierTurnId?: string, + unresolvedServiceTierTurnId?: string | null, + supersedesDedupeKey?: string, ]; interface SerializedFile { @@ -119,9 +121,11 @@ function serializeFile(entry: CachedFile, tables: InternTables): SerializedFile record.rateModel === undefined ? null : intern(tables.models, tables.modelIndex, record.rateModel), - ...(record.unresolvedServiceTierTurnId === undefined - ? ([] as const) - : ([record.unresolvedServiceTierTurnId] as const)), + ...(record.supersedesDedupeKey !== undefined + ? ([record.unresolvedServiceTierTurnId ?? null, record.supersedesDedupeKey] as const) + : record.unresolvedServiceTierTurnId === undefined + ? ([] as const) + : ([record.unresolvedServiceTierTurnId] as const)), ]; return { s: entry.size, @@ -212,6 +216,7 @@ export function decodeScanCache( speedIndex, rateModelIndex, unresolvedServiceTierTurnId, + supersedesDedupeKey, ] = row as SerializedRecord; const speed = typeof speedIndex === "number" ? SPEEDS[speedIndex] : undefined; @@ -221,8 +226,8 @@ export function decodeScanCache( typeof timestampMs !== "number" || !Number.isFinite(timestampMs) || model === undefined || - (unresolvedServiceTierTurnId !== undefined && - typeof unresolvedServiceTierTurnId !== "string") || + (unresolvedServiceTierTurnId != null && typeof unresolvedServiceTierTurnId !== "string") || + (supersedesDedupeKey !== undefined && typeof supersedesDedupeKey !== "string") || (rateModelIndex != null && rateModel === undefined) || !Number.isFinite(uncached) || !Number.isFinite(cached) || @@ -253,9 +258,10 @@ export function decodeScanCache( }, reportedCostUsd: typeof reportedCostUsd === "number" ? reportedCostUsd : null, speed, - ...(version >= 7 && unresolvedServiceTierTurnId !== undefined + ...(version >= 7 && unresolvedServiceTierTurnId != null ? { unresolvedServiceTierTurnId } : {}), + ...(supersedesDedupeKey === undefined ? {} : { supersedesDedupeKey }), dedupeKey: typeof dedupeKey === "string" ? dedupeKey : null, }); } @@ -292,9 +298,9 @@ export function decodeScanCache( } // Keep history for deleted rollouts, but re-parse incompatible reducer state. // v9 and v11 repair skipped Codex requests; clean v8+ entries stay served while unchanged. - // Claude history before v13 lacks per-attempt fallback usage. + // Claude history before v14 lacks authoritative fallback replacement keys. const legacy = - (entry.p === "claude" && version < 13) || + (entry.p === "claude" && version < 14) || (format.state !== undefined && (version < 6 || (entry.p === "codex" && (version < 8 || (version < 11 && malformedRecords > 0))))); diff --git a/packages/provider-core/src/server/usage.ts b/packages/provider-core/src/server/usage.ts index 3915ce7b2d9fa..928fd25cbea6d 100644 --- a/packages/provider-core/src/server/usage.ts +++ b/packages/provider-core/src/server/usage.ts @@ -22,6 +22,8 @@ export interface UsageRecord { readonly speed: UsageSpeed; readonly unresolvedServiceTierTurnId?: string; readonly dedupeKey: string | null; + /** An authoritative attempt replaces this provisional message, including cached copies. */ + readonly supersedesDedupeKey?: string; } export const EMPTY_TOTALS: UsageTokenTotals = { From d6ef06516f7e53822c4c1be0e600da0a95a63e36 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Wed, 11 Nov 2026 22:21:28 +1100 Subject: [PATCH 3/6] fix(usage): count sessions only for surviving records --- apps/server/src/usage/usageAggregation.test.ts | 12 ++++++++++-- apps/server/src/usage/usageAggregation.ts | 2 +- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index 1f8b319e6ec55..989248e652234 100644 --- a/apps/server/src/usage/usageAggregation.test.ts +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -184,11 +184,17 @@ describe("UsageAggregator", () => { const original = new Map( [ ...new Map([ - ["/partial.jsonl", early], + ["/partial.jsonl", early.map((entry) => ({ ...entry, sessionId: "session-fork" }))], ["/complete.jsonl", dedupeWithinFile([...early, ...complete, ...complete])], [ "/deleted.jsonl", - [{ ...early[0]!, totals: { ...early[0]!.totals, cacheCreationTokens: 30000 } }], + [ + { + ...early[0]!, + sessionId: "session-deleted", + totals: { ...early[0]!.totals, cacheCreationTokens: 30000 }, + }, + ], ], ]), ].map( @@ -233,6 +239,7 @@ describe("UsageAggregator", () => { bucket.totals.cacheCreationTokens, bucket.totals.outputTokens, bucket.records, + bucket.sessions, ]), ).toEqual( expected @@ -242,6 +249,7 @@ describe("UsageAggregator", () => { entry.cache_creation_input_tokens, entry.output_tokens, 1, + 1, ]), ); } diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts index d72833390cfa8..cce86f8ab2dfd 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -233,7 +233,6 @@ export class UsageAggregator { } // Keyed usage is counted in finish(), once all its copies are merged. if (record.dedupeKey === null) this.#accumulate(bucket, record); - if (record.sessionId.length > 0) bucket.sessions.add(record.sessionId); return true; } @@ -262,6 +261,7 @@ export class UsageAggregator { } #accumulate(bucket: MutableBucket, record: UsageRecord): void { + if (record.sessionId.length > 0) bucket.sessions.add(record.sessionId); const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides); const totals = bucket.totals; From 66298d2fa9224ce23806ff3757509137b2c1ecbe Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 13 Nov 2026 12:25:40 +1100 Subject: [PATCH 4/6] test(server): capture usage-limit observer cursor before forking Use the persisted-event cursor pattern already used by the sibling cancellation test. Co-Authored-By: Claude Opus 5.5 --- .../TaskCancelUsageLimit.integration.test.ts | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts index 2812259af3379..0c6d93f903e7c 100644 --- a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts +++ b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts @@ -24,6 +24,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 EffectWorker from "../orchestration-v2/EffectWorker.ts"; +import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -199,17 +200,25 @@ 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"); + // Read the cursor before forking: provider events offered right after a watch can + // commit before a live-tail subscriber has read its starting sequence. 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( From 4ea94433f6232fe618165c94aa825eb7c98e8b24 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 02:49:03 +0000 Subject: [PATCH 5/6] style(server): format usage merge resolution --- apps/server/src/usage/UsageService.ts | 1 - apps/server/src/usage/usageScanCache.ts | 2 -- 2 files changed, 3 deletions(-) diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 433dcb54a9c67..8f98e4a96cda1 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -908,7 +908,6 @@ export const make = Effect.gen(function* () { const modelAliases = resolveModelAliases(settings.usageModelAliases); const sources: Array> = []; - // Deleted transcripts' saved usage still aggregates and deduplicates normally. const retainedSinceMs = Math.max(windowStartMs, retentionCutoffMs); // Saved paths are files below a scanned root, so their folder decides; folders repeat thousands of times. diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index c20c8bddde541..2981d9b0b1d9c 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -286,7 +286,6 @@ export function decodeScanCache( !dedupeKey.endsWith(`:${timestampMs}:${model}`) ? `${dedupeKey.slice(0, -model.length)}${timestampMs}:${model}` : dedupeKey, - }); } return records; @@ -328,7 +327,6 @@ export function decodeScanCache( const legacy = (entry.p === "claude" && version < 18) || (entry.p === "grok" && version < 17) || - (format.state !== undefined && (version < 6 || (entry.p === "codex" && (version < 14 || (version < 15 && malformedRecords > 0))))); From c095ec24419b4a75c2dd14ce216e55eb1a557099 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 02:51:13 +0000 Subject: [PATCH 6/6] style(server): keep cache migration comment within lint limit --- apps/server/src/usage/usageScanCache.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 2981d9b0b1d9c..bbcebc296a882 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -320,8 +320,7 @@ export function decodeScanCache( continue; } // Keep history for deleted rollouts, but re-parse incompatible reducer state. - // v13 adds Claude 1-hour cache writes; v14 uses Codex request ledgers. - // v15 repairs skipped Codex requests whose counters rewound; clean v14 entries stay served. + // v14 uses Codex request ledgers; v15 repairs rewinds while serving clean v14 entries. // v17 keys Grok turns by time, restoring turns previously dropped as repeats. // v18 adds authoritative Claude fallback replacement keys. const legacy =