/** * UsageService - scans provider transcripts and returns priced usage buckets. * * The scan reads the provider CLIs' own session files (Claude Code, Codex, and * Grok Build) 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, and a file that merely grew resumes * from its cached parse position so only the appended bytes are read. * * @module UsageService */import * as NodeOS from "node:os"; import { USAGE_CONTRACT_VERSION, type ServerSettings as ServerSettingsValue, type UsageProviderKind, type UsageSource, type UsagePricing, type UsageSummary, type UsageSummaryInput, UsageReadError,} from "@t3tools/contracts";import { HostProcessEnvironment } from "@t3tools/shared/hostProcess";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 Deferred from "effect/Deferred";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 * as Semaphore from "effect/Semaphore";import { HttpClient, HttpClientResponse } from "effect/unstable/http"; import { ServerConfig } from "../config.ts";import { expandHomePath } from "../pathExpansion.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 { createOverrideRateTable, 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; /** An explicit refresh ignores the TTL, but not a table fetched this recently. */const RATES_REFRESH_FLOOR_MS = 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;const MAX_HOURLY_WINDOW_MS = 24 * 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<typeof RatesCacheFile.Type>),);const encodeRatesCache = Schema.encodeEffect( Schema.fromJsonString(RatesCacheFile as unknown as Schema.Codec<typeof RatesCacheFile.Type>),); /** 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<unknown>);const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson); export class UsageService extends Context.Service< UsageService, { readonly readSummary: (input: UsageSummaryInput) => Effect.Effect<UsageSummary, UsageReadError>; /** Refetches the rate table ahead of its TTL. See `ensureRates`. */ readonly refreshRates: Effect.Effect<UsagePricing>; }>()("t3/usage/UsageService") {} const EMPTY_PRICING: UsagePricing = { status: "unavailable", source: LITELLM_RATES_URL, fetchedAt: null, knownModels: 0,}; /** 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: EMPTY_PRICING, scanDurationMs: 0, }), refreshRates: Effect.succeed(EMPTY_PRICING), }),); 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 hostEnvironment = yield* HostProcessEnvironment; 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: UsagePricing["status"] = "unavailable"; // One fetch at a time. A burst of refreshes from several clients waits on // the first fetch and then sees a table young enough to skip its own. const ratesLock = yield* Semaphore.make(1); const pricing = (): UsagePricing => ({ status: ratesStatus, source: LITELLM_RATES_URL, fetchedAt: ratesFetchedAtMs === null ? null : DateTime.formatIso(DateTime.makeUnsafe(ratesFetchedAtMs)), knownModels: rates.size, }); /** * 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. `force` refetches inside the TTL so a model that * LiteLLM added since the last fetch gets priced now. */ const loadRates = Effect.fn("UsageService.loadRates")(function* (force: boolean) { const now = yield* Clock.currentTimeMillis; const maxAgeMs = force ? RATES_REFRESH_FLOOR_MS : RATES_TTL_MS; if (ratesFetchedAtMs !== null && now - ratesFetchedAtMs < maxAgeMs) 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 < maxAgeMs) 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), ); }); const ensureRates = (force: boolean) => ratesLock.withPermit(loadRates(force)); const refreshRates = ensureRates(true).pipe( Effect.map(pricing), Effect.withSpan("UsageService.refreshRates"), ); /** * 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"); }); // A settings failure must not silently discard custom rates or transcript homes. const readSettings = settingsService.getSettings.pipe( Effect.catchCause( (cause) => new UsageReadError({ reason: "scanFailed", detail: "Server settings could not be read.", cause: Cause.squash(cause), }), ), ); /** Resolves the transcript directory for each provider. */ const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* ( settings: ServerSettingsValue, ) { const claudeHome = yield* resolveClaudeHomePath(settings.providers.claudeAgent); const claudeDir = yield* resolveClaudeTranscriptDir(claudeHome); const codexLayout = yield* resolveCodexHomeLayout(settings.providers.codex); // Grok Settings only expose the binary path; home is `$GROK_HOME` or `~/.grok`. // Empty/whitespace GROK_HOME must fall back: coalescing alone would scan cwd. const grokHomeEnv = hostEnvironment["GROK_HOME"]?.trim() ?? ""; const grokHome = grokHomeEnv.length > 0 ? path.resolve(expandHomePath(grokHomeEnv)) : path.join(NodeOS.homedir(), ".grok"); return [ { provider: "claude" as const, dir: claudeDir }, { provider: "codex" as const, dir: path.join(codexLayout.sharedHomePath, "sessions") }, { provider: "grok" as const, dir: path.join(grokHome, "sessions"), fileName: "updates.jsonl", }, ]; }); /** * 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. * * A file that only grew re-parses from the cached position, so an actively * written multi-hundred-megabyte rollout costs its appended bytes per scan * rather than a full re-read. The reader verifies the position's guard bytes * and silently restarts from byte 0 when they no longer match. */ const readFileRecords = ( filePath: string, size: number, mtimeMs: number, provider: UsageProviderKind, ): Effect.Effect<readonly UsageRecord[]> => 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.tailRecords.length === 0 ? cached.records : [...cached.records, ...cached.tailRecords]; } // Only a strictly grown file may resume. Same size with a new mtime, or // a shrunken file, means rewritten content; re-parse it whole. const resumeFrom = cached !== undefined && cached.provider === provider && size > cached.size ? cached.position : undefined; const parsed = yield* Effect.promise(() => readTranscriptRecords(filePath, provider, resumeFrom), ); // 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. One // seen set spans the cached base, the new lines, and the tail so a // resumed parse dedupes exactly like a full one. const base = parsed.resumed && cached !== undefined ? cached.records : []; const seen = new Set<string>(); const records = dedupeWithinFile([...base, ...parsed.records], seen); const tailRecords = dedupeWithinFile(parsed.tailRecords, seen); fileCache.set(filePath, { size, mtimeMs, provider, records, tailRecords, position: parsed.position, }); cacheDirty = true; return tailRecords.length === 0 ? records : [...records, ...tailRecords]; }); /** One provider directory's walk and parse, before rates are involved. */ interface ScannedDir { readonly provider: UsageProviderKind; readonly dir: string; readonly volumeId: string; /** Parsed records per file, or `null` when the directory does not exist. */ readonly files: | readonly { readonly path: string; readonly records: readonly UsageRecord[] }[] | null; } const collectDirs = Effect.fn("UsageService.collectDirs")(function* ( windowStartMs: number, settings: ServerSettingsValue, ) { // The home resolvers ask for `Path` themselves; satisfy them from the // instance we already hold so the scan stays context-free. const dirs = yield* resolveTranscriptDirs(settings).pipe( Effect.provideService(Path.Path, path), ); const scanned: ScannedDir[] = []; for (const { provider, dir, fileName } of dirs) { const volumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir)); const exists = yield* fileSystem .exists(dir) .pipe(Effect.catchCause(() => Effect.succeed(false))); if (!exists) { scanned.push({ provider, dir, volumeId, files: null }); continue; } const files = yield* Effect.promise(() => listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }), ); const parsedFiles: { path: string; records: readonly UsageRecord[] }[] = []; for (const file of files) { const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider); parsedFiles.push({ path: file.path, records }); } scanned.push({ provider, dir, volumeId, files: parsedFiles }); } return scanned; }); const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, ) { if (input.sinceDay > input.untilDay) { return yield* new UsageReadError({ reason: "invalidWindow", detail: `sinceDay '${input.sinceDay}' is after untilDay '${input.untilDay}'`, }); } let hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null = null; if (input.resolution === "hour") { const sinceTime = input.sinceTime === undefined ? Option.none() : DateTime.make(input.sinceTime); const untilTime = input.untilTime === undefined ? Option.none() : DateTime.make(input.untilTime); if (Option.isNone(sinceTime) || Option.isNone(untilTime)) { return yield* new UsageReadError({ reason: "invalidWindow", detail: "Hourly usage requires valid sinceTime and untilTime instants", }); } const sinceTimeMs = DateTime.toEpochMillis(sinceTime.value); const untilTimeMs = DateTime.toEpochMillis(untilTime.value); const durationMs = untilTimeMs - sinceTimeMs; if (durationMs <= 0 || durationMs > MAX_HOURLY_WINDOW_MS) { return yield* new UsageReadError({ reason: "invalidWindow", detail: "Hourly usage window must be greater than zero and at most 24 hours", }); } hourlyWindow = { sinceTimeMs, untilTimeMs }; } const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; const hostId = NodeOS.hostname(); 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 = (hourlyWindow?.sinceTimeMs ?? DateTime.toEpochMillis(windowStart.value)) - MTIME_SLACK_MS; // Pricing only matters once records are aggregated, so the rate table // loads while transcripts stream instead of gating them: a cold rates // fetch on a slow network no longer delays the scan by its own timeout. const [, scannedDirs] = yield* Effect.all( [ensureRates(false), collectDirs(windowStartMs, settings)], { concurrency: 2 }, ); const aggregator = new UsageAggregator({ timeZone: input.timeZone, sinceDay: input.sinceDay, untilDay: input.untilDay, resolution: input.resolution ?? "day", ...hourlyWindow, rates, priceOverrides: createOverrideRateTable(settings.usagePriceOverrides), }); const sources: UsageSource[] = []; const livePaths = new Set<string>(); const walkedRoots: string[] = []; for (const { provider, dir, volumeId, files } of scannedDirs) { if (files === null) { 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); let scannedFiles = 0; let skippedFiles = 0;Lines 1–500 of 612. Load more
/** * UsageService - scans provider transcripts and returns priced usage buckets. * * The scan reads the provider CLIs' own session files (Claude Code, Codex, and * Grok Build) 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, and a file that merely grew resumes * from its cached parse position so only the appended bytes are read. * * @module UsageService */ import * as NodeOS from "node:os"; import { USAGE_CONTRACT_VERSION, type ServerSettings as ServerSettingsValue, type UsageProviderKind, type UsageSource, type UsagePricing, type UsageSummary, type UsageSummaryInput, UsageReadError, } from "@t3tools/contracts"; import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; 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 Deferred from "effect/Deferred"; 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 * as Semaphore from "effect/Semaphore"; import { HttpClient, HttpClientResponse } from "effect/unstable/http"; import { ServerConfig } from "../config.ts"; import { expandHomePath } from "../pathExpansion.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 { createOverrideRateTable, 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; /** An explicit refresh ignores the TTL, but not a table fetched this recently. */ const RATES_REFRESH_FLOOR_MS = 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; const MAX_HOURLY_WINDOW_MS = 24 * 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<typeof RatesCacheFile.Type>), ); const encodeRatesCache = Schema.encodeEffect( Schema.fromJsonString(RatesCacheFile as unknown as Schema.Codec<typeof RatesCacheFile.Type>), ); /** 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<unknown>); const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson); const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson); export class UsageService extends Context.Service< UsageService, { readonly readSummary: (input: UsageSummaryInput) => Effect.Effect<UsageSummary, UsageReadError>; /** Refetches the rate table ahead of its TTL. See `ensureRates`. */ readonly refreshRates: Effect.Effect<UsagePricing>; } >()("t3/usage/UsageService") {} const EMPTY_PRICING: UsagePricing = { status: "unavailable", source: LITELLM_RATES_URL, fetchedAt: null, knownModels: 0, }; /** 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: EMPTY_PRICING, scanDurationMs: 0, }), refreshRates: Effect.succeed(EMPTY_PRICING), }), ); 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 hostEnvironment = yield* HostProcessEnvironment; 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: UsagePricing["status"] = "unavailable"; // One fetch at a time. A burst of refreshes from several clients waits on // the first fetch and then sees a table young enough to skip its own. const ratesLock = yield* Semaphore.make(1); const pricing = (): UsagePricing => ({ status: ratesStatus, source: LITELLM_RATES_URL, fetchedAt: ratesFetchedAtMs === null ? null : DateTime.formatIso(DateTime.makeUnsafe(ratesFetchedAtMs)), knownModels: rates.size, }); /** * 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. `force` refetches inside the TTL so a model that * LiteLLM added since the last fetch gets priced now. */ const loadRates = Effect.fn("UsageService.loadRates")(function* (force: boolean) { const now = yield* Clock.currentTimeMillis; const maxAgeMs = force ? RATES_REFRESH_FLOOR_MS : RATES_TTL_MS; if (ratesFetchedAtMs !== null && now - ratesFetchedAtMs < maxAgeMs) 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 < maxAgeMs) 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), ); }); const ensureRates = (force: boolean) => ratesLock.withPermit(loadRates(force)); const refreshRates = ensureRates(true).pipe( Effect.map(pricing), Effect.withSpan("UsageService.refreshRates"), ); /** * 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"); }); // A settings failure must not silently discard custom rates or transcript homes. const readSettings = settingsService.getSettings.pipe( Effect.catchCause( (cause) => new UsageReadError({ reason: "scanFailed", detail: "Server settings could not be read.", cause: Cause.squash(cause), }), ), ); /** Resolves the transcript directory for each provider. */ const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* ( settings: ServerSettingsValue, ) { const claudeHome = yield* resolveClaudeHomePath(settings.providers.claudeAgent); const claudeDir = yield* resolveClaudeTranscriptDir(claudeHome); const codexLayout = yield* resolveCodexHomeLayout(settings.providers.codex); // Grok Settings only expose the binary path; home is `$GROK_HOME` or `~/.grok`. // Empty/whitespace GROK_HOME must fall back: coalescing alone would scan cwd. const grokHomeEnv = hostEnvironment["GROK_HOME"]?.trim() ?? ""; const grokHome = grokHomeEnv.length > 0 ? path.resolve(expandHomePath(grokHomeEnv)) : path.join(NodeOS.homedir(), ".grok"); return [ { provider: "claude" as const, dir: claudeDir }, { provider: "codex" as const, dir: path.join(codexLayout.sharedHomePath, "sessions") }, { provider: "grok" as const, dir: path.join(grokHome, "sessions"), fileName: "updates.jsonl", }, ]; }); /** * 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. * * A file that only grew re-parses from the cached position, so an actively * written multi-hundred-megabyte rollout costs its appended bytes per scan * rather than a full re-read. The reader verifies the position's guard bytes * and silently restarts from byte 0 when they no longer match. */ const readFileRecords = ( filePath: string, size: number, mtimeMs: number, provider: UsageProviderKind, ): Effect.Effect<readonly UsageRecord[]> => 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.tailRecords.length === 0 ? cached.records : [...cached.records, ...cached.tailRecords]; } // Only a strictly grown file may resume. Same size with a new mtime, or // a shrunken file, means rewritten content; re-parse it whole. const resumeFrom = cached !== undefined && cached.provider === provider && size > cached.size ? cached.position : undefined; const parsed = yield* Effect.promise(() => readTranscriptRecords(filePath, provider, resumeFrom), ); // 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. One // seen set spans the cached base, the new lines, and the tail so a // resumed parse dedupes exactly like a full one. const base = parsed.resumed && cached !== undefined ? cached.records : []; const seen = new Set<string>(); const records = dedupeWithinFile([...base, ...parsed.records], seen); const tailRecords = dedupeWithinFile(parsed.tailRecords, seen); fileCache.set(filePath, { size, mtimeMs, provider, records, tailRecords, position: parsed.position, }); cacheDirty = true; return tailRecords.length === 0 ? records : [...records, ...tailRecords]; }); /** One provider directory's walk and parse, before rates are involved. */ interface ScannedDir { readonly provider: UsageProviderKind; readonly dir: string; readonly volumeId: string; /** Parsed records per file, or `null` when the directory does not exist. */ readonly files: | readonly { readonly path: string; readonly records: readonly UsageRecord[] }[] | null; } const collectDirs = Effect.fn("UsageService.collectDirs")(function* ( windowStartMs: number, settings: ServerSettingsValue, ) { // The home resolvers ask for `Path` themselves; satisfy them from the // instance we already hold so the scan stays context-free. const dirs = yield* resolveTranscriptDirs(settings).pipe( Effect.provideService(Path.Path, path), ); const scanned: ScannedDir[] = []; for (const { provider, dir, fileName } of dirs) { const volumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir)); const exists = yield* fileSystem .exists(dir) .pipe(Effect.catchCause(() => Effect.succeed(false))); if (!exists) { scanned.push({ provider, dir, volumeId, files: null }); continue; } const files = yield* Effect.promise(() => listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }), ); const parsedFiles: { path: string; records: readonly UsageRecord[] }[] = []; for (const file of files) { const records = yield* readFileRecords(file.path, file.size, file.mtimeMs, provider); parsedFiles.push({ path: file.path, records }); } scanned.push({ provider, dir, volumeId, files: parsedFiles }); } return scanned; }); const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, ) { if (input.sinceDay > input.untilDay) { return yield* new UsageReadError({ reason: "invalidWindow", detail: `sinceDay '${input.sinceDay}' is after untilDay '${input.untilDay}'`, }); } let hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null = null; if (input.resolution === "hour") { const sinceTime = input.sinceTime === undefined ? Option.none() : DateTime.make(input.sinceTime); const untilTime = input.untilTime === undefined ? Option.none() : DateTime.make(input.untilTime); if (Option.isNone(sinceTime) || Option.isNone(untilTime)) { return yield* new UsageReadError({ reason: "invalidWindow", detail: "Hourly usage requires valid sinceTime and untilTime instants", }); } const sinceTimeMs = DateTime.toEpochMillis(sinceTime.value); const untilTimeMs = DateTime.toEpochMillis(untilTime.value); const durationMs = untilTimeMs - sinceTimeMs; if (durationMs <= 0 || durationMs > MAX_HOURLY_WINDOW_MS) { return yield* new UsageReadError({ reason: "invalidWindow", detail: "Hourly usage window must be greater than zero and at most 24 hours", }); } hourlyWindow = { sinceTimeMs, untilTimeMs }; } const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; const hostId = NodeOS.hostname(); 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 = (hourlyWindow?.sinceTimeMs ?? DateTime.toEpochMillis(windowStart.value)) - MTIME_SLACK_MS; // Pricing only matters once records are aggregated, so the rate table // loads while transcripts stream instead of gating them: a cold rates // fetch on a slow network no longer delays the scan by its own timeout. const [, scannedDirs] = yield* Effect.all( [ensureRates(false), collectDirs(windowStartMs, settings)], { concurrency: 2 }, ); const aggregator = new UsageAggregator({ timeZone: input.timeZone, sinceDay: input.sinceDay, untilDay: input.untilDay, resolution: input.resolution ?? "day", ...hourlyWindow, rates, priceOverrides: createOverrideRateTable(settings.usagePriceOverrides), }); const sources: UsageSource[] = []; const livePaths = new Set<string>(); const walkedRoots: string[] = []; for (const { provider, dir, volumeId, files } of scannedDirs) { if (files === null) { 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); let scannedFiles = 0; let skippedFiles = 0;