diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index 2aa057ee0ca..4ad28691a4f 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -45,6 +45,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverGetProcessResourceHistory]: AuthOrchestrationReadScope, [WS_METHODS.serverGetResourceTelemetryHistory]: AuthOrchestrationReadScope, [WS_METHODS.serverRetryResourceTelemetry]: AuthOrchestrationOperateScope, + [WS_METHODS.serverGetUsageSummary]: AuthOrchestrationReadScope, [WS_METHODS.serverSignalProcess]: AuthOrchestrationOperateScope, [WS_METHODS.serverReportClientActivity]: AuthOrchestrationReadScope, [WS_METHODS.serverReportHostPowerState]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 4ddb01e09dd..d982c2e192c 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -149,6 +149,7 @@ import * as DesktopTelemetryReceiver from "./resourceTelemetry/DesktopTelemetryR import * as NativeTelemetryClient from "./resourceTelemetry/NativeTelemetryClient.ts"; import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; +import * as UsageService from "./usage/UsageService.ts"; import * as Data from "effect/Data"; import { makeOrchestrationIntegrationHarness } from "../integration/OrchestrationEngineHarness.integration.ts"; @@ -825,6 +826,7 @@ const buildAppUnderTest = (options?: { const appLayer = servedRoutesLayer.pipe( Layer.provide(resourceTelemetryLayer), + Layer.provide(UsageService.layerTest), Layer.provide( Layer.mock(BrowserTraceCollector.BrowserTraceCollector)({ record: () => Effect.void, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 1d824afbd1b..ff21c07a861 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -101,6 +101,7 @@ import * as NativeTelemetryClient from "./resourceTelemetry/NativeTelemetryClien import * as ResourceAttribution from "./resourceTelemetry/ResourceAttribution.ts"; import * as ResourceMonitorBinary from "./resourceTelemetry/ResourceMonitorBinary.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; +import * as UsageService from "./usage/UsageService.ts"; import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts"; import { clearPersistedServerRuntimeState, @@ -158,6 +159,8 @@ const BackgroundLayerLive = BackgroundPolicy.layer.pipe( Layer.provideMerge(ServerSettingsLayerLive), ); +const UsageLayerLive = UsageService.layer.pipe(Layer.provide(ServerSettingsLayerLive)); + const ResourceDiagnosticsLayerLive = Layer.mergeAll( ResourceTelemetryLayerLive, ProcessDiagnostics.layer.pipe(Layer.provide(ResourceTelemetryLayerLive)), @@ -410,6 +413,7 @@ const RuntimeDependenciesLive = RuntimeCoreDependenciesLive.pipe( // Misc. Layer.provideMerge(BackgroundLayerLive), Layer.provideMerge(ResourceDiagnosticsLayerLive), + Layer.provideMerge(UsageLayerLive), Layer.provideMerge(TraceDiagnostics.layer), Layer.provideMerge(AnalyticsService.layer), Layer.provideMerge(ExternalLauncher.layer), diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts new file mode 100644 index 00000000000..2ad2a729ecb --- /dev/null +++ b/apps/server/src/usage/UsageService.ts @@ -0,0 +1,420 @@ +/** + * UsageService - scans provider transcripts and returns priced daily usage. + * + * The scan reads the provider CLIs' own session files rather than T3 Code's + * orchestration projections, so usage covers turns driven outside T3 Code too. + * This is the approach `ccusage` takes. + * + * Transcripts are append-only, so parsed records are memoised per file by + * `(size, mtime)`. A cold 30-day scan of ~1.4 GB lands around 2-3 seconds; warm + * scans only reparse files that changed. + * + * @module UsageService + */ +import * as NodeOS from "node:os"; + +import { + USAGE_CONTRACT_VERSION, + type UsageProviderKind, + type UsageSource, + type UsageSummary, + type UsageSummaryInput, + UsageReadError, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import * as Schema from "effect/Schema"; +import { HttpClient, HttpClientResponse } from "effect/unstable/http"; + +import { ServerConfig } from "../config.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import { resolveClaudeHomePath } from "../provider/Drivers/ClaudeHome.ts"; +import { resolveCodexHomeLayout } from "../provider/Drivers/CodexHomeLayout.ts"; +import { UsageAggregator } from "./usageAggregation.ts"; +import { parseRateTable, type RateTable } from "./usagePricing.ts"; +import { + listTranscriptFiles, + readDirectoryVolumeId, + readTranscriptRecords, +} from "./usageTranscriptReader.ts"; +import { + decodeScanCache, + dedupeWithinFile, + encodeScanCache, + pruneScanCache, + type ScanCache, +} from "./usageScanCache.ts"; +import type { UsageRecord } from "./usageTranscripts.ts"; + +const LITELLM_RATES_URL = + "https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json"; + +/** Rates move rarely; a day-old table keeps the page working offline. */ +const RATES_TTL_MS = 24 * 60 * 60 * 1000; + +/** + * Files are filtered by mtime before opening. The slack covers a session whose + * last write lands just before local midnight on the window's first day. + */ +const MTIME_SLACK_MS = 36 * 60 * 60 * 1000; + +/** Longest window the UI offers, plus slack. Older entries are pruned. */ +const CACHE_RETENTION_DAYS = 90; + +/** On-disk shape of the rate snapshot. */ +const RatesCacheFile = Schema.Struct({ + fetchedAtMs: Schema.Number, + document: Schema.Unknown, +}); +const decodeRatesCache = Schema.decodeUnknownEffect( + Schema.fromJsonString(RatesCacheFile as unknown as Schema.Codec), +); +const encodeRatesCache = Schema.encodeEffect( + Schema.fromJsonString(RatesCacheFile as unknown as Schema.Codec), +); + +/** The scan cache is narrowed by hand in `usageScanCache`, so JSON is enough here. */ +const ScanCacheJson = Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec); +const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson); +const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson); + +export class UsageService extends Context.Service< + UsageService, + { + readonly readSummary: (input: UsageSummaryInput) => Effect.Effect; + } +>()("t3/usage/UsageService") {} + +/** Empty summary, for suites that only need the RPC surface to resolve. */ +export const layerTest = Layer.succeed( + UsageService, + UsageService.of({ + readSummary: (input) => + Effect.succeed({ + contractVersion: USAGE_CONTRACT_VERSION, + readAt: "1970-01-01T00:00:00.000Z", + timeZone: input.timeZone, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + buckets: [], + sources: [], + pricing: { + status: "unavailable", + source: LITELLM_RATES_URL, + fetchedAt: null, + knownModels: 0, + }, + scanDurationMs: 0, + }), + }), +); + +export const make = Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const config = yield* ServerConfig; + const settingsService = yield* ServerSettings.ServerSettingsService; + const httpClient = yield* HttpClient.HttpClient; + + const fileCache: ScanCache = new Map(); + let cacheDirty = false; + + const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json"); + const scanCachePath = path.join(config.stateDir, "usage-scan-cache.json"); + let rates: RateTable = new Map(); + let ratesFetchedAtMs: number | null = null; + let ratesStatus: UsageSummary["pricing"]["status"] = "unavailable"; + + /** + * Loads the LiteLLM rate table, preferring a fresh copy and falling back to + * the on-disk snapshot. With neither, every model reports as unpriced rather + * than the page failing. + */ + const ensureRates = Effect.fn("UsageService.ensureRates")(function* () { + const now = yield* Clock.currentTimeMillis; + if (ratesFetchedAtMs !== null && now - ratesFetchedAtMs < RATES_TTL_MS) return; + + if (ratesFetchedAtMs === null) { + const fromDisk = yield* fileSystem.readFileString(ratesCachePath).pipe( + Effect.flatMap((raw) => decodeRatesCache(raw)), + Effect.catchCause(() => Effect.succeed(null)), + ); + if (fromDisk !== null) { + const parsed = parseRateTable(fromDisk.document); + if (parsed.size > 0) { + rates = parsed; + ratesFetchedAtMs = fromDisk.fetchedAtMs; + ratesStatus = "cached"; + if (now - fromDisk.fetchedAtMs < RATES_TTL_MS) return; + } + } + } + + const fetched = yield* httpClient.get(LITELLM_RATES_URL).pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap((response) => response.json), + Effect.timeout(10_000), + Effect.catchCause(() => Effect.succeed(null)), + ); + if (fetched === null) { + // The refresh failed; whatever we are serving is now past its TTL and + // must not keep claiming to be fresh. + if (rates.size > 0) ratesStatus = "cached"; + return; + } + + const parsed = parseRateTable(fetched); + if (parsed.size === 0) return; + + rates = parsed; + ratesFetchedAtMs = now; + ratesStatus = "fresh"; + + yield* encodeRatesCache({ fetchedAtMs: now, document: fetched }).pipe( + Effect.flatMap((serialized) => fileSystem.writeFileString(ratesCachePath, serialized)), + Effect.catchCause(() => Effect.void), + ); + }); + + /** + * Claude's config dir is the home itself when overridden, but a default + * install nests transcripts under `~/.claude/projects`. Probe both. + */ + const resolveClaudeTranscriptDir = (homePath: string) => + Effect.gen(function* () { + const nested = path.join(homePath, ".claude", "projects"); + const nestedExists = yield* fileSystem + .exists(nested) + .pipe(Effect.catchCause(() => Effect.succeed(false))); + return nestedExists ? nested : path.join(homePath, "projects"); + }); + + /** Resolves the transcript directory for each provider. */ + const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* () { + // A settings failure must surface as an error: swallowing it here would + // present "zero usage from every provider" as a valid answer. + const settings = yield* settingsService.getSettings.pipe( + Effect.catchCause( + (cause) => + new UsageReadError({ + reason: "scanFailed", + // Bounded description; the squashed failure travels as the cause. + // Squashed, not the Cause tree: a full tree in a Defect field is + // the unbounded wire payload the bounded detail exists to avoid. + detail: "Server settings could not be read.", + cause: Cause.squash(cause), + }), + ), + ); + + const claudeHome = yield* resolveClaudeHomePath(settings.providers.claudeAgent); + const claudeDir = yield* resolveClaudeTranscriptDir(claudeHome); + const codexLayout = yield* resolveCodexHomeLayout(settings.providers.codex); + + return [ + { provider: "claude" as const, dir: claudeDir }, + { provider: "codex" as const, dir: path.join(codexLayout.sharedHomePath, "sessions") }, + ]; + }); + + /** + * Loads the persisted scan cache exactly once per process. + * + * `Effect.cached` makes concurrent first readers await the same load rather + * than each seeing a "loaded" flag set before the read finished and cold + * scanning against an empty cache. + */ + const ensureScanCacheLoaded = yield* Effect.cached( + Effect.gen(function* () { + const document = yield* fileSystem.readFileString(scanCachePath).pipe( + Effect.flatMap((raw) => decodeScanCacheFile(raw)), + Effect.catchCause(() => Effect.succeed(null)), + ); + if (document === null) return; + for (const [path, entry] of decodeScanCache(document)) fileCache.set(path, entry); + }), + ); + + const persistScanCache = Effect.fn("UsageService.persistScanCache")(function* () { + if (!cacheDirty) return; + // Cleared only after the write lands, so a failed persist is retried on + // the next scan instead of leaving disk permanently stale. + yield* encodeScanCacheFile(encodeScanCache(fileCache)).pipe( + Effect.flatMap((serialized) => fileSystem.writeFileString(scanCachePath, serialized)), + Effect.map(() => { + cacheDirty = false; + }), + // A cache we cannot write is a slower next start, not a failed read. + Effect.catchCause(() => Effect.void), + ); + }); + + /** Parses one transcript, reusing the cached result when it is unchanged. */ + const readFileRecords = ( + filePath: string, + size: number, + mtimeMs: number, + provider: UsageProviderKind, + ): Effect.Effect => + Effect.gen(function* () { + const cached = fileCache.get(filePath); + // Provider is part of the identity: if both providers were ever pointed + // at one directory, a hit parsed by the other parser must not be reused. + if ( + cached && + cached.size === size && + cached.mtimeMs === mtimeMs && + cached.provider === provider + ) { + return cached.records; + } + + const parsed = yield* Effect.promise(() => readTranscriptRecords(filePath, provider)); + // A read failure is not an empty transcript: caching it under this + // (size, mtime) would silently drop the file's usage until it changes. + if (parsed === null) return []; + // Stored already de-duplicated within the file, which is 99% of all + // duplicates. The aggregator still runs the cross-file dedupe pass. + const records = dedupeWithinFile(parsed); + + fileCache.set(filePath, { size, mtimeMs, provider, records }); + cacheDirty = true; + return records; + }); + + const readSummary = Effect.fn("UsageService.readSummary")(function* (input: UsageSummaryInput) { + if (input.sinceDay > input.untilDay) { + return yield* new UsageReadError({ + reason: "invalidWindow", + detail: `sinceDay '${input.sinceDay}' is after untilDay '${input.untilDay}'`, + }); + } + + const startedAtMs = yield* Clock.currentTimeMillis; + yield* ensureRates(); + yield* ensureScanCacheLoaded; + + const hostId = NodeOS.hostname(); + // The home resolvers ask for `Path` themselves; satisfy them from the + // instance we already hold so `readSummary` stays context-free. + const dirs = yield* resolveTranscriptDirs().pipe(Effect.provideService(Path.Path, path)); + const windowStart = DateTime.make(`${input.sinceDay}T00:00:00Z`); + if (Option.isNone(windowStart)) { + return yield* new UsageReadError({ + reason: "invalidWindow", + detail: `sinceDay '${input.sinceDay}' is not a valid date`, + }); + } + const windowStartMs = DateTime.toEpochMillis(windowStart.value) - MTIME_SLACK_MS; + + const aggregator = new UsageAggregator({ + timeZone: input.timeZone, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + rates, + }); + + const sources: UsageSource[] = []; + const livePaths = new Set(); + const walkedRoots: string[] = []; + + for (const { provider, dir } of dirs) { + const volumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir)); + const exists = yield* fileSystem + .exists(dir) + .pipe(Effect.catchCause(() => Effect.succeed(false))); + + if (!exists) { + sources.push({ + fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId }, + status: "missing", + scannedFiles: 0, + skippedFiles: 0, + malformedRecords: 0, + distinctSessions: 0, + message: "No transcript directory on this environment.", + }); + continue; + } + + walkedRoots.push(dir); + const files = yield* Effect.promise(() => listTranscriptFiles(dir, windowStartMs)); + let scannedFiles = 0; + let skippedFiles = 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(); + + for (const file of files) { + livePaths.add(file.path); + const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider); + if (records.length === 0) { + skippedFiles += 1; + continue; + } + scannedFiles += 1; + for (const record of records) { + // Only sessions that contributed in-window count: the mtime slack + // admits boundary files whose records fall outside the range. + if (aggregator.add(record) && record.sessionId.length > 0) { + sessionIds.add(record.sessionId); + } + } + } + + sources.push({ + fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId }, + status: "ok", + scannedFiles, + skippedFiles, + malformedRecords: 0, + distinctSessions: sessionIds.size, + message: null, + }); + } + + const pruned = pruneScanCache(fileCache, { + livePaths, + walkedRoots, + windowStartMs, + retentionCutoffMs: startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000, + }); + if (pruned > 0) cacheDirty = true; + yield* persistScanCache(); + + const aggregated = aggregator.finish(); + const readAt = yield* DateTime.now; + const finishedAtMs = yield* Clock.currentTimeMillis; + + return { + contractVersion: USAGE_CONTRACT_VERSION, + readAt: DateTime.formatIso(readAt), + timeZone: input.timeZone, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + buckets: aggregated.buckets, + sources, + pricing: { + status: ratesStatus, + source: LITELLM_RATES_URL, + fetchedAt: + ratesFetchedAtMs === null + ? null + : DateTime.formatIso(DateTime.makeUnsafe(ratesFetchedAtMs)), + knownModels: rates.size, + }, + scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), + } satisfies UsageSummary; + }); + + return { readSummary } as const; +}); + +export const layer = Layer.effect(UsageService, make); diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts new file mode 100644 index 00000000000..9117e216f12 --- /dev/null +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -0,0 +1,132 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { UsageAggregator } from "./usageAggregation.ts"; +import type { RateTable } from "./usagePricing.ts"; +import type { UsageRecord } from "./usageTranscripts.ts"; + +const rates: RateTable = new Map([ + [ + "claude-fable-5", + { + inputCostPerToken: 1e-5, + outputCostPerToken: 5e-5, + cacheReadCostPerToken: 1e-6, + cacheCreationCostPerToken: 1.25e-5, + }, + ], +]); + +function record(overrides: Partial = {}): UsageRecord { + return { + provider: "claude", + // 2026-08-07T04:05Z is still Aug 6 in Los Angeles. + timestampMs: Date.parse("2026-08-07T04:05:13.944Z"), + model: "claude-fable-5", + sessionId: "session-a", + totals: { + uncachedInputTokens: 100, + cachedInputTokens: 1000, + cacheCreationTokens: 10, + outputTokens: 50, + reasoningTokens: 0, + }, + reportedCostUsd: null, + dedupeKey: null, + ...overrides, + }; +} + +function aggregate(records: readonly UsageRecord[], timeZone = "UTC") { + const aggregator = new UsageAggregator({ + timeZone, + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + }); + for (const item of records) aggregator.add(item); + return aggregator.finish(); +} + +describe("UsageAggregator", () => { + it("keeps only the first record for a repeated dedupe key", () => { + const result = aggregate([ + record({ dedupeKey: "msg_1:" }), + record({ dedupeKey: "msg_1:" }), + record({ dedupeKey: "msg_1:" }), + ]); + + expect(result.duplicatesDropped).toBe(2); + expect(result.buckets).toHaveLength(1); + expect(result.buckets[0]?.records).toBe(1); + expect(result.buckets[0]?.totals.outputTokens).toBe(50); + }); + + it("still sums records that carry no dedupe key", () => { + const result = aggregate([record(), record()]); + + expect(result.duplicatesDropped).toBe(0); + expect(result.buckets[0]?.totals.outputTokens).toBe(100); + }); + + it("buckets by the day in the requested time zone", () => { + const utc = aggregate([record()], "UTC"); + const losAngeles = aggregate([record()], "America/Los_Angeles"); + + expect(utc.buckets[0]?.day).toBe("2026-08-07"); + expect(losAngeles.buckets[0]?.day).toBe("2026-08-06"); + }); + + it("prices against the rate table", () => { + const result = aggregate([record()]); + + // 100*1e-5 + 1000*1e-6 + 10*1.25e-5 + 50*5e-5 + expect(result.buckets[0]?.costUsd).toBeCloseTo(0.004625, 9); + expect(result.buckets[0]?.costSource).toBe("modelPriced"); + }); + + it("counts tokens but not cost for a model with no rate", () => { + const result = aggregate([record({ model: "kimi-k3" })]); + + expect(result.buckets[0]?.costUsd).toBe(0); + expect(result.buckets[0]?.costSource).toBe("unpriced"); + expect(result.buckets[0]?.unpricedRecords).toBe(1); + expect(result.buckets[0]?.totals.outputTokens).toBe(50); + }); + + it("prefers a reported cost over the rate table", () => { + const result = aggregate([record({ reportedCostUsd: 1.25 })]); + + expect(result.buckets[0]?.costUsd).toBe(1.25); + expect(result.buckets[0]?.costSource).toBe("providerReported"); + }); + + it("drops records outside the window", () => { + const result = aggregate([record({ timestampMs: Date.parse("2026-07-01T12:00:00Z") })]); + + expect(result.outOfWindow).toBe(1); + expect(result.buckets).toHaveLength(0); + }); + + it("reports whether a record contributed", () => { + const aggregator = new UsageAggregator({ + timeZone: "UTC", + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + }); + + expect(aggregator.add(record({ dedupeKey: "msg_1:" }))).toBe(true); + expect(aggregator.add(record({ dedupeKey: "msg_1:" }))).toBe(false); + expect(aggregator.add(record({ timestampMs: Date.parse("2026-07-01T12:00:00Z") }))).toBe(false); + }); + + it("separates providers and models into their own buckets", () => { + const result = aggregate([ + record(), + record({ provider: "codex", model: "gpt-5.6-sol" }), + record({ model: "claude-opus-5" }), + ]); + + expect(result.buckets).toHaveLength(3); + }); +}); diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts new file mode 100644 index 00000000000..4f04a318c52 Binary files /dev/null and b/apps/server/src/usage/usageAggregation.ts differ diff --git a/apps/server/src/usage/usagePricing.ts b/apps/server/src/usage/usagePricing.ts new file mode 100644 index 00000000000..f0e59a87439 --- /dev/null +++ b/apps/server/src/usage/usagePricing.ts @@ -0,0 +1,148 @@ +/** + * Model rate lookup and cost arithmetic. + * + * Rates come from LiteLLM's `model_prices_and_context_window.json`, the same + * table `ccusage` prices against. Everything here is pure: fetching and caching + * the table lives in `UsageService`. + * + * @module usagePricing + */ +import type { UsageCostSource, UsageTokenTotals } from "@t3tools/contracts"; + +/** + * The subset of a LiteLLM entry we price against. All values are USD per token. + * + * LiteLLM also publishes tiered variants (`*_above_272k_tokens`, `*_flex`, + * `*_priority`, `*_batches`). We deliberately price at the base tier: the + * transcripts don't record which tier served a request, so anything else would + * be a guess dressed up as precision. + */ +export interface ModelRate { + readonly inputCostPerToken: number; + readonly outputCostPerToken: number; + readonly cacheReadCostPerToken: number; + readonly cacheCreationCostPerToken: number; +} + +export type RateTable = ReadonlyMap; + +/** Raw shape of one LiteLLM entry, narrowed to the fields we read. */ +interface LiteLlmEntry { + readonly input_cost_per_token?: unknown; + readonly output_cost_per_token?: unknown; + readonly cache_read_input_token_cost?: unknown; + readonly cache_creation_input_token_cost?: unknown; +} + +function finiteNumber(value: unknown): number | null { + return typeof value === "number" && Number.isFinite(value) ? value : null; +} + +/** + * Projects the LiteLLM document into a rate table. + * + * Entries without both an input and an output rate are dropped: a half-priced + * model would silently under-report cost, which is worse than reporting the + * model as unpriced. + */ +export function parseRateTable(document: unknown): RateTable { + const table = new Map(); + if (typeof document !== "object" || document === null) return table; + + for (const [name, raw] of Object.entries(document as Record)) { + if (typeof raw !== "object" || raw === null) continue; + const entry = raw as LiteLlmEntry; + const input = finiteNumber(entry.input_cost_per_token); + const output = finiteNumber(entry.output_cost_per_token); + if (input === null || output === null) continue; + + table.set(normalizeModelName(name), { + inputCostPerToken: input, + outputCostPerToken: output, + // Anthropic bills cache reads at a discount and cache writes at a + // premium. When a model omits them, cached input is priced as plain + // input rather than as free. + cacheReadCostPerToken: finiteNumber(entry.cache_read_input_token_cost) ?? input, + cacheCreationCostPerToken: finiteNumber(entry.cache_creation_input_token_cost) ?? input, + }); + } + return table; +} + +/** + * Canonicalises a model name for lookup. + * + * Strips a `provider/` prefix (LiteLLM publishes both `claude-opus-5` and + * `anthropic/claude-opus-5`) and lowercases, since transcripts are inconsistent + * about casing. + */ +export function normalizeModelName(model: string): string { + const trimmed = model.trim().toLowerCase(); + const slash = trimmed.lastIndexOf("/"); + return slash === -1 ? trimmed : trimmed.slice(slash + 1); +} + +/** + * Models we never price, regardless of the table. + * + * `` marks locally generated messages that were never billed. Bare + * family names ("opus", "sonnet") are genuinely ambiguous across generations, + * so we report them as unpriced instead of guessing a generation. + */ +const UNPRICEABLE_MODELS = new Set([ + "", + "synthetic", + "opus", + "sonnet", + "haiku", + "fable", +]); + +export function lookupRate(table: RateTable, model: string): ModelRate | null { + const normalized = normalizeModelName(model); + if (normalized.length === 0 || UNPRICEABLE_MODELS.has(normalized)) return null; + return table.get(normalized) ?? null; +} + +export interface PricedUsage { + readonly costUsd: number; + readonly costSource: UsageCostSource; +} + +/** + * Prices a bucket's tokens. + * + * `reasoningTokens` is intentionally not charged separately: it is already + * counted inside `outputTokens`. + */ +export function priceUsage( + table: RateTable, + model: string, + totals: UsageTokenTotals, + reportedCostUsd: number | null, +): PricedUsage { + if (reportedCostUsd !== null && Number.isFinite(reportedCostUsd)) { + return { costUsd: reportedCostUsd, costSource: "providerReported" }; + } + + const rate = lookupRate(table, model); + if (rate === null) return { costUsd: 0, costSource: "unpriced" }; + + const costUsd = + totals.uncachedInputTokens * rate.inputCostPerToken + + totals.cachedInputTokens * rate.cacheReadCostPerToken + + totals.cacheCreationTokens * rate.cacheCreationCostPerToken + + totals.outputTokens * rate.outputCostPerToken; + + return { costUsd, costSource: "modelPriced" }; +} + +/** + * What the cached input would have cost at full input rates, minus what it + * actually cost. Drives the "cache savings" figure. + */ +export function cacheSavingsUsd(table: RateTable, model: string, totals: UsageTokenTotals): number { + const rate = lookupRate(table, model); + if (rate === null) return 0; + return totals.cachedInputTokens * (rate.inputCostPerToken - rate.cacheReadCostPerToken); +} diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts new file mode 100644 index 00000000000..64673e96c09 --- /dev/null +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -0,0 +1,206 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + decodeScanCache, + dedupeWithinFile, + encodeScanCache, + pruneScanCache, + type ScanCache, +} from "./usageScanCache.ts"; +import type { UsageRecord } from "./usageTranscripts.ts"; + +function record(overrides: Partial = {}): UsageRecord { + return { + provider: "claude", + timestampMs: 1_786_000_000_000, + model: "claude-fable-5", + sessionId: "session-a", + totals: { + uncachedInputTokens: 2, + cachedInputTokens: 1000, + cacheCreationTokens: 10, + outputTokens: 50, + reasoningTokens: 0, + }, + reportedCostUsd: null, + dedupeKey: "msg_1:", + ...overrides, + }; +} + +function cacheWith(entries: readonly [string, number, readonly UsageRecord[]][]): ScanCache { + const cache: ScanCache = new Map(); + for (const [path, mtimeMs, records] of entries) { + cache.set(path, { size: records.length * 10, mtimeMs, provider: "claude", records }); + } + return cache; +} + +describe("scan cache round trip", () => { + it("restores records unchanged", () => { + const original = cacheWith([ + ["/a.jsonl", 100, [record(), record({ dedupeKey: "msg_2:", model: "claude-opus-5" })]], + ["/b.jsonl", 200, [record({ sessionId: "session-b", reportedCostUsd: 1.5 })]], + ]); + + const restored = decodeScanCache(JSON.parse(JSON.stringify(encodeScanCache(original)))); + + expect(restored.size).toBe(2); + expect(restored.get("/a.jsonl")).toEqual(original.get("/a.jsonl")); + expect(restored.get("/b.jsonl")).toEqual(original.get("/b.jsonl")); + }); + + it("interns repeated model and session strings", () => { + const encoded = encodeScanCache( + cacheWith([["/a.jsonl", 100, [record(), record({ dedupeKey: "msg_2:" }), record()]]]), + ); + + expect(encoded.models).toEqual(["claude-fable-5"]); + expect(encoded.sessions).toEqual(["session-a"]); + }); + + it("treats a corrupt or foreign document as an empty cache", () => { + // A bad cache should cost one cold scan, never a broken page. + expect(decodeScanCache(null).size).toBe(0); + expect(decodeScanCache("nonsense").size).toBe(0); + expect(decodeScanCache({ version: 999, models: [], sessions: [], files: {} }).size).toBe(0); + }); + + it("skips malformed file entries but keeps good ones", () => { + const encoded = encodeScanCache(cacheWith([["/good.jsonl", 100, [record()]]])); + const withJunk = { + ...encoded, + files: { ...encoded.files, "/bad.jsonl": { s: "nope", m: 1, p: "claude", r: [] } }, + }; + + const restored = decodeScanCache(JSON.parse(JSON.stringify(withJunk))); + expect([...restored.keys()]).toEqual(["/good.jsonl"]); + }); + + it("rejects the whole cache when an intern table holds a non-string", () => { + // models: [1] would pass the undefined guard, put a number in a record's + // model, and crash normalizeModelName at aggregate time. + const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]])); + const poisoned = { ...encoded, models: [1] }; + + expect(decodeScanCache(JSON.parse(JSON.stringify(poisoned))).size).toBe(0); + }); + + it("drops the whole entry when any row is corrupt, forcing a cold re-parse", () => { + // Keeping the surviving rows under the original (size, mtime) would read + // as a valid warm hit and the file would never be re-parsed. + const encoded = encodeScanCache( + cacheWith([["/a.jsonl", 100, [record(), record({ dedupeKey: "msg_2:" })]]]), + ); + const rows = encoded.files["/a.jsonl"]!.r; + const poisoned = { + ...encoded, + files: { + "/a.jsonl": { + ...encoded.files["/a.jsonl"]!, + r: [rows[0]!, [...rows[1]!.slice(0, 3), "not-a-number", ...rows[1]!.slice(4)]], + }, + }, + }; + + const restored = decodeScanCache(JSON.parse(JSON.stringify(poisoned))); + expect(restored.has("/a.jsonl")).toBe(false); + }); +}); + +describe("pruneScanCache", () => { + const retentionCutoffMs = 1000; + + it("drops entries older than retention", () => { + const cache = cacheWith([["/old.jsonl", 500, [record()]]]); + + const removed = pruneScanCache(cache, { + livePaths: new Set(), + walkedRoots: ["/"], + windowStartMs: 400, + retentionCutoffMs, + }); + + expect(removed).toBe(1); + expect(cache.size).toBe(0); + }); + + it("drops in-window entries whose file has disappeared", () => { + const cache = cacheWith([["/gone.jsonl", 5000, [record()]]]); + + pruneScanCache(cache, { + livePaths: new Set(), + walkedRoots: ["/"], + windowStartMs: 4000, + retentionCutoffMs, + }); + + expect(cache.size).toBe(0); + }); + + it("keeps entries outside the walked window that are still within retention", () => { + // Viewing 7 days must not evict the 30-day entries, which that walk never + // looked for and so cannot prove are gone. + const cache = cacheWith([["/older-but-valid.jsonl", 2000, [record()]]]); + + const removed = pruneScanCache(cache, { + livePaths: new Set(), + walkedRoots: ["/"], + windowStartMs: 4000, + retentionCutoffMs, + }); + + expect(removed).toBe(0); + expect(cache.size).toBe(1); + }); + + it("keeps entries the walk saw", () => { + const cache = cacheWith([["/live.jsonl", 5000, [record()]]]); + + pruneScanCache(cache, { + livePaths: new Set(["/live.jsonl"]), + walkedRoots: ["/"], + windowStartMs: 4000, + retentionCutoffMs, + }); + + expect(cache.size).toBe(1); + }); +}); + +describe("pruneScanCache with an unwalked root", () => { + it("keeps in-window entries for a provider whose directory was not walked", () => { + // A missing provider root or failed settings read leaves livePaths without + // that provider's files. Its warm entries must survive the pass. + const cache = cacheWith([["/codex/sessions/a.jsonl", 5000, [record()]]]); + + const removed = pruneScanCache(cache, { + livePaths: new Set(), + walkedRoots: ["/claude/projects"], + windowStartMs: 4000, + retentionCutoffMs: 1000, + }); + + expect(removed).toBe(0); + expect(cache.size).toBe(1); + }); +}); + +describe("dedupeWithinFile", () => { + it("keeps the first record per dedupe key", () => { + const kept = dedupeWithinFile([ + record({ totals: { ...record().totals, outputTokens: 1 } }), + record({ totals: { ...record().totals, outputTokens: 999 } }), + record({ dedupeKey: "msg_2:" }), + ]); + + expect(kept).toHaveLength(2); + expect(kept[0]?.totals.outputTokens).toBe(1); + }); + + it("keeps every record that has no dedupe key", () => { + expect( + dedupeWithinFile([record({ dedupeKey: null }), record({ dedupeKey: null })]), + ).toHaveLength(2); + }); +}); diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts new file mode 100644 index 00000000000..0dafa7a6daf --- /dev/null +++ b/apps/server/src/usage/usageScanCache.ts @@ -0,0 +1,253 @@ +/** + * Durable per-file scan cache. + * + * Transcripts are append-only and a file that has not changed can never yield + * different usage, so parsed records are keyed by `(size, mtime)` and reused. + * Without this every server restart re-parses the whole window: roughly 3.5s + * for a 30-day scan here, against ~11ms to reload this cache. + * + * Caching *per file* rather than per day is deliberate. It is timezone + * independent, so changing the reporting zone does not invalidate anything, and + * it keeps cross-file de-duplication exact: cached entries are de-duplicated + * within their own file only, and the aggregator still applies the global + * dedupe pass over the small surviving key set. + * + * @module usageScanCache + */ +import type { UsageProviderKind } from "@t3tools/contracts"; + +import type { UsageRecord } from "./usageTranscripts.ts"; + +export const USAGE_SCAN_CACHE_VERSION = 1 as const; + +export interface CachedFile { + readonly size: number; + readonly mtimeMs: number; + readonly provider: UsageProviderKind; + readonly records: readonly UsageRecord[]; +} + +export type ScanCache = Map; + +/** + * Row layout for the serialised form. Positional and interned rather than + * object-per-record: on a 30-day window that is the difference between a file + * measured in tens of megabytes and one under six. + */ +type SerializedRecord = readonly [ + timestampMs: number, + modelIndex: number, + sessionIndex: number, + uncachedInputTokens: number, + cachedInputTokens: number, + cacheCreationTokens: number, + outputTokens: number, + reasoningTokens: number, + dedupeKey: string | null, + reportedCostUsd: number | null, +]; + +interface SerializedFile { + readonly s: number; + readonly m: number; + readonly p: UsageProviderKind; + readonly r: readonly SerializedRecord[]; +} + +interface SerializedCache { + readonly version: number; + readonly models: readonly string[]; + readonly sessions: readonly string[]; + readonly files: Readonly>; +} + +/** Serialises the cache, interning the repeated model and session strings. */ +export function encodeScanCache(cache: ScanCache): SerializedCache { + const models: string[] = []; + const sessions: string[] = []; + const modelIndex = new Map(); + const sessionIndex = new Map(); + + const intern = (table: string[], index: Map, value: string): number => { + const existing = index.get(value); + if (existing !== undefined) return existing; + const next = table.length; + table.push(value); + index.set(value, next); + return next; + }; + + const files: Record = {}; + for (const [path, entry] of cache) { + files[path] = { + s: entry.size, + m: entry.mtimeMs, + p: entry.provider, + r: entry.records.map((record) => [ + record.timestampMs, + intern(models, modelIndex, record.model), + intern(sessions, sessionIndex, record.sessionId), + record.totals.uncachedInputTokens, + record.totals.cachedInputTokens, + record.totals.cacheCreationTokens, + record.totals.outputTokens, + record.totals.reasoningTokens, + record.dedupeKey, + record.reportedCostUsd, + ]), + }; + } + + return { version: USAGE_SCAN_CACHE_VERSION, models, sessions, files }; +} + +function isRecordArray(value: unknown): value is readonly unknown[] { + return Array.isArray(value); +} + +/** + * Rebuilds the cache from a parsed document. + * + * Anything malformed yields an empty cache rather than an error: a corrupt + * cache should cost one cold scan, never a broken page. + */ +export function decodeScanCache(document: unknown): ScanCache { + const cache: ScanCache = new Map(); + if (typeof document !== "object" || document === null) return cache; + + const root = document as Partial; + if (root.version !== USAGE_SCAN_CACHE_VERSION) return cache; + if (!isRecordArray(root.models) || !isRecordArray(root.sessions)) return cache; + if (typeof root.files !== "object" || root.files === null) return cache; + + // The intern tables must be all strings: a numeric entry would pass the + // undefined guard below, land in a record's model, and crash the aggregate + // at normalizeModelName. A corrupt table rejects the whole cache. + if (!root.models.every((value) => typeof value === "string")) return cache; + if (!root.sessions.every((value) => typeof value === "string")) return cache; + const models = root.models as readonly string[]; + const sessions = root.sessions as readonly string[]; + + for (const [path, raw] of Object.entries(root.files)) { + if (typeof raw !== "object" || raw === null) continue; + const entry = raw as Partial; + if (typeof entry.s !== "number" || typeof entry.m !== "number") continue; + if (entry.p !== "claude" && entry.p !== "codex") continue; + if (!isRecordArray(entry.r)) continue; + + const provider: UsageProviderKind = entry.p; + const records: UsageRecord[] = []; + // Any corrupt row disqualifies the whole entry. Keeping the survivors + // under the original (size, mtime) would read as a valid warm hit and the + // file would never be re-parsed, silently losing the dropped rows' usage. + let corrupt = false; + for (const row of entry.r) { + if (!isRecordArray(row) || row.length < 10) { + corrupt = true; + break; + } + const [ + timestampMs, + modelIndex, + sessionIndex, + uncached, + cached, + cacheCreation, + output, + reasoning, + dedupeKey, + reportedCostUsd, + ] = row as SerializedRecord; + + const model = typeof modelIndex === "number" ? models[modelIndex] : undefined; + if ( + typeof timestampMs !== "number" || + !Number.isFinite(timestampMs) || + model === undefined || + !Number.isFinite(uncached) || + !Number.isFinite(cached) || + !Number.isFinite(cacheCreation) || + !Number.isFinite(output) || + !Number.isFinite(reasoning) + ) { + corrupt = true; + break; + } + + records.push({ + provider, + timestampMs, + model, + sessionId: (typeof sessionIndex === "number" ? sessions[sessionIndex] : undefined) ?? "", + totals: { + uncachedInputTokens: uncached, + cachedInputTokens: cached, + cacheCreationTokens: cacheCreation, + outputTokens: output, + reasoningTokens: reasoning, + }, + reportedCostUsd: typeof reportedCostUsd === "number" ? reportedCostUsd : null, + dedupeKey: typeof dedupeKey === "string" ? dedupeKey : null, + }); + } + + if (corrupt) continue; + cache.set(path, { size: entry.s, mtimeMs: entry.m, provider, records }); + } + + return cache; +} + +export interface PruneOptions { + /** Files the walk just saw. Only meaningful inside the walked window. */ + readonly livePaths: ReadonlySet; + /** + * Roots the walk actually completed. Absence from `livePaths` only proves a + * file is gone when its root was walked: a provider whose directory failed to + * resolve this pass must not have its warm entries purged. + */ + readonly walkedRoots: readonly string[]; + /** Start of the walked window; entries older than this were not looked for. */ + readonly windowStartMs: number; + /** Entries older than this are dropped regardless. */ + readonly retentionCutoffMs: number; +} + +/** + * Drops aged-out entries, and entries for files that have disappeared. + * + * The walk only covers the requested window, so absence from `livePaths` only + * proves deletion for entries *inside* that window. Pruning everything the walk + * missed would evict the 30-day entries every time someone looked at 7 days. + * + * Replaces an earlier record cap that cleared the whole cache once exceeded, + * which meant a large enough window never warmed up at all. + */ +export function pruneScanCache(cache: ScanCache, options: PruneOptions): number { + let removed = 0; + for (const [path, entry] of cache) { + const agedOut = entry.mtimeMs < options.retentionCutoffMs; + const underWalkedRoot = options.walkedRoots.some((root) => path.startsWith(root)); + const deleted = + underWalkedRoot && entry.mtimeMs >= options.windowStartMs && !options.livePaths.has(path); + if (agedOut || deleted) { + cache.delete(path); + removed += 1; + } + } + return removed; +} + +/** Within-file de-duplication, applied before an entry is cached. */ +export function dedupeWithinFile(records: readonly UsageRecord[]): readonly UsageRecord[] { + const seen = new Set(); + const kept: UsageRecord[] = []; + for (const record of records) { + if (record.dedupeKey !== null) { + if (seen.has(record.dedupeKey)) continue; + seen.add(record.dedupeKey); + } + kept.push(record); + } + return kept; +} diff --git a/apps/server/src/usage/usageTranscriptReader.ts b/apps/server/src/usage/usageTranscriptReader.ts new file mode 100644 index 00000000000..c72f0c24db6 --- /dev/null +++ b/apps/server/src/usage/usageTranscriptReader.ts @@ -0,0 +1,141 @@ +// @effect-diagnostics nodeBuiltinImport:off +/** + * Raw filesystem access for transcript scanning. + * + * Isolated here so the rest of the usage code stays on Effect's `FileSystem`. + * The direct `node:fs` streaming is deliberate: a cold 30-day window is ~1.4 GB + * across ~1,500 files, and `readline` over a read stream is roughly an order of + * magnitude cheaper than materialising each file. The equivalent Effect stream + * pipeline is idiomatic but not fast enough to sit behind a page load. + * + * @module usageTranscriptReader + */ +import * as NodeFS from "node:fs"; +import * as NodeFSP from "node:fs/promises"; +import * as NodePath from "node:path"; +import * as NodeReadline from "node:readline"; + +import type { UsageProviderKind } from "@t3tools/contracts"; + +import { + initialCodexScanState, + mightCarryUsage, + parseClaudeLine, + parseCodexLine, + type UsageRecord, +} from "./usageTranscripts.ts"; + +export interface TranscriptFile { + readonly path: string; + readonly size: number; + readonly mtimeMs: number; +} + +/** + * Lists `.jsonl` transcripts under `root` last modified at or after `sinceMs`. + * + * Errors on individual entries are swallowed: session files rotate and get + * removed while the walk is in flight, and a partial listing is far better than + * failing the page. + */ +export async function listTranscriptFiles( + root: string, + sinceMs: number, +): Promise { + const found: TranscriptFile[] = []; + + const walk = async (dir: string): Promise => { + let entries; + try { + entries = await NodeFSP.readdir(dir, { withFileTypes: true }); + } catch { + return; + } + for (const entry of entries) { + const child = NodePath.join(dir, entry.name); + if (entry.isDirectory()) { + await walk(child); + continue; + } + if (!entry.name.endsWith(".jsonl")) continue; + try { + const stats = await NodeFSP.stat(child); + if (stats.mtimeMs >= sinceMs) { + found.push({ path: child, size: stats.size, mtimeMs: stats.mtimeMs }); + } + } catch { + // Vanished between readdir and stat. + } + } + }; + + await walk(root); + return found; +} + +/** + * Filesystem identity of a directory, as `device:inode`. + * + * Used to tell "two servers reading the same transcript directory" apart from + * "two machines whose hostname and home path happen to match". Returns an empty + * string when the directory cannot be stat'd. + */ +export async function readDirectoryVolumeId(path: string): Promise { + try { + const stats = await NodeFSP.stat(path); + return `${stats.dev}:${stats.ino}`; + } catch { + return ""; + } +} + +/** + * Streams one transcript and returns the usage records it contains, or `null` + * when the file could not be read. + * + * The distinction matters to the caller's cache: a genuinely empty transcript + * is a stable fact worth memoising, while a transient read failure memoised + * under the same `(size, mtime)` key would silently drop that file's usage + * until the file next changes. + * + * Codex carries the active model on `turn_context` lines that hold no usage of + * their own, so those still have to pass through the reducer to keep model + * attribution correct. + */ +export async function readTranscriptRecords( + filePath: string, + provider: UsageProviderKind, +): Promise { + const records: UsageRecord[] = []; + const codexState = initialCodexScanState(); + + try { + const lines = NodeReadline.createInterface({ + input: NodeFS.createReadStream(filePath, { encoding: "utf8" }), + crlfDelay: Infinity, + }); + + for await (const line of lines) { + if (provider === "codex") { + if ( + !mightCarryUsage(line, provider) && + !line.includes('"turn_context"') && + !line.includes('"session_meta"') + ) { + continue; + } + const record = parseCodexLine(line, codexState); + if (record !== null) records.push(record); + continue; + } + + if (!mightCarryUsage(line, provider)) continue; + const record = parseClaudeLine(line); + if (record !== null) records.push(record); + } + } catch { + return null; + } + + return records; +} diff --git a/apps/server/src/usage/usageTranscripts.test.ts b/apps/server/src/usage/usageTranscripts.test.ts new file mode 100644 index 00000000000..1fec9d28d9b --- /dev/null +++ b/apps/server/src/usage/usageTranscripts.test.ts @@ -0,0 +1,151 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + initialCodexScanState, + parseClaudeLine, + parseCodexLine, + totalTokens, +} from "./usageTranscripts.ts"; + +/** Shaped after a real Claude Code assistant record. */ +function claudeLine(overrides: { + messageId: string; + contentType: string; + model?: string; + outputTokens?: number; +}): string { + return JSON.stringify({ + type: "assistant", + timestamp: "2026-08-07T04:05:13.944Z", + sessionId: "5a128faa-8253-489e-b935-6c08e8e670c0", + cwd: "/home/theo/project", + message: { + id: overrides.messageId, + role: "assistant", + model: overrides.model ?? "claude-fable-5", + content: [{ type: overrides.contentType }], + usage: { + input_tokens: 2, + cache_creation_input_tokens: 66818, + cache_read_input_tokens: 1000, + output_tokens: overrides.outputTokens ?? 286, + }, + }, + }); +} + +describe("parseClaudeLine", () => { + it("extracts token totals and a dedupe key", () => { + const record = parseClaudeLine(claudeLine({ messageId: "msg_1", contentType: "text" })); + + expect(record).not.toBeNull(); + expect(record?.provider).toBe("claude"); + expect(record?.model).toBe("claude-fable-5"); + expect(record?.totals).toEqual({ + uncachedInputTokens: 2, + cachedInputTokens: 1000, + cacheCreationTokens: 66818, + outputTokens: 286, + reasoningTokens: 0, + }); + expect(record?.dedupeKey).toBe("msg_1:"); + }); + + it("gives every content block of one message the same dedupe key", () => { + // T3 Code writes one record per content block, each repeating the parent + // message's full usage. Summing them would overcount ~2.4x on real data. + const text = parseClaudeLine(claudeLine({ messageId: "msg_2", contentType: "text" })); + const toolUse = parseClaudeLine(claudeLine({ messageId: "msg_2", contentType: "tool_use" })); + + expect(text?.dedupeKey).toBe(toolUse?.dedupeKey); + expect(text?.totals).toEqual(toolUse?.totals); + }); + + it("ignores records that are not assistant messages", () => { + expect(parseClaudeLine(JSON.stringify({ type: "user", message: {} }))).toBeNull(); + expect(parseClaudeLine("not json")).toBeNull(); + }); +}); + +describe("parseCodexLine", () => { + const sessionMeta = JSON.stringify({ + type: "session_meta", + timestamp: "2026-08-01T05:17:41.289Z", + payload: { type: "session_meta", id: "019fbbc1-b12c-7360-a685-28c181f0025f" }, + }); + const turnContext = JSON.stringify({ + type: "turn_context", + timestamp: "2026-08-01T05:17:42.694Z", + payload: { type: "turn_context", model: "gpt-5.6-sol" }, + }); + const tokenCount = (inputTokens: number, cached: number, output: number, reasoning: number) => + JSON.stringify({ + type: "event_msg", + timestamp: "2026-08-01T05:17:49.919Z", + payload: { + type: "token_count", + info: { + last_token_usage: { + input_tokens: inputTokens, + cached_input_tokens: cached, + cache_write_input_tokens: 0, + output_tokens: output, + reasoning_output_tokens: reasoning, + }, + }, + }, + }); + + it("attributes usage to the model from the preceding turn context", () => { + const state = initialCodexScanState(); + parseCodexLine(sessionMeta, state); + parseCodexLine(turnContext, state); + const record = parseCodexLine(tokenCount(19239, 11008, 299, 116), state); + + expect(record?.provider).toBe("codex"); + expect(record?.model).toBe("gpt-5.6-sol"); + expect(record?.sessionId).toBe("019fbbc1-b12c-7360-a685-28c181f0025f"); + // Codex reports input_tokens inclusive of the cached portion. + expect(record?.totals.uncachedInputTokens).toBe(19239 - 11008); + expect(record?.totals.cachedInputTokens).toBe(11008); + expect(record?.totals.reasoningTokens).toBe(116); + }); + + it("skips a repeated token_count so deltas are not double counted", () => { + const state = initialCodexScanState(); + parseCodexLine(turnContext, state); + const first = parseCodexLine(tokenCount(100, 0, 10, 0), state); + const repeat = parseCodexLine(tokenCount(100, 0, 10, 0), state); + + expect(first).not.toBeNull(); + expect(repeat).toBeNull(); + }); + + it("drops usage that arrives before any model is known", () => { + const state = initialCodexScanState(); + expect(parseCodexLine(tokenCount(100, 0, 10, 0), state)).toBeNull(); + }); + + it("does not let a pre-model event poison the duplicate signature", () => { + // A token_count before its turn_context is dropped; the identical event + // re-emitted once the model is known must still be counted. + const state = initialCodexScanState(); + expect(parseCodexLine(tokenCount(100, 0, 10, 0), state)).toBeNull(); + parseCodexLine(turnContext, state); + expect(parseCodexLine(tokenCount(100, 0, 10, 0), state)).not.toBeNull(); + }); +}); + +describe("totalTokens", () => { + it("does not add reasoning on top of output", () => { + expect( + totalTokens({ + uncachedInputTokens: 10, + cachedInputTokens: 20, + cacheCreationTokens: 30, + outputTokens: 40, + reasoningTokens: 25, + }), + ).toBe(100); + }); +}); diff --git a/apps/server/src/usage/usageTranscripts.ts b/apps/server/src/usage/usageTranscripts.ts new file mode 100644 index 00000000000..338713d8b1b --- /dev/null +++ b/apps/server/src/usage/usageTranscripts.ts @@ -0,0 +1,246 @@ +/** + * Pure parsers for the provider CLIs' on-disk session transcripts. + * + * Both parsers are line-at-a-time reducers so callers can stream large files + * without materialising them. Neither touches the filesystem. + * + * @module usageTranscripts + */ +import type { UsageProviderKind, UsageTokenTotals } from "@t3tools/contracts"; + +export interface UsageRecord { + readonly provider: UsageProviderKind; + readonly timestampMs: number; + readonly model: string; + readonly sessionId: string; + readonly totals: UsageTokenTotals; + readonly reportedCostUsd: number | null; + /** + * Key for cross-file de-duplication, or `null` when the record is inherently + * unique and needs no dedup. + */ + readonly dedupeKey: string | null; +} + +const EMPTY_TOTALS: UsageTokenTotals = { + uncachedInputTokens: 0, + cachedInputTokens: 0, + cacheCreationTokens: 0, + outputTokens: 0, + reasoningTokens: 0, +}; + +function int(value: unknown): number { + return typeof value === "number" && Number.isFinite(value) && value > 0 ? Math.trunc(value) : 0; +} + +function parseTimestampMs(value: unknown): number | null { + if (typeof value !== "string") return null; + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? null : parsed; +} + +export function addTotals(a: UsageTokenTotals, b: UsageTokenTotals): UsageTokenTotals { + return { + uncachedInputTokens: a.uncachedInputTokens + b.uncachedInputTokens, + cachedInputTokens: a.cachedInputTokens + b.cachedInputTokens, + cacheCreationTokens: a.cacheCreationTokens + b.cacheCreationTokens, + outputTokens: a.outputTokens + b.outputTokens, + reasoningTokens: a.reasoningTokens + b.reasoningTokens, + }; +} + +export function totalTokens(totals: UsageTokenTotals): number { + // reasoningTokens is a subset of outputTokens and must not be added again. + return ( + totals.uncachedInputTokens + + totals.cachedInputTokens + + totals.cacheCreationTokens + + totals.outputTokens + ); +} + +/** + * Cheap substring gate applied before `JSON.parse`. + * + * Transcripts are mostly tool output; only a minority of lines carry usage. On + * a 30-day window this skips roughly half the lines outright and is worth about + * an order of magnitude. + */ +export function mightCarryUsage(line: string, provider: UsageProviderKind): boolean { + return provider === "claude" ? line.includes('"usage"') : line.includes('"token_count"'); +} + +/* -------------------------------------------------------------------------- */ +/* Claude Code */ +/* -------------------------------------------------------------------------- */ + +/** + * Parses one line of a Claude Code transcript. + * + * T3 Code writes one record per assistant *content block*, and every one of + * those records repeats the same complete `usage` object for the parent + * message. Summing them overcounts by roughly 2.4x on a real workload, so the + * caller must drop repeats by `dedupeKey` and keep the first. + */ +export function parseClaudeLine(line: string): UsageRecord | null { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + return null; + } + if (typeof parsed !== "object" || parsed === null) return null; + + const record = parsed as Record; + if (record["type"] !== "assistant") return null; + + const message = record["message"]; + if (typeof message !== "object" || message === null) return null; + const messageRecord = message as Record; + + const usage = messageRecord["usage"]; + if (typeof usage !== "object" || usage === null) return null; + const usageRecord = usage as Record; + + const timestampMs = parseTimestampMs(record["timestamp"]); + if (timestampMs === null) return null; + + const model = typeof messageRecord["model"] === "string" ? messageRecord["model"] : ""; + if (model.length === 0) return null; + + const messageId = typeof messageRecord["id"] === "string" ? messageRecord["id"] : null; + const requestId = typeof record["requestId"] === "string" ? record["requestId"] : null; + // Matches ccusage: prefer the message/request pair, fall back to whichever + // half exists. Records with neither cannot be de-duplicated. + const dedupeKey = + messageId === null && requestId === null ? null : `${messageId ?? ""}:${requestId ?? ""}`; + + const cost = record["costUSD"]; + + return { + provider: "claude", + timestampMs, + model, + sessionId: typeof record["sessionId"] === "string" ? record["sessionId"] : "", + totals: { + uncachedInputTokens: int(usageRecord["input_tokens"]), + cachedInputTokens: int(usageRecord["cache_read_input_tokens"]), + cacheCreationTokens: int(usageRecord["cache_creation_input_tokens"]), + outputTokens: int(usageRecord["output_tokens"]), + // Anthropic folds thinking tokens into output and does not break them out. + reasoningTokens: 0, + }, + reportedCostUsd: typeof cost === "number" && Number.isFinite(cost) ? cost : null, + dedupeKey, + }; +} + +/* -------------------------------------------------------------------------- */ +/* Codex */ +/* -------------------------------------------------------------------------- */ + +/** + * Rolling state for a single Codex rollout file. + * + * Codex `token_count` events carry no model, so the model is carried forward + * from the most recent `turn_context`. Sessions that switch models mid-run + * attribute correctly from the switch onward. + */ +export interface CodexScanState { + model: string; + sessionId: string; + lastUsageSignature: string | null; +} + +export function initialCodexScanState(): CodexScanState { + return { model: "", sessionId: "", lastUsageSignature: null }; +} + +/** + * Feeds one line of a Codex rollout into `state`, returning a record when the + * line was a usage event. + * + * Deltas come from `last_token_usage`. Summing those across a session + * reconciles with the session's final `total_token_usage`, provided + * consecutive duplicate events are dropped, which this does. + */ +export function parseCodexLine(line: string, state: CodexScanState): UsageRecord | null { + let parsed: unknown; + try { + parsed = JSON.parse(line); + } catch { + return null; + } + if (typeof parsed !== "object" || parsed === null) return null; + + const record = parsed as Record; + const payload = record["payload"]; + if (typeof payload !== "object" || payload === null) return null; + const payloadRecord = payload as Record; + const payloadType = payloadRecord["type"]; + + if (record["type"] === "session_meta") { + const id = payloadRecord["id"] ?? payloadRecord["session_id"]; + if (typeof id === "string") state.sessionId = id; + return null; + } + + if (record["type"] === "turn_context") { + if (typeof payloadRecord["model"] === "string") state.model = payloadRecord["model"]; + return null; + } + + if (payloadType !== "token_count") return null; + + const info = payloadRecord["info"]; + if (typeof info !== "object" || info === null) return null; + const last = (info as Record)["last_token_usage"]; + if (typeof last !== "object" || last === null) return null; + const lastRecord = last as Record; + + // Only an event that is otherwise eligible may consume the duplicate + // signature. A token_count arriving before its turn_context (no model yet) + // must not poison it, or the re-emitted copy after the model is known would + // be skipped as a duplicate and those tokens never counted. + const timestampMs = parseTimestampMs(record["timestamp"]); + if (timestampMs === null) return null; + if (state.model.length === 0) return null; + + // Codex re-emits an unchanged token_count on some stream boundaries. Summing + // those would double count, so identical consecutive payloads are skipped. + const signature = JSON.stringify(lastRecord); + if (signature === state.lastUsageSignature) return null; + state.lastUsageSignature = signature; + + const inputTokens = int(lastRecord["input_tokens"]); + const cachedInputTokens = int(lastRecord["cached_input_tokens"]); + const cacheCreationTokens = int(lastRecord["cache_write_input_tokens"]); + const outputTokens = int(lastRecord["output_tokens"]); + + const totals: UsageTokenTotals = { + // Codex reports `input_tokens` inclusive of the cached portion. + uncachedInputTokens: Math.max(0, inputTokens - cachedInputTokens - cacheCreationTokens), + cachedInputTokens, + cacheCreationTokens, + outputTokens, + // Reported inside output_tokens, surfaced separately for the token mix. + reasoningTokens: Math.min(outputTokens, int(lastRecord["reasoning_output_tokens"])), + }; + + if (totalTokens(totals) === 0) return null; + + return { + provider: "codex", + timestampMs, + model: state.model, + sessionId: state.sessionId, + totals, + // Codex does not report cost in the rollout. + reportedCostUsd: null, + // Rollout files are unique per session, so events need no global dedup. + dedupeKey: null, + }; +} + +export { EMPTY_TOTALS }; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index a6b155c296f..6d518fe16cf 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -105,6 +105,7 @@ import { requiredScopeForRpcMethod } from "./auth/RpcAuthorization.ts"; import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; +import * as UsageService from "./usage/UsageService.ts"; import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; import * as SourceControlDiscovery from "./sourceControl/SourceControlDiscovery.ts"; import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts"; @@ -412,6 +413,7 @@ const makeWsRpcLayer = ( const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics; const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor; const resourceTelemetry = yield* ResourceTelemetry.ResourceTelemetry; + const usage = yield* UsageService.UsageService; const relayClient = yield* RelayClient.RelayClient; const authorizationError = (requiredScope: AuthEnvironmentScope) => new EnvironmentAuthorizationError({ @@ -1529,6 +1531,10 @@ const makeWsRpcLayer = ( "rpc.aggregate": "server", }, ), + [WS_METHODS.serverGetUsageSummary]: (input) => + observeRpcEffect(WS_METHODS.serverGetUsageSummary, usage.readSummary(input), { + "rpc.aggregate": "server", + }), [WS_METHODS.serverRetryResourceTelemetry]: (_input) => observeRpcEffect(WS_METHODS.serverRetryResourceTelemetry, resourceTelemetry.retry, { "rpc.aggregate": "server", diff --git a/apps/web/src/components/sidebar/SidebarChrome.tsx b/apps/web/src/components/sidebar/SidebarChrome.tsx index df06c431fd2..a8d3ef41416 100644 --- a/apps/web/src/components/sidebar/SidebarChrome.tsx +++ b/apps/web/src/components/sidebar/SidebarChrome.tsx @@ -1,4 +1,4 @@ -import { SettingsIcon } from "lucide-react"; +import { ChartNoAxesColumnIcon, SettingsIcon } from "lucide-react"; import { memo, useCallback } from "react"; import { Link, useNavigate } from "@tanstack/react-router"; @@ -118,11 +118,24 @@ export const SidebarChromeFooter = memo(function SidebarChromeFooter() { void navigate({ to: "/settings" }); }, [isMobile, navigate, setOpenMobile]); + const handleUsageClick = useCallback(() => { + if (isMobile) { + setOpenMobile(false); + } + void navigate({ to: "/usage" }); + }, [isMobile, navigate, setOpenMobile]); + return ( + + + + Usage + + diff --git a/apps/web/src/components/usage/UsagePage.tsx b/apps/web/src/components/usage/UsagePage.tsx new file mode 100644 index 00000000000..2f3ab4b574c --- /dev/null +++ b/apps/web/src/components/usage/UsagePage.tsx @@ -0,0 +1,454 @@ +import type { UsageProviderKind } from "@t3tools/contracts"; +import { RefreshCwIcon } from "lucide-react"; +import { useMemo, useState } from "react"; + +import { cn } from "../../lib/utils"; +import { useUsage } from "../../state/usage"; +import { + enumerateDays, + formatCount, + formatDayShort, + formatPercent, + formatTokens, + formatUsd, + makeWindow, +} from "../../usage/usageFormat"; +import { ScrollArea } from "../ui/scroll-area"; +import { UsageChartLegend, UsageProviderChart, type UsageChartMetric } from "./UsageProviderChart"; +import { PROVIDER_COLOR, PROVIDER_LABEL, PROVIDER_MARK, PROVIDER_ORDER } from "./usageProviders"; + +const WINDOW_OPTIONS = [ + { days: 7, label: "7 days" }, + { days: 30, label: "30 days" }, + { days: 90, label: "90 days" }, +] as const; + +export function UsagePage() { + const [windowDays, setWindowDays] = useState(30); + const [metric, setMetric] = useState("cost"); + const [breakdown, setBreakdown] = useState<"model" | "day">("model"); + + // Recomputed only when the window length changes, so a re-render does not + // shift the range and refetch every environment. + const window = useMemo(() => makeWindow(windowDays), [windowDays]); + const { merged, environments, isPending, isPartial, refresh } = useUsage(window); + + const days = useMemo( + () => enumerateDays(window.sinceDay, window.untilDay), + [window.sinceDay, window.untilDay], + ); + const recentDays = useMemo(() => merged.daily.toReversed().slice(0, 8), [merged.daily]); + + // Ranked by whatever the toggle is showing, so the bars always descend. + const orderedProviders = useMemo( + () => + merged.providers.toSorted((a, b) => + metric === "cost" ? b.costUsd - a.costUsd : b.totalTokens - a.totalTokens, + ), + [merged.providers, metric], + ); + + const activeDays = merged.daily.filter((day) => day.totalTokens > 0).length; + const dailyAverage = activeDays === 0 ? 0 : merged.totalTokens / activeDays; + const observedInput = merged.uncachedInputTokens + merged.cachedInputTokens; + const cachedShare = observedInput === 0 ? 0 : merged.cachedInputTokens / observedInput; + + return ( + +
+
+
+

Usage

+

+ {formatDayShort(window.sinceDay)} to {formatDayShort(window.untilDay)} +

+
+
+
+ {WINDOW_OPTIONS.map((option) => ( + + ))} +
+ +
+
+ + + + {isPending ? ( +

+ Scanning provider transcripts… +

+ ) : ( + <> + {/* Cost first: the financial answer, then the provider split. */} +
+ {/* The summary follows the chart toggle, so the headline and the + series are always reading the same units. */} +
+
+ + {metric === "cost" ? "Raw token cost" : "Processed tokens"} + + + {metric === "cost" + ? `${formatUsd(merged.costUsd)}*` + : formatTokens(merged.totalTokens)} + + + {metric === "cost" + ? "* if billed at full API rate" + : `Input, cache reads and output across ${formatCount(merged.sessions)} sessions.`} + +
+ + {orderedProviders.map((provider) => { + const share = metric === "cost" ? provider.costShare : provider.tokenShare; + return ( +
+
+ + + {PROVIDER_LABEL[provider.provider]} + + + {metric === "cost" + ? formatUsd(provider.costUsd) + : formatTokens(provider.totalTokens)} + +
+
+
+
+ + {metric === "cost" + ? `${formatPercent(share)} of cost · ${formatTokens(provider.totalTokens)} tokens` + : `${formatPercent(share)} of tokens · ${formatUsd(provider.costUsd)}`} + +
+ ); + })} +
+ +
+
+

+ Daily {metric === "tokens" ? "processed tokens" : "cost"} +

+
+
+ {(["cost", "tokens"] as const).map((option) => ( + + ))} +
+ +
+
+ +
+
+ +
+ + + + + 0 + ? `${(merged.costQuality.cacheSavingsUsd / merged.costUsd).toFixed(1)}x the raw token cost` + : "vs full input rates" + } + /> +
+ +
+
+
+

Breakdown

+
+ {(["model", "day"] as const).map((option) => ( + + ))} +
+
+ + {breakdown === "model" ? ( + + + + + + + + + + + {merged.models.length === 0 ? ( + + + + ) : ( + merged.models.map((model) => ( + + + + + + + )) + )} + +
ModelCostShareTokens
+ No activity in this window. +
+ + + {model.model} + + + {formatUsd(model.costUsd)} + + {formatPercent(model.costShare)} + + {formatTokens(model.totalTokens)} +
+ ) : ( + + + + + {PROVIDER_ORDER.map((provider) => ( + + ))} + + + + + + {recentDays.length === 0 ? ( + + + + ) : ( + recentDays.map((day) => ( + + + {PROVIDER_ORDER.map((provider) => ( + + ))} + + + + )) + )} + +
Day + {PROVIDER_LABEL[provider]} + TotalTokens
+ No activity in this window. +
{formatDayShort(day.day)} + {formatUsd(day.byProvider.get(provider)?.costUsd ?? 0)} + + {formatUsd(day.costUsd)} + + {formatTokens(day.totalTokens)} +
+ )} +
+ +
+

Cost quality

+
+ + + + +
+
+
+ + )} +
+ + ); +} + +/** Brand mark for the harness a row belongs to. */ +function ProviderMark({ + provider, + className, +}: { + readonly provider: UsageProviderKind; + readonly className: string; +}) { + const Mark = PROVIDER_MARK[provider]; + return ; +} + +function Metric({ + label, + value, + detail, +}: { + readonly label: string; + readonly value: string; + readonly detail: string; +}) { + return ( +
+ {label} + {value} + {detail} +
+ ); +} + +function QualityRow({ label, value }: { readonly label: string; readonly value: string }) { + return ( +
+
{label}
+
{value}
+
+ ); +} + +/** + * Says plainly when the totals are incomplete: an environment still answering, + * one that failed, or one whose transcripts another environment already + * reported. + */ +function UsageCoverageNotice({ + environments, + duplicateSources, + staleEnvironments, + isPartial, +}: { + readonly environments: readonly { + environmentId: string; + label: string; + error: string | null; + isPending: boolean; + }[]; + readonly duplicateSources: readonly string[]; + readonly staleEnvironments: readonly string[]; + readonly isPartial: boolean; +}) { + const failed = environments.filter((environment) => environment.error !== null); + const stale = environments.filter((environment) => + staleEnvironments.includes(environment.environmentId), + ); + if (failed.length === 0 && stale.length === 0 && duplicateSources.length === 0 && !isPartial) { + return null; + } + + return ( +
+ {isPartial ? Some environments are still reporting. Totals are partial. : null} + {failed.map((environment) => ( + {environment.label} could not report usage. + ))} + {stale.map((environment) => ( + + {environment.label} runs an older server version and is excluded from totals. + + ))} + {duplicateSources.length > 0 ? ( + + Counted once across environments sharing a transcript directory:{" "} + {duplicateSources.join(", ")} + + ) : null} +
+ ); +} diff --git a/apps/web/src/components/usage/UsageProviderChart.test.ts b/apps/web/src/components/usage/UsageProviderChart.test.ts new file mode 100644 index 00000000000..2b647153f20 --- /dev/null +++ b/apps/web/src/components/usage/UsageProviderChart.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, it } from "vite-plus/test"; + +import { buildDayColumns, niceScale } from "./UsageProviderChart"; + +describe("niceScale", () => { + it("never puts the peak above the top of the scale", () => { + // Regression: an earlier version stopped at the last step below the peak, + // so the tallest day was drawn past the plot and clipped. + for (const peak of [1122.71, 999, 1, 0.04, 1_400_000_000, 37.5, 5000, 100.001]) { + const { max } = niceScale(peak, 4); + expect(max, `peak ${peak}`).toBeGreaterThanOrEqual(peak); + } + }); + + it("starts at zero and ends at the maximum", () => { + const { max, ticks } = niceScale(1122.71, 4); + + expect(ticks[0]).toBe(0); + expect(ticks[ticks.length - 1]).toBeCloseTo(max, 6); + }); + + it("uses evenly spaced 1/2/5 steps", () => { + const { ticks } = niceScale(1122.71, 4); + const steps = ticks.slice(1).map((tick, index) => tick - (ticks[index] ?? 0)); + + for (const step of steps) expect(step).toBeCloseTo(steps[0] ?? 0, 6); + const [first = 0] = steps; + const normalized = first / 10 ** Math.floor(Math.log10(first)); + expect([1, 2, 5, 10]).toContain(Math.round(normalized)); + }); + + it("keeps the tick count near the requested resolution", () => { + const { ticks } = niceScale(1122.71, 4); + expect(ticks.length).toBeGreaterThanOrEqual(3); + expect(ticks.length).toBeLessThanOrEqual(7); + }); + + it("degrades to a single zero tick with no data", () => { + expect(niceScale(0, 4)).toEqual({ max: 0, ticks: [0] }); + }); +}); + +describe("buildDayColumns", () => { + const days = ["2026-08-01", "2026-08-02", "2026-08-03"]; + const byDay = new Map([ + [ + "2026-08-01", + { + day: "2026-08-01", + costUsd: 30, + totalTokens: 300, + byProvider: new Map([ + ["codex" as const, { costUsd: 10, totalTokens: 100 }], + ["claude" as const, { costUsd: 20, totalTokens: 200 }], + ]), + }, + ], + // 2026-08-02 is deliberately absent: a day with no activity. + [ + "2026-08-03", + { + day: "2026-08-03", + costUsd: 5, + totalTokens: 50, + byProvider: new Map([["claude" as const, { costUsd: 5, totalTokens: 50 }]]), + }, + ], + ]); + + it("plots each day on its own", () => { + expect(buildDayColumns(days, byDay, "cost").map((column) => column.total)).toEqual([30, 0, 5]); + }); + + it("reads the requested metric", () => { + expect(buildDayColumns(days, byDay, "tokens").map((column) => column.total)).toEqual([ + 300, 0, 50, + ]); + }); + + it("keeps the bands contiguous so the areas stay additive", () => { + for (const column of buildDayColumns(days, byDay, "cost")) { + let expectedBase = 0; + for (const band of column.bands) { + expect(band.base).toBeCloseTo(expectedBase, 9); + expect(band.top).toBeCloseTo(band.base + band.value, 9); + expectedBase = band.top; + } + expect(column.total).toBeCloseTo(expectedBase, 9); + } + }); +}); diff --git a/apps/web/src/components/usage/UsageProviderChart.tsx b/apps/web/src/components/usage/UsageProviderChart.tsx new file mode 100644 index 00000000000..d1ffce25e65 --- /dev/null +++ b/apps/web/src/components/usage/UsageProviderChart.tsx @@ -0,0 +1,411 @@ +import type { UsageProviderKind } from "@t3tools/contracts"; +import { useCallback, useMemo, useRef, useState } from "react"; + +import type { DailyTotals } from "../../usage/usageMerge"; +import { formatDayShort, formatTokens, formatUsd } from "../../usage/usageFormat"; +import { PROVIDER_COLOR, PROVIDER_LABEL, PROVIDER_MARK, PROVIDER_ORDER } from "./usageProviders"; + +const VIEW_WIDTH = 960; +const VIEW_HEIGHT = 260; +const TICK_COUNT = 4; +const PLOT_TOP = 8; + +export type UsageChartMetric = "tokens" | "cost"; + +interface UsageProviderChartProps { + readonly days: readonly string[]; + readonly daily: readonly DailyTotals[]; + readonly metric: UsageChartMetric; +} + +/** One day's stacked bands, shared by the paths and the hover readout. */ +export interface DayColumn { + readonly bands: readonly { + readonly provider: UsageProviderKind; + readonly value: number; + readonly base: number; + readonly top: number; + }[]; + readonly total: number; +} + +interface Point { + readonly x: number; + readonly y: number; +} + +function valueFor( + daily: DailyTotals | undefined, + provider: UsageProviderKind, + metric: UsageChartMetric, +): number { + const entry = daily?.byProvider.get(provider); + if (entry === undefined) return 0; + return metric === "tokens" ? entry.totalTokens : entry.costUsd; +} + +/** + * Monotone cubic tangents (Fritsch-Carlson). + * + * Plain cubic smoothing overshoots on spiky daily data and would dip the area + * below zero between points, which reads as negative spend. This variant is + * shape-preserving, so a smoothed series never leaves the range of its samples. + */ +function monotoneTangents(points: readonly Point[]): readonly number[] { + const count = points.length; + if (count < 2) return [0]; + + const slopes: number[] = []; + for (let index = 0; index < count - 1; index += 1) { + const dx = (points[index + 1]?.x ?? 0) - (points[index]?.x ?? 0); + const dy = (points[index + 1]?.y ?? 0) - (points[index]?.y ?? 0); + slopes.push(dx === 0 ? 0 : dy / dx); + } + + const tangents: number[] = Array.from({ length: count }, () => 0); + tangents[0] = slopes[0] ?? 0; + tangents[count - 1] = slopes[count - 2] ?? 0; + for (let index = 1; index < count - 1; index += 1) { + const previous = slopes[index - 1] ?? 0; + const next = slopes[index] ?? 0; + tangents[index] = previous * next <= 0 ? 0 : (previous + next) / 2; + } + + for (let index = 0; index < count - 1; index += 1) { + const slope = slopes[index] ?? 0; + if (slope === 0) { + tangents[index] = 0; + tangents[index + 1] = 0; + continue; + } + const a = (tangents[index] ?? 0) / slope; + const b = (tangents[index + 1] ?? 0) / slope; + const magnitude = a * a + b * b; + if (magnitude > 9) { + const scale = 3 / Math.sqrt(magnitude); + tangents[index] = scale * a * slope; + tangents[index + 1] = scale * b * slope; + } + } + + return tangents; +} + +/** One cubic segment of a smoothed boundary. */ +interface CurveSegment { + readonly from: Point; + readonly c1: Point; + readonly c2: Point; + readonly to: Point; +} + +/** Smoothed polyline through `points`, as explicit cubic control points. */ +function smoothCurve(points: readonly Point[]): readonly CurveSegment[] { + if (points.length < 2) return []; + const tangents = monotoneTangents(points); + const segments: CurveSegment[] = []; + + for (let index = 0; index < points.length - 1; index += 1) { + const from = points[index]; + const to = points[index + 1]; + if (from === undefined || to === undefined) continue; + const dx = to.x - from.x; + segments.push({ + from, + c1: { x: from.x + dx / 3, y: from.y + ((tangents[index] ?? 0) * dx) / 3 }, + c2: { x: to.x - dx / 3, y: to.y - ((tangents[index + 1] ?? 0) * dx) / 3 }, + to, + }); + } + return segments; +} + +function curvePath(segments: readonly CurveSegment[], startCommand: "M" | "L"): string { + const first = segments[0]; + if (first === undefined) return ""; + let path = `${startCommand}${first.from.x.toFixed(2)},${first.from.y.toFixed(2)}`; + for (const segment of segments) { + path += ` C${segment.c1.x.toFixed(2)},${segment.c1.y.toFixed(2)} ${segment.c2.x.toFixed(2)},${segment.c2.y.toFixed(2)} ${segment.to.x.toFixed(2)},${segment.to.y.toFixed(2)}`; + } + return path; +} + +/** + * The same curve walked end to start. A cubic reverses exactly by swapping its + * control points, so this traces the identical geometry. + * + * Bands must use this rather than re-smoothing their base points in reverse: + * the tangent clamp in `monotoneTangents` runs left to right, so smoothing is + * not perfectly symmetric under reversal, and independently smoothed edges of + * adjacent bands could hairline-gap or overlap. Sharing one curve per stack + * boundary makes that geometrically impossible. + */ +function reversedCurvePath(segments: readonly CurveSegment[], startCommand: "M" | "L"): string { + const last = segments[segments.length - 1]; + if (last === undefined) return ""; + let path = `${startCommand}${last.to.x.toFixed(2)},${last.to.y.toFixed(2)}`; + for (let index = segments.length - 1; index >= 0; index -= 1) { + const segment = segments[index]; + if (segment === undefined) continue; + path += ` C${segment.c2.x.toFixed(2)},${segment.c2.y.toFixed(2)} ${segment.c1.x.toFixed(2)},${segment.c1.y.toFixed(2)} ${segment.from.x.toFixed(2)},${segment.from.y.toFixed(2)}`; + } + return path; +} + +/** + * Builds a scale whose maximum is a readable 1/2/5 x 10^n step at or above the + * peak. + * + * Rounding the maximum *up* is the point: stopping at the last step below the + * peak leaves the tallest day drawn past the top of the plot, where it is + * clipped. + */ +export function niceScale(peak: number, count: number): { max: number; ticks: readonly number[] } { + if (peak <= 0) return { max: 0, ticks: [0] }; + + const rawStep = peak / count; + const magnitude = 10 ** Math.floor(Math.log10(rawStep)); + const normalized = rawStep / magnitude; + const step = (normalized > 5 ? 10 : normalized > 2 ? 5 : normalized > 1 ? 2 : 1) * magnitude; + + const max = Math.ceil(peak / step) * step; + const ticks: number[] = []; + for (let value = 0; value <= max + step * 1e-6; value += step) ticks.push(value); + return { max, ticks }; +} + +/** + * Turns the merged daily totals into stacked bands, one column per day. + * + * The chart paths and the hover readout both consume this, so the number under + * the cursor is by construction the number that was plotted rather than a + * second derivation that can drift from it. + */ +export function buildDayColumns( + days: readonly string[], + byDay: ReadonlyMap, + metric: UsageChartMetric, +): readonly DayColumn[] { + return days.map((day) => { + const entry = byDay.get(day); + let stackTop = 0; + const bands = PROVIDER_ORDER.map((provider) => { + const value = valueFor(entry, provider, metric); + const base = stackTop; + stackTop += value; + return { provider, value, base, top: stackTop }; + }); + return { bands, total: stackTop }; + }); +} + +export function UsageProviderChart({ days, daily, metric }: UsageProviderChartProps) { + const byDay = useMemo(() => new Map(daily.map((entry) => [entry.day, entry])), [daily]); + const [hoverIndex, setHoverIndex] = useState(null); + const plotRef = useRef(null); + + const { paths, ticks, stepX, toY, series } = useMemo(() => { + if (days.length === 0) { + return { + paths: [], + ticks: [0] as readonly number[], + stepX: 0, + toY: () => VIEW_HEIGHT, + series: [] as readonly DayColumn[], + }; + } + + const stacked = buildDayColumns(days, byDay, metric); + + const peak = stacked.reduce((max, column) => Math.max(max, column.total), 0); + const { max, ticks: tickValues } = niceScale(peak, TICK_COUNT); + const step = days.length === 1 ? 0 : VIEW_WIDTH / (days.length - 1); + // Reserve a sliver above the top gridline so the series stroke, which is + // drawn at constant screen width, is not shaved off at a peak. + const toY = (value: number) => + max === 0 ? VIEW_HEIGHT : VIEW_HEIGHT - (value / max) * (VIEW_HEIGHT - PLOT_TOP); + + // One smoothed curve per stack boundary (baseline, then each provider's + // cumulative top). Band k is the region between boundary k and k+1, both + // drawn from these shared control points. + const boundaries = [ + stacked.map((_, dayIndex) => ({ x: dayIndex * step, y: toY(0) })), + ...PROVIDER_ORDER.map((_, providerIndex) => + stacked.map((column, dayIndex) => ({ + x: dayIndex * step, + y: toY(column.bands[providerIndex]?.top ?? 0), + })), + ), + ].map(smoothCurve); + + const built = PROVIDER_ORDER.map((provider, providerIndex) => { + const top = boundaries[providerIndex + 1] ?? []; + const base = boundaries[providerIndex] ?? []; + return { + provider, + area: `${curvePath(top, "M")} ${reversedCurvePath(base, "L")} Z`, + line: curvePath(top, "M"), + }; + }); + + return { paths: built, ticks: tickValues, stepX: step, toY, series: stacked }; + }, [byDay, days, metric]); + + const format = metric === "tokens" ? formatTokens : formatUsd; + + const handleMove = useCallback( + (event: React.MouseEvent) => { + const bounds = plotRef.current?.getBoundingClientRect(); + if (bounds === undefined || bounds.width === 0 || days.length === 0) return; + const fraction = (event.clientX - bounds.left) / bounds.width; + const index = Math.round(fraction * (days.length - 1)); + setHoverIndex(Math.min(days.length - 1, Math.max(0, index))); + }, + [days.length], + ); + + const hoveredDay = hoverIndex === null ? undefined : days[hoverIndex]; + const hoveredColumn = hoverIndex === null ? undefined : series[hoverIndex]; + const hoverLeft = days.length <= 1 ? 0 : ((hoverIndex ?? 0) / (days.length - 1)) * 100; + + return ( +
+
+ {/* Axis labels sit outside the plot so they stay aligned to gridlines. */} +
+ {ticks.map((tick) => ( + + {tick === 0 ? "0" : format(tick)} + + ))} +
+ +
setHoverIndex(null)} + > + + {ticks.map((tick) => { + const y = toY(tick); + return ( + + ); + })} + + {paths.map(({ provider, area, line }) => ( + + + + + ))} + + {hoverIndex === null ? null : ( + + )} + + + {hoveredDay === undefined ? null : ( +
60 ? "translateX(-100%)" : "translateX(0)", + }} + > +
{formatDayShort(hoveredDay)}
+ {PROVIDER_ORDER.map((provider) => { + const Mark = PROVIDER_MARK[provider]; + return ( +
+ + + {PROVIDER_LABEL[provider]} + + + {format( + hoveredColumn?.bands.find((band) => band.provider === provider)?.value ?? 0, + )} + +
+ ); + })} +
+ Total + + {format(hoveredColumn?.total ?? 0)} + +
+
+ )} +
+
+ +
+ {days[0] === undefined ? "" : formatDayShort(days[0])} + + {days[Math.floor(days.length / 2)] === undefined + ? "" + : formatDayShort(days[Math.floor(days.length / 2)] ?? "")} + + + {days[days.length - 1] === undefined ? "" : formatDayShort(days[days.length - 1] ?? "")} + +
+
+ ); +} + +export function UsageChartLegend() { + return ( +
+ {PROVIDER_ORDER.map((provider) => { + // The marks carry the same fills as the bands, so they key the chart + // just as a colour swatch would. + const Mark = PROVIDER_MARK[provider]; + return ( + + + {PROVIDER_LABEL[provider]} + + ); + })} +
+ ); +} diff --git a/apps/web/src/components/usage/usageProviders.ts b/apps/web/src/components/usage/usageProviders.ts new file mode 100644 index 00000000000..5356f96edc7 --- /dev/null +++ b/apps/web/src/components/usage/usageProviders.ts @@ -0,0 +1,32 @@ +import type { UsageProviderKind } from "@t3tools/contracts"; + +import { ClaudeAI, type Icon, OpenAI } from "../Icons"; + +/** + * Stacking and table order. Codex sits under Claude Code so the larger band + * reads as the top surface, matching the reference layout. + */ +export const PROVIDER_ORDER: readonly UsageProviderKind[] = ["codex", "claude"]; + +export const PROVIDER_LABEL: Record = { + claude: "Claude Code", + codex: "Codex", +}; + +/** Claude's brand orange against a neutral white for Codex. */ +export const PROVIDER_COLOR: Record = { + claude: "#d97757", + codex: "#e6e6e6", +}; + +/** + * Brand marks, reused from the provider picker. + * + * These ship their own fills (`#d97757` for Claude, white on dark for OpenAI), + * which are the same colours as the chart bands, so swapping a colour dot for a + * mark keeps the series association intact rather than trading it away. + */ +export const PROVIDER_MARK: Record = { + claude: ClaudeAI, + codex: OpenAI, +}; diff --git a/apps/web/src/routeTree.gen.ts b/apps/web/src/routeTree.gen.ts index 3da96820ab9..eb31a8de91c 100644 --- a/apps/web/src/routeTree.gen.ts +++ b/apps/web/src/routeTree.gen.ts @@ -9,6 +9,7 @@ // Additionally, you should also exclude this file from your linter and/or formatter to prevent it from being checked or modified. import { Route as rootRouteImport } from './routes/__root' +import { Route as UsageRouteImport } from './routes/usage' import { Route as SettingsRouteImport } from './routes/settings' import { Route as PairRouteImport } from './routes/pair' import { Route as ConnectRouteImport } from './routes/connect' @@ -26,6 +27,11 @@ import { Route as ConnectCallbackRouteImport } from './routes/connect_.callback' import { Route as ChatDraftDraftIdRouteImport } from './routes/_chat.draft.$draftId' import { Route as ChatEnvironmentIdThreadIdRouteImport } from './routes/_chat.$environmentId.$threadId' +const UsageRoute = UsageRouteImport.update({ + id: '/usage', + path: '/usage', + getParentRoute: () => rootRouteImport, +} as any) const SettingsRoute = SettingsRouteImport.update({ id: '/settings', path: '/settings', @@ -112,6 +118,7 @@ export interface FileRoutesByFullPath { '/connect': typeof ConnectRoute '/pair': typeof PairRoute '/settings': typeof SettingsRouteWithChildren + '/usage': typeof UsageRoute '/connect/callback': typeof ConnectCallbackRoute '/settings/appearance': typeof SettingsAppearanceRoute '/settings/archived': typeof SettingsArchivedRoute @@ -128,6 +135,7 @@ export interface FileRoutesByTo { '/connect': typeof ConnectRoute '/pair': typeof PairRoute '/settings': typeof SettingsRouteWithChildren + '/usage': typeof UsageRoute '/connect/callback': typeof ConnectCallbackRoute '/settings/appearance': typeof SettingsAppearanceRoute '/settings/archived': typeof SettingsArchivedRoute @@ -147,6 +155,7 @@ export interface FileRoutesById { '/connect': typeof ConnectRoute '/pair': typeof PairRoute '/settings': typeof SettingsRouteWithChildren + '/usage': typeof UsageRoute '/connect_/callback': typeof ConnectCallbackRoute '/settings/appearance': typeof SettingsAppearanceRoute '/settings/archived': typeof SettingsArchivedRoute @@ -167,6 +176,7 @@ export interface FileRouteTypes { | '/connect' | '/pair' | '/settings' + | '/usage' | '/connect/callback' | '/settings/appearance' | '/settings/archived' @@ -183,6 +193,7 @@ export interface FileRouteTypes { | '/connect' | '/pair' | '/settings' + | '/usage' | '/connect/callback' | '/settings/appearance' | '/settings/archived' @@ -201,6 +212,7 @@ export interface FileRouteTypes { | '/connect' | '/pair' | '/settings' + | '/usage' | '/connect_/callback' | '/settings/appearance' | '/settings/archived' @@ -220,11 +232,19 @@ export interface RootRouteChildren { ConnectRoute: typeof ConnectRoute PairRoute: typeof PairRoute SettingsRoute: typeof SettingsRouteWithChildren + UsageRoute: typeof UsageRoute ConnectCallbackRoute: typeof ConnectCallbackRoute } declare module '@tanstack/react-router' { interface FileRoutesByPath { + '/usage': { + id: '/usage' + path: '/usage' + fullPath: '/usage' + preLoaderRoute: typeof UsageRouteImport + parentRoute: typeof rootRouteImport + } '/settings': { id: '/settings' path: '/settings' @@ -385,6 +405,7 @@ const rootRouteChildren: RootRouteChildren = { ConnectRoute: ConnectRoute, PairRoute: PairRoute, SettingsRoute: SettingsRouteWithChildren, + UsageRoute: UsageRoute, ConnectCallbackRoute: ConnectCallbackRoute, } export const routeTree = rootRouteImport diff --git a/apps/web/src/routes/usage.tsx b/apps/web/src/routes/usage.tsx new file mode 100644 index 00000000000..c617e434b2e --- /dev/null +++ b/apps/web/src/routes/usage.tsx @@ -0,0 +1,7 @@ +import { createFileRoute } from "@tanstack/react-router"; + +import { UsagePage } from "../components/usage/UsagePage"; + +export const Route = createFileRoute("/usage")({ + component: UsagePage, +}); diff --git a/apps/web/src/state/usage.ts b/apps/web/src/state/usage.ts new file mode 100644 index 00000000000..57114ade152 --- /dev/null +++ b/apps/web/src/state/usage.ts @@ -0,0 +1,126 @@ +/** + * Multi-environment usage state. + * + * Every connected environment answers the same typed query; the client merges + * the results. Raw transcripts never leave the machine that produced them. + * + * @module state/usage + */ +import { useAtomValue } from "@effect/atom-react"; +import { + USAGE_CONTRACT_VERSION, + type EnvironmentId, + type UsageSummary, + type UsageSummaryInput, +} from "@t3tools/contracts"; +import * as Option from "effect/Option"; +import { AsyncResult, Atom } from "effect/unstable/reactivity"; +import { useCallback, useMemo } from "react"; + +import { mergeUsage, type EnvironmentUsage, type MergedUsage } from "../usage/usageMerge"; +import { appAtomRegistry } from "../rpc/atomRegistry"; +import { environmentPresentations } from "./presentation"; +import { serverEnvironment } from "./server"; + +export interface EnvironmentUsageStatus { + readonly environmentId: EnvironmentId; + readonly label: string; + readonly isPending: boolean; + readonly error: string | null; + readonly summary: UsageSummary | null; +} + +/** + * Reads every environment's summary for one window. + * + * Keyed by the serialised window so switching ranges does not thrash the atom + * cache, and so each environment's query is shared with any other reader of the + * same window. + */ +const usageByWindowAtom = Atom.family((windowKey: string) => + Atom.make((get): readonly EnvironmentUsageStatus[] => { + const input = JSON.parse(windowKey) as UsageSummaryInput; + const presentations = get(environmentPresentations.presentationsAtom); + + const statuses: EnvironmentUsageStatus[] = []; + for (const [environmentId, presentation] of presentations) { + const result = get(serverEnvironment.usageSummary({ environmentId, input })); + statuses.push({ + environmentId, + label: presentation.entry.target.label, + isPending: result.waiting, + error: result._tag === "Failure" ? "This environment could not report usage." : null, + summary: Option.getOrNull(AsyncResult.value(result)), + }); + } + return statuses; + }).pipe(Atom.withLabel(`web-usage:window:${windowKey}`)), +); + +export interface UsageView { + readonly merged: MergedUsage; + readonly environments: readonly EnvironmentUsageStatus[]; + /** True until at least one environment has answered. */ + readonly isPending: boolean; + /** + * True while environments that have not failed are still answering. Failed + * environments are reported through their own error rows: totals will not + * improve by waiting on them, so they must not read as "still reporting". + */ + readonly isPartial: boolean; + readonly refresh: () => void; +} + +export function useUsage(input: UsageSummaryInput): UsageView { + const windowKey = useMemo( + () => + JSON.stringify({ + sinceDay: input.sinceDay, + untilDay: input.untilDay, + timeZone: input.timeZone, + }), + [input.sinceDay, input.untilDay, input.timeZone], + ); + const atom = usageByWindowAtom(windowKey); + const environments = useAtomValue(atom); + + // Refreshing only the derived atom would re-read the per-environment SWR + // queries within their stale window and change nothing. Refresh each + // environment's query so the button always rescans. + const refresh = useCallback(() => { + const input = JSON.parse(windowKey) as UsageSummaryInput; + for (const environment of environments) { + appAtomRegistry.refresh( + serverEnvironment.usageSummary({ environmentId: environment.environmentId, input }), + ); + } + }, [environments, windowKey]); + + const merged = useMemo(() => { + const answered: EnvironmentUsage[] = environments.flatMap((environment) => + environment.summary === null + ? [] + : [ + { + environmentId: environment.environmentId, + label: environment.label, + summary: environment.summary, + }, + ], + ); + return mergeUsage(answered, USAGE_CONTRACT_VERSION); + }, [environments]); + + const answeredCount = environments.filter((environment) => environment.summary !== null).length; + const stillReporting = environments.filter( + (environment) => environment.summary === null && environment.error === null, + ).length; + + return { + merged, + environments, + isPending: answeredCount === 0 && stillReporting > 0, + isPartial: answeredCount > 0 && stillReporting > 0, + refresh, + }; +} diff --git a/apps/web/src/usage/usageFormat.ts b/apps/web/src/usage/usageFormat.ts new file mode 100644 index 00000000000..c7c21605837 --- /dev/null +++ b/apps/web/src/usage/usageFormat.ts @@ -0,0 +1,107 @@ +/** + * Display formatting for the usage page. + * + * @module usageFormat + */ +import { UsageDay, type UsageSummaryInput } from "@t3tools/contracts"; + +const CURRENCY = new Intl.NumberFormat("en-US", { + style: "currency", + currency: "USD", + minimumFractionDigits: 2, + maximumFractionDigits: 2, +}); + +const INTEGER = new Intl.NumberFormat("en-US"); + +export function formatUsd(value: number): string { + return CURRENCY.format(value); +} + +export function formatCount(value: number): string { + return INTEGER.format(Math.round(value)); +} + +/** + * Compacts a token count to three significant figures with a unit suffix, so + * columns of numbers line up at a glance (`19.9B`, `76.7M`, `804K`). + */ +export function formatTokens(value: number): string { + const abs = Math.abs(value); + if (abs >= 1e12) return `${trim(value / 1e12)}T`; + if (abs >= 1e9) return `${trim(value / 1e9)}B`; + if (abs >= 1e6) return `${trim(value / 1e6)}M`; + if (abs >= 1e3) return `${trim(value / 1e3)}K`; + return INTEGER.format(Math.round(value)); +} + +function trim(value: number): string { + const abs = Math.abs(value); + const digits = abs >= 100 ? 0 : abs >= 10 ? 1 : 2; + return value.toFixed(digits).replace(/\.0+$/, ""); +} + +export function formatPercent(share: number, digits = 1): string { + return `${(share * 100).toFixed(digits)}%`; +} + +/** `2026-08-07` to `Aug 7`. */ +export function formatDayShort(day: string): string { + const [year, month, dayOfMonth] = day.split("-").map((part) => Number(part)); + if (year === undefined || month === undefined || dayOfMonth === undefined) return day; + const MONTHS = [ + "Jan", + "Feb", + "Mar", + "Apr", + "May", + "Jun", + "Jul", + "Aug", + "Sep", + "Oct", + "Nov", + "Dec", + ]; + return `${MONTHS[month - 1] ?? ""} ${dayOfMonth}`; +} + +/** Inclusive day list between two `YYYY-MM-DD` bounds. */ +export function enumerateDays(sinceDay: string, untilDay: string): readonly string[] { + const days: string[] = []; + const start = Date.parse(`${sinceDay}T00:00:00Z`); + const end = Date.parse(`${untilDay}T00:00:00Z`); + if (Number.isNaN(start) || Number.isNaN(end) || end < start) return days; + + for (let cursor = start; cursor <= end; cursor += 86_400_000) { + days.push(new Date(cursor).toISOString().slice(0, 10)); + } + return days; +} + +/** + * The window the page requests, expressed in the viewer's own time zone so days + * line up with what they actually experienced. + */ +export function makeWindow(days: number, now = new Date()): UsageSummaryInput { + const timeZone = Intl.DateTimeFormat().resolvedOptions().timeZone || "UTC"; + const format = new Intl.DateTimeFormat("en-CA", { + timeZone, + year: "numeric", + month: "2-digit", + day: "2-digit", + }); + const untilDay = format.format(now); + // Subtracting fixed milliseconds from `now` lands on the wrong calendar day + // around a DST transition. Only "today" needs the zone; the window start is + // pure calendar arithmetic on that day, done in UTC where days are uniform. + const [year = 0, month = 1, dayOfMonth = 1] = untilDay + .split("-") + .map((part) => Number.parseInt(part, 10)); + const start = new Date(Date.UTC(year, month - 1, dayOfMonth - (days - 1))); + return { + sinceDay: UsageDay.make(start.toISOString().slice(0, 10)), + untilDay: UsageDay.make(untilDay), + timeZone, + }; +} diff --git a/apps/web/src/usage/usageMerge.test.ts b/apps/web/src/usage/usageMerge.test.ts new file mode 100644 index 00000000000..7e44631cf5d --- /dev/null +++ b/apps/web/src/usage/usageMerge.test.ts @@ -0,0 +1,258 @@ +import { + USAGE_CONTRACT_VERSION, + type EnvironmentId, + type UsageBucket, + type UsageDay, + type UsageProviderKind, + type UsageSummary, +} from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { mergeUsage, type EnvironmentUsage } from "./usageMerge"; + +function bucket(overrides: Partial = {}): UsageBucket { + return { + day: "2026-08-07" as UsageDay, + provider: "claude", + model: "claude-fable-5", + totals: { + uncachedInputTokens: 100, + cachedInputTokens: 1000, + cacheCreationTokens: 10, + outputTokens: 50, + reasoningTokens: 0, + }, + costUsd: 10, + cacheSavingsUsd: 2, + costSource: "modelPriced", + records: 5, + unpricedRecords: 0, + sessions: 1, + ...overrides, + }; +} + +function summary( + buckets: readonly UsageBucket[], + sources: readonly { + provider: UsageProviderKind; + hostId: string; + homePath: string; + volumeId?: string; + distinctSessions?: number; + }[], + contractVersion: number = USAGE_CONTRACT_VERSION, +): UsageSummary { + return { + contractVersion, + readAt: "2026-08-07T00:00:00.000Z", + timeZone: "UTC", + sinceDay: "2026-08-01" as UsageDay, + untilDay: "2026-08-31" as UsageDay, + buckets, + sources: sources.map((source) => ({ + fingerprint: { + hostId: source.hostId, + provider: source.provider, + resolvedHomePath: source.homePath, + volumeId: source.volumeId ?? `vol-${source.hostId}`, + }, + status: "ok" as const, + scannedFiles: 1, + skippedFiles: 0, + malformedRecords: 0, + distinctSessions: source.distinctSessions ?? 1, + message: null, + })), + pricing: { status: "fresh", source: "litellm", fetchedAt: null, knownModels: 10 }, + scanDurationMs: 1, + }; +} + +function environment(id: string, usageSummary: UsageSummary): EnvironmentUsage { + return { environmentId: id as EnvironmentId, label: id, summary: usageSummary }; +} + +describe("mergeUsage", () => { + it("sums environments that read different transcript directories", () => { + const merged = mergeUsage( + [ + environment( + "env-a", + summary([bucket()], [{ provider: "claude", hostId: "mac", homePath: "/a/.claude" }]), + ), + environment( + "env-b", + summary([bucket()], [{ provider: "claude", hostId: "linux", homePath: "/b/.claude" }]), + ), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.costUsd).toBe(20); + expect(merged.records).toBe(10); + expect(merged.duplicateSources).toHaveLength(0); + }); + + it("counts a shared transcript directory once", () => { + // Two worktree servers on one machine resolve the same provider home. + const shared = { provider: "claude" as const, hostId: "mac", homePath: "/home/theo/.claude" }; + const merged = mergeUsage( + [ + environment("env-a", summary([bucket()], [shared])), + environment("env-b", summary([bucket()], [shared])), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.costUsd).toBe(10); + expect(merged.records).toBe(5); + expect(merged.sessions).toBe(1); + expect(merged.duplicateSources).toHaveLength(1); + expect(merged.contributingEnvironments).toEqual(["env-a"]); + }); + + it("drops only the duplicated provider, keeping the environment's other one", () => { + const sharedClaude = { + provider: "claude" as const, + hostId: "mac", + homePath: "/home/theo/.claude", + }; + const merged = mergeUsage( + [ + environment("env-a", summary([bucket()], [sharedClaude])), + environment( + "env-b", + summary( + [bucket(), bucket({ provider: "codex", model: "gpt-5.6-sol", costUsd: 4 })], + [sharedClaude, { provider: "codex", hostId: "mac", homePath: "/home/theo/.codex" }], + ), + ), + ], + USAGE_CONTRACT_VERSION, + ); + + // env-b's claude bucket is dropped, its codex bucket survives. + expect(merged.costUsd).toBe(14); + expect(merged.providers.map((provider) => provider.provider).sort()).toEqual([ + "claude", + "codex", + ]); + }); + + it("excludes an environment reporting an older contract version", () => { + const merged = mergeUsage( + [ + environment( + "env-a", + summary([bucket()], [{ provider: "claude", hostId: "mac", homePath: "/a" }]), + ), + environment( + "env-b", + summary( + [bucket()], + [{ provider: "claude", hostId: "linux", homePath: "/b" }], + USAGE_CONTRACT_VERSION - 1, + ), + ), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.costUsd).toBe(10); + expect(merged.staleEnvironments).toEqual(["env-b"]); + }); + + it("derives provider shares and cost quality", () => { + const merged = mergeUsage( + [ + environment( + "env-a", + summary( + [ + bucket({ costUsd: 75 }), + bucket({ provider: "codex", model: "gpt-5.6-sol", costUsd: 25, unpricedRecords: 5 }), + ], + [ + { provider: "claude", hostId: "mac", homePath: "/a/.claude" }, + { provider: "codex", hostId: "mac", homePath: "/a/.codex" }, + ], + ), + ), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.providers[0]?.provider).toBe("claude"); + expect(merged.providers[0]?.costShare).toBeCloseTo(0.75, 5); + expect(merged.costQuality.unpricedShare).toBeCloseTo(0.5, 5); + expect(merged.costQuality.cacheSavingsUsd).toBe(4); + }); + + it("keeps two machines apart when hostname and home path collide", () => { + // Every Mac resolves /Users/theo/.claude, so a hostname clash used to make + // one machine's usage vanish. Filesystem identity separates them. + const shape = { provider: "claude" as const, hostId: "mac", homePath: "/Users/theo/.claude" }; + const merged = mergeUsage( + [ + environment("env-a", summary([bucket()], [{ ...shape, volumeId: "16777220:1234" }])), + environment("env-b", summary([bucket()], [{ ...shape, volumeId: "16777221:9999" }])), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.costUsd).toBe(20); + expect(merged.duplicateSources).toHaveLength(0); + }); + + it("still collapses two servers reading the same directory", () => { + const same = { + provider: "claude" as const, + hostId: "mac", + homePath: "/Users/theo/.claude", + volumeId: "16777220:1234", + }; + const merged = mergeUsage( + [ + environment("env-a", summary([bucket()], [same])), + environment("env-b", summary([bucket()], [same])), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.costUsd).toBe(10); + expect(merged.duplicateSources).toHaveLength(1); + }); + + it("totals sessions from per-directory distinct counts, not per-bucket sums", () => { + // One session that spans two days appears in two buckets. Summing bucket + // sessions would say 2; the source's distinct count says 1. + const merged = mergeUsage( + [ + environment( + "env-a", + summary( + [bucket({ day: "2026-08-06" as UsageDay }), bucket({ day: "2026-08-07" as UsageDay })], + [ + { + provider: "claude", + hostId: "mac", + homePath: "/a/.claude", + distinctSessions: 1, + }, + ], + ), + ), + ], + USAGE_CONTRACT_VERSION, + ); + + expect(merged.sessions).toBe(1); + }); + + it("returns empty totals with no environments", () => { + const merged = mergeUsage([], USAGE_CONTRACT_VERSION); + expect(merged.costUsd).toBe(0); + expect(merged.daily).toHaveLength(0); + }); +}); diff --git a/apps/web/src/usage/usageMerge.ts b/apps/web/src/usage/usageMerge.ts new file mode 100644 index 00000000000..fd73c0a31c0 --- /dev/null +++ b/apps/web/src/usage/usageMerge.ts @@ -0,0 +1,353 @@ +/** + * Merges per-environment usage summaries into the single view the page renders. + * + * Pure, so the de-duplication and derivation rules can be tested without a + * connected environment. + * + * @module usageMerge + */ +import type { + EnvironmentId, + UsageBucket, + UsageProviderKind, + UsageSourceFingerprint, + UsageSummary, +} from "@t3tools/contracts"; + +export interface EnvironmentUsage { + readonly environmentId: EnvironmentId; + readonly label: string; + readonly summary: UsageSummary; +} + +export interface ProviderTotals { + readonly provider: UsageProviderKind; + readonly costUsd: number; + readonly totalTokens: number; + readonly records: number; + readonly costShare: number; + readonly tokenShare: number; +} + +export interface ModelTotals { + readonly model: string; + readonly provider: UsageProviderKind; + readonly costUsd: number; + readonly totalTokens: number; + readonly records: number; + readonly costShare: number; +} + +export interface DailyTotals { + readonly day: string; + readonly costUsd: number; + readonly totalTokens: number; + readonly byProvider: ReadonlyMap; +} + +export interface CostQuality { + readonly providerReportedShare: number; + readonly modelPricedShare: number; + readonly unpricedShare: number; + readonly cacheSavingsUsd: number; +} + +export interface MergedUsage { + readonly costUsd: number; + readonly uncachedInputTokens: number; + readonly cachedInputTokens: number; + readonly cacheCreationTokens: number; + readonly outputTokens: number; + readonly reasoningTokens: number; + readonly totalTokens: number; + readonly records: number; + readonly sessions: number; + readonly providers: readonly ProviderTotals[]; + readonly models: readonly ModelTotals[]; + readonly daily: readonly DailyTotals[]; + readonly costQuality: CostQuality; + /** Environments whose data was dropped as a duplicate of another's. */ + readonly duplicateSources: readonly string[]; + readonly contributingEnvironments: readonly EnvironmentId[]; + readonly staleEnvironments: readonly EnvironmentId[]; +} + +/** + * Two sources are the same physical transcript directory only when host, + * provider, path and filesystem identity all agree. + * + * `volumeId` is what stops two machines that happen to share a hostname and a + * home path, which is every Mac in a fleet, from collapsing into one source and + * having one of them silently dropped. + */ +function fingerprintKey(fingerprint: UsageSourceFingerprint): string { + return [ + fingerprint.hostId, + fingerprint.provider, + fingerprint.resolvedHomePath, + fingerprint.volumeId, + ].join(" "); +} + +/** + * Decides which environment owns each physical transcript directory. + * + * Several environments on one machine (worktree servers, for instance) resolve + * the same provider home and would otherwise double count every token. The + * first environment in a stable order claims a fingerprint; the rest have that + * provider's buckets dropped. Environments are sorted by id so the winner does + * not change between renders. + */ +function claimSources(environments: readonly EnvironmentUsage[]): { + readonly ownerByFingerprint: ReadonlyMap; + readonly duplicates: readonly string[]; +} { + const ownerByFingerprint = new Map(); + const duplicates: string[] = []; + + const ordered = [...environments].sort((a, b) => a.environmentId.localeCompare(b.environmentId)); + + for (const environment of ordered) { + for (const source of environment.summary.sources) { + if (source.status === "missing") continue; + const key = fingerprintKey(source.fingerprint); + if (ownerByFingerprint.has(key)) { + duplicates.push(`${environment.label}: ${source.fingerprint.resolvedHomePath}`); + continue; + } + ownerByFingerprint.set(key, environment.environmentId); + } + } + + return { ownerByFingerprint, duplicates }; +} + +/** Sources this environment owns after fingerprint claims, plus their buckets. */ +function ownedContribution( + environment: EnvironmentUsage, + ownerByFingerprint: ReadonlyMap, +): { readonly buckets: readonly UsageBucket[]; readonly sessions: number } { + const ownedProviders = new Set(); + let sessions = 0; + for (const source of environment.summary.sources) { + if (source.status === "missing") continue; + const key = fingerprintKey(source.fingerprint); + if (ownerByFingerprint.get(key) === environment.environmentId) { + ownedProviders.add(source.fingerprint.provider); + // Distinct within a directory. Summing per-bucket session counts instead + // would count a session once per day and model it spans. + sessions += source.distinctSessions; + } + } + return { + buckets: environment.summary.buckets.filter((bucket) => ownedProviders.has(bucket.provider)), + sessions, + }; +} + +function bucketTokens(bucket: UsageBucket): number { + // reasoningTokens is a subset of outputTokens and must not be added again. + return ( + bucket.totals.uncachedInputTokens + + bucket.totals.cachedInputTokens + + bucket.totals.cacheCreationTokens + + bucket.totals.outputTokens + ); +} + +const EMPTY_MERGED: MergedUsage = { + costUsd: 0, + uncachedInputTokens: 0, + cachedInputTokens: 0, + cacheCreationTokens: 0, + outputTokens: 0, + reasoningTokens: 0, + totalTokens: 0, + records: 0, + sessions: 0, + providers: [], + models: [], + daily: [], + costQuality: { + providerReportedShare: 0, + modelPricedShare: 0, + unpricedShare: 0, + cacheSavingsUsd: 0, + }, + duplicateSources: [], + contributingEnvironments: [], + staleEnvironments: [], +}; + +/** + * Merges every connected environment's summary. + * + * `expectedContractVersion` guards against an environment running older server + * code: rather than blocking the page, its data is excluded and its id is + * reported so the UI can say coverage is partial. + */ +export function mergeUsage( + environments: readonly EnvironmentUsage[], + expectedContractVersion: number, +): MergedUsage { + if (environments.length === 0) return EMPTY_MERGED; + + const current: EnvironmentUsage[] = []; + const staleEnvironments: EnvironmentId[] = []; + for (const environment of environments) { + if (environment.summary.contractVersion === expectedContractVersion) { + current.push(environment); + } else { + staleEnvironments.push(environment.environmentId); + } + } + + const { ownerByFingerprint, duplicates } = claimSources(current); + + let costUsd = 0; + let uncachedInputTokens = 0; + let cachedInputTokens = 0; + let cacheCreationTokens = 0; + let outputTokens = 0; + let reasoningTokens = 0; + let records = 0; + let sessions = 0; + let cacheSavingsUsd = 0; + let providerReportedRecords = 0; + let unpricedRecords = 0; + + const providerAccumulator = new Map< + UsageProviderKind, + { costUsd: number; totalTokens: number; records: number } + >(); + const modelAccumulator = new Map< + string, + { provider: UsageProviderKind; costUsd: number; totalTokens: number; records: number } + >(); + const dailyAccumulator = new Map< + string, + { + costUsd: number; + totalTokens: number; + byProvider: Map; + } + >(); + const contributingEnvironments: EnvironmentId[] = []; + + for (const environment of current) { + const { buckets, sessions: environmentSessions } = ownedContribution( + environment, + ownerByFingerprint, + ); + if (buckets.length > 0) contributingEnvironments.push(environment.environmentId); + sessions += environmentSessions; + + for (const bucket of buckets) { + const tokens = bucketTokens(bucket); + + costUsd += bucket.costUsd; + cacheSavingsUsd += bucket.cacheSavingsUsd; + uncachedInputTokens += bucket.totals.uncachedInputTokens; + cachedInputTokens += bucket.totals.cachedInputTokens; + cacheCreationTokens += bucket.totals.cacheCreationTokens; + outputTokens += bucket.totals.outputTokens; + reasoningTokens += bucket.totals.reasoningTokens; + records += bucket.records; + unpricedRecords += bucket.unpricedRecords; + if (bucket.costSource === "providerReported") providerReportedRecords += bucket.records; + + const provider = providerAccumulator.get(bucket.provider) ?? { + costUsd: 0, + totalTokens: 0, + records: 0, + }; + provider.costUsd += bucket.costUsd; + provider.totalTokens += tokens; + provider.records += bucket.records; + providerAccumulator.set(bucket.provider, provider); + + const modelKey = `${bucket.provider} ${bucket.model}`; + const model = modelAccumulator.get(modelKey) ?? { + provider: bucket.provider, + costUsd: 0, + totalTokens: 0, + records: 0, + }; + model.costUsd += bucket.costUsd; + model.totalTokens += tokens; + model.records += bucket.records; + modelAccumulator.set(modelKey, model); + + const day = dailyAccumulator.get(bucket.day) ?? { + costUsd: 0, + totalTokens: 0, + byProvider: new Map(), + }; + day.costUsd += bucket.costUsd; + day.totalTokens += tokens; + const dayProvider = day.byProvider.get(bucket.provider) ?? { costUsd: 0, totalTokens: 0 }; + dayProvider.costUsd += bucket.costUsd; + dayProvider.totalTokens += tokens; + day.byProvider.set(bucket.provider, dayProvider); + dailyAccumulator.set(bucket.day, day); + } + } + + const totalTokens = uncachedInputTokens + cachedInputTokens + cacheCreationTokens + outputTokens; + + const providers: ProviderTotals[] = [...providerAccumulator.entries()] + .map(([provider, totals]) => ({ + provider, + costUsd: totals.costUsd, + totalTokens: totals.totalTokens, + records: totals.records, + costShare: costUsd === 0 ? 0 : totals.costUsd / costUsd, + tokenShare: totalTokens === 0 ? 0 : totals.totalTokens / totalTokens, + })) + .sort((a, b) => b.costUsd - a.costUsd); + + const models: ModelTotals[] = [...modelAccumulator.entries()] + .map(([key, totals]) => ({ + model: key.slice(key.indexOf(" ") + 1), + provider: totals.provider, + costUsd: totals.costUsd, + totalTokens: totals.totalTokens, + records: totals.records, + costShare: costUsd === 0 ? 0 : totals.costUsd / costUsd, + })) + .sort((a, b) => b.costUsd - a.costUsd || b.totalTokens - a.totalTokens); + + const daily: DailyTotals[] = [...dailyAccumulator.entries()] + .map(([day, totals]) => ({ + day, + costUsd: totals.costUsd, + totalTokens: totals.totalTokens, + byProvider: totals.byProvider, + })) + .sort((a, b) => a.day.localeCompare(b.day)); + + return { + costUsd, + uncachedInputTokens, + cachedInputTokens, + cacheCreationTokens, + outputTokens, + reasoningTokens, + totalTokens, + records, + sessions, + providers, + models, + daily, + costQuality: { + providerReportedShare: records === 0 ? 0 : providerReportedRecords / records, + unpricedShare: records === 0 ? 0 : unpricedRecords / records, + modelPricedShare: + records === 0 ? 0 : (records - providerReportedRecords - unpricedRecords) / records, + cacheSavingsUsd, + }, + duplicateSources: duplicates, + contributingEnvironments, + staleEnvironments, + }; +} diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 8c61a939e9e..f579453c27f 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -707,6 +707,13 @@ export function createServerEnvironmentAtoms( tag: WS_METHODS.serverGetResourceTelemetryHistory, staleTimeMs: 5_000, }), + // A cold transcript scan is measured in seconds, so keep the result around + // long enough that switching windows or re-rendering does not rescan. + usageSummary: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:server:usage-summary", + tag: WS_METHODS.serverGetUsageSummary, + staleTimeMs: 60_000, + }), configProjection, welcome: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:server:welcome", diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index f0ee1889177..6181391eca3 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -28,4 +28,5 @@ export * from "./review.ts"; export * from "./preview.ts"; export * from "./previewAutomation.ts"; export * from "./resourceTelemetry.ts"; +export * from "./usage.ts"; export * from "./rpc.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index db40b10fed9..59255639995 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -152,6 +152,7 @@ import { ResourceTelemetryRetryResult, ResourceTelemetrySnapshot, } from "./resourceTelemetry.ts"; +import { UsageReadError, UsageSummary, UsageSummaryInput } from "./usage.ts"; import { ServerSettings, ServerSettingsError, ServerSettingsPatch } from "./settings.ts"; import { SourceControlCloneRepositoryInput, @@ -244,6 +245,7 @@ export const WS_METHODS = { serverReportClientActivity: "server.reportClientActivity", serverReportHostPowerState: "server.reportHostPowerState", serverGetBackgroundPolicy: "server.getBackgroundPolicy", + serverGetUsageSummary: "server.getUsageSummary", // Cloud environment methods cloudGetRelayClientStatus: "cloud.getRelayClientStatus", @@ -381,6 +383,12 @@ export const WsServerRetryResourceTelemetryRpc = Rpc.make(WS_METHODS.serverRetry error: EnvironmentAuthorizationError, }); +export const WsServerGetUsageSummaryRpc = Rpc.make(WS_METHODS.serverGetUsageSummary, { + payload: UsageSummaryInput, + success: UsageSummary, + error: Schema.Union([EnvironmentAuthorizationError, UsageReadError]), +}); + export const WsServerSignalProcessRpc = Rpc.make(WS_METHODS.serverSignalProcess, { payload: ServerSignalProcessInput, success: ServerSignalProcessResult, @@ -819,6 +827,7 @@ export const WsRpcGroup = RpcGroup.make( WsServerGetProcessResourceHistoryRpc, WsServerGetResourceTelemetryHistoryRpc, WsServerRetryResourceTelemetryRpc, + WsServerGetUsageSummaryRpc, WsServerSignalProcessRpc, WsServerReportClientActivityRpc, WsServerReportHostPowerStateRpc, diff --git a/packages/contracts/src/usage.ts b/packages/contracts/src/usage.ts new file mode 100644 index 00000000000..1aa639fe4a0 --- /dev/null +++ b/packages/contracts/src/usage.ts @@ -0,0 +1,194 @@ +/** + * Usage reporting contract. + * + * Each environment scans the provider CLIs' own on-disk session transcripts + * (`~/.claude/projects/**\/*.jsonl`, `~/.codex/sessions/**\/*.jsonl`) rather than + * relying on T3 Code's own orchestration projections, so usage stays complete + * even for turns that were never driven through T3 Code. This mirrors the + * approach `ccusage` takes. + * + * Environments return pre-aggregated `(day, provider, model)` buckets. Raw + * transcript records never cross the wire. + * + * @module usage + */ +import * as Schema from "effect/Schema"; + +import { NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +/** + * Bumped whenever the shape of {@link UsageSummary} changes incompatibly. The + * client renders partial coverage when an environment reports an older version + * rather than failing the whole page. + */ +export const USAGE_CONTRACT_VERSION = 3 as const; + +export const UsageProviderKind = Schema.Literals(["claude", "codex"]); +export type UsageProviderKind = typeof UsageProviderKind.Type; + +/** + * A calendar day in the reporting time zone, formatted `YYYY-MM-DD`. + * + * Days are bucketed server-side so that a turn always lands on the day the user + * experienced it, not the UTC day. + */ +const USAGE_DAY_PATTERN = /^\d{4}-\d{2}-\d{2}$/; + +export const UsageDay = TrimmedNonEmptyString.check(Schema.isPattern(USAGE_DAY_PATTERN)).pipe( + Schema.brand("UsageDay"), +); +export type UsageDay = typeof UsageDay.Type; + +/** + * Why a bucket's cost is what it is. + * + * - `providerReported` - the transcript carried an explicit cost figure. + * - `modelPriced` - we matched the model against the LiteLLM rate table. + * - `unpriced` - tokens are known, rates are not. Counted in totals, excluded + * from cost. + */ +export const UsageCostSource = Schema.Literals(["providerReported", "modelPriced", "unpriced"]); +export type UsageCostSource = typeof UsageCostSource.Type; + +/** + * Token counts for a bucket. + * + * `cachedInputTokens` and `cacheCreationTokens` are disjoint from + * `uncachedInputTokens`; summing all three gives total input. `reasoningTokens` + * is a *subset* of `outputTokens` (Codex reports it that way, and Anthropic + * folds thinking into output), so it must never be added on top. + */ +export const UsageTokenTotals = Schema.Struct({ + uncachedInputTokens: NonNegativeInt, + cachedInputTokens: NonNegativeInt, + cacheCreationTokens: NonNegativeInt, + outputTokens: NonNegativeInt, + reasoningTokens: NonNegativeInt, +}); +export type UsageTokenTotals = typeof UsageTokenTotals.Type; + +/** + * One `(day, provider, model)` cell. + * + * `costUsd` is the raw API-equivalent cost of these tokens. It is not money + * spent: subscription plans bill separately. `unpricedRecords` counts records + * whose tokens are included in the token totals but which contributed nothing + * to `costUsd`. + */ +export const UsageBucket = Schema.Struct({ + day: UsageDay, + provider: UsageProviderKind, + model: TrimmedNonEmptyString, + totals: UsageTokenTotals, + costUsd: Schema.Number, + /** + * What the cached input would have cost at full input rates minus what it + * actually cost. Requires the rate table, so it is computed alongside cost + * rather than derived on the client. + */ + cacheSavingsUsd: Schema.Number, + costSource: UsageCostSource, + /** Distinct assistant responses, after de-duplication. */ + records: NonNegativeInt, + unpricedRecords: NonNegativeInt, + /** Distinct transcript sessions that contributed to this cell. */ + sessions: NonNegativeInt, +}); +export type UsageBucket = typeof UsageBucket.Type; + +/** + * Identifies the physical transcript directory a source read from. + * + * Two environments on the same machine (worktree servers, for example) resolve + * the same provider home and would otherwise double count. The client drops + * duplicate fingerprints before merging. + */ +export const UsageSourceFingerprint = Schema.Struct({ + hostId: TrimmedNonEmptyString, + provider: UsageProviderKind, + resolvedHomePath: TrimmedNonEmptyString, + /** + * Filesystem identity of the transcript directory, as `device:inode`. + * + * Hostname and path alone are not enough: every Mac in a fleet resolves + * `/Users//.claude`, so two machines that happen to share a hostname + * would look like one source and have their usage silently dropped. The + * device/inode pair is stable for two servers reading the same directory and + * effectively never collides across machines. Empty when it cannot be read. + */ + volumeId: Schema.String, +}); +export type UsageSourceFingerprint = typeof UsageSourceFingerprint.Type; + +export const UsageSourceStatus = Schema.Literals(["ok", "missing", "partial", "failed"]); +export type UsageSourceStatus = typeof UsageSourceStatus.Type; + +export const UsageSource = Schema.Struct({ + fingerprint: UsageSourceFingerprint, + status: UsageSourceStatus, + scannedFiles: NonNegativeInt, + skippedFiles: NonNegativeInt, + /** Records that parsed but carried no recognisable usage payload. */ + malformedRecords: NonNegativeInt, + /** + * Distinct transcript sessions seen under this directory. Buckets also carry + * per-bucket session counts, but a session spans days and models, so summing + * those overcounts; this is the figure clients should total. + */ + distinctSessions: NonNegativeInt, + message: Schema.NullOr(TrimmedNonEmptyString), +}); +export type UsageSource = typeof UsageSource.Type; + +export const UsagePricingStatus = Schema.Literals(["fresh", "cached", "unavailable"]); +export type UsagePricingStatus = typeof UsagePricingStatus.Type; + +/** + * Provenance for the rate table, so the UI can be honest about how good the + * cost figures are. + */ +export const UsagePricing = Schema.Struct({ + status: UsagePricingStatus, + source: TrimmedNonEmptyString, + fetchedAt: Schema.NullOr(Schema.String), + knownModels: NonNegativeInt, +}); +export type UsagePricing = typeof UsagePricing.Type; + +export const UsageSummaryInput = Schema.Struct({ + /** Inclusive first day of the window, in `timeZone`. */ + sinceDay: UsageDay, + /** Inclusive last day of the window, in `timeZone`. */ + untilDay: UsageDay, + /** + * IANA zone the client wants days bucketed in. An offset would be wrong for + * any window that crosses a DST boundary. + */ + timeZone: TrimmedNonEmptyString, +}); +export type UsageSummaryInput = typeof UsageSummaryInput.Type; + +export const UsageSummary = Schema.Struct({ + contractVersion: Schema.Number, + readAt: Schema.String, + timeZone: TrimmedNonEmptyString, + sinceDay: UsageDay, + untilDay: UsageDay, + buckets: Schema.Array(UsageBucket), + sources: Schema.Array(UsageSource), + pricing: UsagePricing, + /** Wall-clock cost of the scan, surfaced in diagnostics. */ + scanDurationMs: NonNegativeInt, +}); +export type UsageSummary = typeof UsageSummary.Type; + +export class UsageReadError extends Schema.TaggedErrorClass()("UsageReadError", { + reason: Schema.Literals(["scanFailed", "invalidWindow"]), + /** Stable, bounded description. The underlying failure travels in `cause`. */ + detail: TrimmedNonEmptyString, + cause: Schema.optional(Schema.Defect()), +}) { + override get message(): string { + return `Usage read failed (${this.reason}): ${this.detail}`; + } +}