diff --git a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts b/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts index c3e549d14dd75..43593a54ead6a 100644 --- a/apps/server/src/mcp/TaskCancelUsageLimit.integration.test.ts +++ b/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( diff --git a/apps/server/src/provider/Drivers/claudeUsage.test.ts b/apps/server/src/provider/Drivers/claudeUsage.test.ts index bc10ba5b7b29a..661899cef59c0 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.test.ts +++ b/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, "", ""].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"; diff --git a/apps/server/src/provider/Drivers/claudeUsage.ts b/apps/server/src/provider/Drivers/claudeUsage.ts index 713b35884292b..2b6fe4f98f477 100644 --- a/apps/server/src/provider/Drivers/claudeUsage.ts +++ b/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; - 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 | 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 d40beb989ec21..038aa19ea6b1f 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/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", () => diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index d84de15a59ed9..8f98e4a96cda1 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -906,7 +906,7 @@ export const make = Effect.gen(function* () { const priceOverrides = createOverrideRateTable(settings.usagePriceOverrides); const modelAliases = resolveModelAliases(settings.usageModelAliases); - const sources: UsageSource[] = []; + const sources: Array> = []; // Deleted transcripts' saved usage still aggregates and deduplicates normally. const retainedSinceMs = Math.max(windowStartMs, retentionCutoffMs); @@ -1063,9 +1063,6 @@ export const make = Effect.gen(function* () { let scannedFiles = 0; let skippedFiles = 0; let malformedRecords = 0; - // Distinct per directory. Buckets carry per-cell session counts, but a - // session spans days and models, so clients total this figure instead. - const sessionIds = new Set(); let counted = false; for (const file of filesByDir[index] ?? []) { @@ -1095,10 +1092,7 @@ export const make = Effect.gen(function* () { eventOccurrences.set(key, occurrence); usageRecord = { ...record, dedupeKey: key + ":" + occurrence }; } - if (aggregator.add(usageRecord, dir)) { - counted = true; - if (record.sessionId.length > 0) sessionIds.add(record.sessionId); - } + if (aggregator.add(usageRecord, dir)) counted = true; } } @@ -1123,7 +1117,6 @@ export const make = Effect.gen(function* () { scannedFiles, skippedFiles, malformedRecords, - distinctSessions: sessionIds.size, message: malformedRecords > 0 ? [sourceMessage, formatMalformedUsageWarning(malformedRecords)] @@ -1158,8 +1151,15 @@ export const make = Effect.gen(function* () { sinceDay: input.sinceDay, untilDay: input.untilDay, buckets: aggregated.buckets, - sources, + sources: sources.map((source) => ({ + ...source, + distinctSessions: + aggregated.sourceSessionCounts + .get(source.fingerprint.provider) + ?.get(source.fingerprint.resolvedHomePath) ?? 0, + })), pricing: summaryPricing, + scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), } satisfies UsageSummary; diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index 3819903606be2..8a632ec874555 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"; @@ -178,6 +181,110 @@ 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.map((entry) => ({ ...entry, sessionId: "session-fork" }))], + ["/complete.jsonl", dedupeWithinFile([...early, ...complete, ...complete])], + [ + "/deleted.jsonl", + [ + { + ...early[0]!, + sessionId: "session-deleted", + 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, + bucket.sessions, + ]), + ).toEqual( + expected + .toSorted((a, b) => a.model.localeCompare(b.model)) + .map((entry) => [ + entry.model, + entry.cache_creation_input_tokens, + entry.output_tokens, + 1, + 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 ce7a78a99d9c1..8596f8d8805ae 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -137,6 +137,7 @@ export function canonicalModelFilter( export interface AggregateResult { readonly buckets: readonly UsageBucket[]; + readonly sourceSessionCounts: ReadonlyMap>; readonly duplicatesDropped: number; readonly outOfWindow: number; } @@ -215,6 +216,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; @@ -246,6 +248,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) { @@ -270,7 +281,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; } @@ -299,6 +309,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; @@ -399,10 +410,24 @@ export class UsageAggregator { if (bucket !== null) this.#accumulate(bucket, record); } this.#seen.clear(); + this.#supersededKeys.clear(); const buckets: UsageBucket[] = []; + const sourceSessions = new Map>>(); for (const [key, bucket] of this.#buckets) { + if (bucket.records === 0) continue; const [day = "", hourStart = "", provider = "", model = "", sourcePath = ""] = key.split("\u0000"); + let providerSources = sourceSessions.get(provider); + if (providerSources === undefined) { + providerSources = new Map(); + sourceSessions.set(provider, providerSources); + } + let sessions = providerSources.get(sourcePath); + if (sessions === undefined) { + sessions = new Set(); + providerSources.set(sourcePath, sessions); + } + for (const sessionId of bucket.sessions) sessions.add(sessionId); const category = bucket.categoryCostUsd; const fastCostUsd = roundUsd(bucket.fastCostUsd); const ultrafastCostUsd = roundUsd(bucket.ultrafastCostUsd); @@ -446,6 +471,12 @@ export class UsageAggregator { return { buckets, + sourceSessionCounts: new Map( + [...sourceSessions].map(([provider, paths]) => [ + provider, + new Map([...paths].map(([path, sessions]) => [path, sessions.size])), + ]), + ), duplicatesDropped: this.#duplicatesDropped, outOfWindow: this.#outOfWindow, }; diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index 36864768eb525..a8181d01a1d47 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -68,7 +68,8 @@ describe("scan cache round trip", () => { [ record(), record({ - dedupeKey: "msg_2:", + dedupeKey: "msg_2::fallback", + supersedesDedupeKey: "msg_2:", model: "claude-opus-5-5", speed: "fast", cacheCreation1hTokens: 8, @@ -168,7 +169,7 @@ describe("scan cache round trip", () => { expect(restored).toEqual(original); }); - it.each([5, 6, 7, 8, 10, 11, 12, 13])( + it.each([5, 6, 7, 8, 10, 11, 12, 13, 14, 15, 16])( "retains v%s history and compatible positions but invalidates obsolete state", (version) => { const original = cacheWith([["/gone.jsonl", 100, [record({ speed: "fast" })]]]); @@ -225,25 +226,29 @@ describe("scan cache round trip", () => { }; const restored = decodeScanCache(JSON.parse(JSON.stringify(legacy)), TEST_FORMAT_MAP); - expect(restored.get("/gone.jsonl")).toEqual( - version >= 13 - ? original.get("/gone.jsonl") + expect(restored.get("/gone.jsonl")).toEqual({ + ...original.get("/gone.jsonl"), + size: -1, + position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, + }); + expect(restored.get("/gone-codex.jsonl")).toEqual( + version >= 16 + ? original.get("/gone-codex.jsonl") : { - ...original.get("/gone.jsonl"), - size: -1, + ...original.get("/gone-codex.jsonl"), + size: version === 15 ? original.get("/gone-codex.jsonl")!.size : -1, + position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, + }, + ); + expect(restored.get("/clean-codex.jsonl")).toEqual( + version >= 16 + ? cleanCodex + : { + ...cleanCodex, + size: version >= 14 ? cleanCodex.size : -1, position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, }, ); - expect(restored.get("/gone-codex.jsonl")).toEqual({ - ...original.get("/gone-codex.jsonl"), - size: -1, - position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, - }); - expect(restored.get("/clean-codex.jsonl")).toEqual({ - ...cleanCodex, - size: -1, - position: { resumeOffset: 0, guardLength: 0, guardHash: 0, state: null }, - }); expect(restored.get("/gone-pi.jsonl")).toEqual( version >= 6 ? original.get("/gone-pi.jsonl") @@ -327,17 +332,20 @@ describe("scan cache round trip", () => { expect(decodeScanCache(current, TEST_FORMAT_MAP).get("/codex.jsonl")).toEqual(entry); }); - it("re-parses a Claude transcript cached before 1-hour cache writes were counted", () => { - const original = cacheWith([["/claude.jsonl", 100, [record()]]]); - const v12 = { ...parse(makeScanCacheWriter()(original, {})), version: 12 }; + it.each([11, 12, 13, 14, 15, 16, 17])( + "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(v12)), 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.each(["s:p:grok-4.5", "s:p:1786000000000:grok-4.5"])( "re-parses v16 Grok transcripts with key %s", diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 843b414b97f32..bbcebc296a882 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 = 17 as const; +const USAGE_SCAN_CACHE_VERSION = 18 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-v17.json"; +export const SCAN_CACHE_FILE_NAME = "usage-scan-cache-v18.json"; export const LEGACY_SCAN_CACHE_FILE_NAMES = [ + "usage-scan-cache-v17.json", "usage-scan-cache-v16.json", "usage-scan-cache-v15.json", "usage-scan-cache-v14.json", @@ -66,6 +67,7 @@ type SerializedRecord = readonly [ rateModelIndex?: number | null, unresolvedServiceTierTurnId?: string | null, cacheCreation1hTokens?: number, + supersedesDedupeKey?: string, ]; interface SerializedFile { @@ -124,11 +126,17 @@ function serializeFile(entry: CachedFile, tables: InternTables): SerializedFile record.rateModel === undefined ? null : intern(tables.models, tables.modelIndex, record.rateModel), - ...(record.cacheCreation1hTokens !== undefined - ? ([record.unresolvedServiceTierTurnId ?? null, record.cacheCreation1hTokens] as const) - : record.unresolvedServiceTierTurnId !== undefined - ? ([record.unresolvedServiceTierTurnId] as const) - : ([] as const)), + ...(record.supersedesDedupeKey !== undefined + ? ([ + record.unresolvedServiceTierTurnId ?? null, + record.cacheCreation1hTokens ?? 0, + record.supersedesDedupeKey, + ] as const) + : record.cacheCreation1hTokens !== undefined + ? ([record.unresolvedServiceTierTurnId ?? null, record.cacheCreation1hTokens] as const) + : record.unresolvedServiceTierTurnId !== undefined + ? ([record.unresolvedServiceTierTurnId] as const) + : ([] as const)), ]; return { s: entry.size, @@ -220,6 +228,7 @@ export function decodeScanCache( rateModelIndex, unresolvedServiceTierTurnId, cacheCreation1h = 0, + supersedesDedupeKey, ] = row as SerializedRecord; const speed = typeof speedIndex === "number" ? SPEEDS[speedIndex] : undefined; @@ -231,6 +240,7 @@ export function decodeScanCache( model === undefined || (unresolvedServiceTierTurnId != null && typeof unresolvedServiceTierTurnId !== "string") || !Number.isFinite(cacheCreation1h) || + (supersedesDedupeKey !== undefined && typeof supersedesDedupeKey !== "string") || (rateModelIndex != null && rateModel === undefined) || !Number.isFinite(uncached) || !Number.isFinite(cached) || @@ -265,6 +275,7 @@ export function decodeScanCache( ...(version >= 7 && unresolvedServiceTierTurnId != null ? { unresolvedServiceTierTurnId } : {}), + ...(supersedesDedupeKey === undefined ? {} : { supersedesDedupeKey }), dedupeKey: typeof dedupeKey !== "string" ? null @@ -309,11 +320,11 @@ 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 = - (entry.p === "claude" && version < 13) || + (entry.p === "claude" && version < 18) || (entry.p === "grok" && version < 17) || (format.state !== undefined && (version < 6 || diff --git a/packages/provider-core/src/server/usage.ts b/packages/provider-core/src/server/usage.ts index c2c75a641e53a..d3efe606c7929 100644 --- a/packages/provider-core/src/server/usage.ts +++ b/packages/provider-core/src/server/usage.ts @@ -24,6 +24,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 = {