1import {
2 type ProviderInstanceConfig,
3 ProviderInstanceId,
4 USAGE_CONTRACT_VERSION,
5 type ServerSettings as ServerSettingsValue,
6 type UsageProviderKind,
7 type UsageSource,
8 type UsagePricing,
9 type UsageSummary,
10 type UsageSummaryInput,
11 UsageReadError,
12} from "@t3tools/contracts";
13import * as HostProcess from "@t3tools/shared/HostProcess";
14import * as Cause from "effect/Cause";
15import * as Clock from "effect/Clock";
16import * as Context from "effect/Context";
17import * as DateTime from "effect/DateTime";
18import * as Deferred from "effect/Deferred";
19import * as Effect from "effect/Effect";
20import * as Exit from "effect/Exit";
21import * as Fiber from "effect/Fiber";
22import * as FileSystem from "effect/FileSystem";
23import * as Layer from "effect/Layer";
24import * as Option from "effect/Option";
25import * as Path from "effect/Path";
26import * as Schema from "effect/Schema";
27import * as Semaphore from "effect/Semaphore";
28import { HttpClient, HttpClientResponse } from "effect/http";
29
30import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite";
31import { formatMalformedUsageWarning } from "@t3tools/shared/usageFormat";
32import * as ServerConfig from "../config.ts";
33import * as ServerSettings from "../serverSettings.ts";
34import { BUILT_IN_USAGE_DRIVERS, type BuiltInUsageReadersEnv } from "../provider/builtInDrivers.ts";
35import type { ProviderDriver } from "@t3tools/provider-core/server/driver";
36import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment";
37import type {
38 ProviderUsageInstance,
39 TranscriptUsageFormat,
40 UsageRecord,
41} from "@t3tools/provider-core/server/usage";
42import { canonicalModelFilter, resolveModelAliases, UsageAggregator } from "./usageAggregation.ts";
43import { createOverrideRateTable, parseRateTable, type RateTable } from "./usagePricing.ts";
44import {
45 listTranscriptFiles,
46 readDirectoryVolumeId,
47 readTranscriptRecords,
48} from "./usageTranscriptReader.ts";
49import {
50 decodeScanCache,
51 dedupeWithinFile,
52 LEGACY_SCAN_CACHE_FILE_NAMES,
53 makeScanCacheWriter,
54 pruneScanCache,
55 SCAN_CACHE_FILE_NAME,
56 type CachedFile,
57 type ScanCache,
58} from "./usageScanCache.ts";
59
60const LITELLM_RATES_URL =
61 "https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json";
62const LITELLM_RATES_API_URL =
63 "https://api.github.com/repos/BerriAI/litellm/contents/model_prices_and_context_window.json?ref=main";
64
65const RATES_TTL_MS = 24 * 60 * 60 * 1000;
66
67const RATES_REFRESH_FLOOR_MS = 60 * 1000;
68
69// mtime slack includes writes just before local midnight on the window's first day.
70const MTIME_SLACK_MS = 36 * 60 * 60 * 1000;
71const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000;
72
73const CACHE_RETENTION_DAYS = 92;
74const STALE_TEMP_DIRECTORY_MS = 60 * 60 * 1000;
75
76const TRANSCRIPT_READ_CONCURRENCY = 4;
77
78const transcriptReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) =>
79 driver.usage?.kind === "transcripts" ? [{ driver, reader: driver.usage }] : [],
80);
81
82const scanReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) =>
83 driver.usage?.kind === "scan" ? [{ driver, reader: driver.usage }] : [],
84);
85
86const transcriptFormats = new Map(
87 transcriptReaders.map(({ reader }) => [reader.provider, reader.format] as const),
88);
89
90interface TranscriptSource {
91 readonly provider: UsageProviderKind;
92 readonly format: TranscriptUsageFormat<unknown>;
93 readonly dir: string;
94 readonly volumeId: string;
95 readonly fileName?: string;
96}
97
98const RatesCacheFile = Schema.Struct({
99 fetchedAtMs: Schema.Number,
100 document: Schema.Unknown,
101});
102const decodeRatesCache = Schema.decodeUnknownEffect(Schema.fromJsonString(RatesCacheFile));
103const encodeRatesCache = Schema.encodeEffect(Schema.fromJsonString(RatesCacheFile));
104
105const ScanCacheJson = Schema.fromJsonString(Schema.Unknown);
106const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);
107const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson);
108const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String });
109
110function isLaterRead(a: CachedFile, b: CachedFile): boolean {
111 return a.mtimeMs > b.mtimeMs || (a.mtimeMs === b.mtimeMs && a.size > b.size);
112}
113
114const sharedSessionProviders: ReadonlySet<UsageProviderKind> = new Set(
115 transcriptReaders.flatMap(({ reader }) =>
116 reader.format.sharedSessionsAcrossFiles ? [reader.provider] : [],
117 ),
118);
119
120// Only shared sessions need cross-file dedupe; per-file occurrence counts make other keys unique.
121function sharedSessions(
122 files: readonly { readonly records: readonly UsageRecord[] }[],
123): ReadonlySet<string> {
124 const firstFile = new Map<string, number>();
125 const shared = new Set<string>();
126 for (const [index, file] of files.entries()) {
127 let previous = "";
128 for (const { provider, sessionId } of file.records) {
129 if (!sharedSessionProviders.has(provider) || sessionId === previous || sessionId.length === 0)
130 continue;
131 previous = sessionId;
132 const first = firstFile.get(sessionId);
133 if (first === undefined) firstFile.set(sessionId, index);
134 else if (first !== index) shared.add(sessionId);
135 }
136 }
137 return shared;
138}
139const decodeCachedSources = Schema.decodeUnknownOption(
140 Schema.Struct({
141 sources: Schema.Record(Schema.String, CachedSource),
142 coveredSince: Schema.optional(Schema.Record(Schema.String, Schema.Number)),
143 }),
144);
145
146export class UsageService extends Context.Service<
147 UsageService,
148 {
149 readonly readSummary: (input: UsageSummaryInput) => Effect.Effect<UsageSummary, UsageReadError>;
150 readonly refreshRates: Effect.Effect<UsagePricing>;
151 }
152>()("t3/usage/UsageService") {}
153
154export const make = Effect.gen(function* () {
155 const fileSystem = yield* FileSystem.FileSystem;
156 const path = yield* Path.Path;
157 const config = yield* ServerConfig.ServerConfig;
158 const settingsService = yield* ServerSettings.ServerSettingsService;
159 const httpClient = yield* HttpClient.HttpClient;
160 const hostEnvironment = yield* HostProcess.Environment;
161 const hostname = yield* HostProcess.Hostname;
162 const readerContext = yield* Effect.context<BuiltInUsageReadersEnv>();
163
164 const fileCache: ScanCache = new Map();
165 // Reads from concurrent windows can still hold the original record after it is corrected.
166 const resolvedRecords = new WeakMap<UsageRecord, UsageRecord>();
167 const correctedRecord = (record: UsageRecord) => resolvedRecords.get(record) ?? record;
168 const correctedBatch = (records: readonly UsageRecord[]) =>
169 records.some((record) => resolvedRecords.has(record)) ? records.map(correctedRecord) : records;
170 const sourceCache = new Map<string, typeof CachedSource.Type>();
171 // Earliest window start a complete scan read each transcript directory from.
172 const coveredSince = new Map<string, number>();
173 const coverageKey = (source: TranscriptSource) => `${source.provider}\0${source.dir}`;
174 let cacheDirty = false;
175 const cover = (source: TranscriptSource, windowStartMs: number) => {
176 if ((coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs) return;
177 coveredSince.set(coverageKey(source), windowStartMs);
178 cacheDirty = true;
179 };
180 const isWithinDirectory = (filePath: string, dir: string) => {
181 const relative = path.relative(dir, filePath);
182 return relative !== ".." && !relative.startsWith(".." + path.sep) && !path.isAbsolute(relative);
183 };
184
185 const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json");
186 const scanCachePath = path.join(config.stateDir, SCAN_CACHE_FILE_NAME);
187 const writeCacheFile = (filePath: string, contents: string | Iterable<string>) =>
188 writeFileStringAtomically({ filePath, contents }).pipe(
189 Effect.provideService(FileSystem.FileSystem, fileSystem),
190 Effect.provideService(Path.Path, path),
191 );
192 const hostIdPath = path.join(config.baseDir, "usage-host-id");
193 const explicitHostId = hostEnvironment["T3CODE_HOST_ID"]?.trim();
194 const hostId =
195 explicitHostId ||
196 (yield* fileSystem.readFileString(hostIdPath).pipe(
197 Effect.catchIf(
198 (error) => error.reason._tag === "NotFound",
199 () =>
200 Effect.scoped(
201 Effect.gen(function* () {
202 yield* fileSystem.makeDirectory(config.baseDir, { recursive: true });
203 const tempDirectory = yield* fileSystem.makeTempDirectoryScoped({
204 directory: config.baseDir,
205 prefix: "usage-host-id.",
206 });
207 const tempFile = path.join(tempDirectory, "identity");
208 yield* fileSystem.writeFileString(tempFile, `${hostname}\n`);
209 // Publish complete contents without replacing another initializer's identity.
210 yield* fileSystem.link(tempFile, hostIdPath).pipe(
211 Effect.catchIf(
212 (error) => error.reason._tag === "AlreadyExists",
213 () => Effect.void,
214 ),
215 );
216 return yield* fileSystem.readFileString(hostIdPath);
217 }),
218 ),
219 ),
220 Effect.map((value) => value.trim() || hostname),
221 Effect.orElseSucceed(() => hostname),
222 ));
223 let rates: RateTable = new Map();
224 let ratesFetchedAtMs: number | null = null;
225 let ratesLastFailedAtMs: number | null = null;
226 let ratesStatus: UsagePricing["status"] = "unavailable";
227 const ratesLock = yield* Semaphore.make(1);
228
229 const pricing = (): UsagePricing => ({
230 status: ratesStatus,
231 source: LITELLM_RATES_URL,
232 fetchedAt:
233 ratesFetchedAtMs === null ? null : DateTime.formatIso(DateTime.makeUnsafe(ratesFetchedAtMs)),
234 knownModels: rates.size,
235 });
236
237 const fetchRates = (url: string, accept = "application/json") =>
238 httpClient.get(url, { headers: { accept } }).pipe(
239 Effect.flatMap(HttpClientResponse.filterStatusOk),
240 Effect.flatMap((response) => response.json),
241 Effect.map((document) => {
242 const parsed = parseRateTable(document);
243 return parsed.size > 0 ? { document, parsed } : null;
244 }),
245 Effect.timeout(10_000),
246 Effect.catchCause(() => Effect.succeed(null)),
247 );
248
249 // Outside ratesLock, which a slow fetch holds: a provisional answer prices with these.
250 const loadSavedRates = yield* Effect.cached(
251 fileSystem.readFileString(ratesCachePath).pipe(
252 Effect.flatMap((raw) => decodeRatesCache(raw)),
253 Effect.map((saved) => {
254 const parsed = parseRateTable(saved.document);
255 if (parsed.size === 0) return;
256 rates = parsed;
257 ratesFetchedAtMs = saved.fetchedAtMs;
258 ratesStatus = "cached";
259 }),
260 Effect.ignoreCause,
261 ),
262 );
263
264 const loadRates = Effect.fn("UsageService.loadRates")(function* (force: boolean) {
265 const now = yield* Clock.currentTimeMillis;
266 const maxAgeMs = force ? RATES_REFRESH_FLOOR_MS : RATES_TTL_MS;
267 yield* loadSavedRates;
268 if (ratesFetchedAtMs !== null && now - ratesFetchedAtMs < maxAgeMs) return;
269 if (ratesLastFailedAtMs !== null && now - ratesLastFailedAtMs < RATES_REFRESH_FLOOR_MS) return;
270
271 let fetched = yield* fetchRates(LITELLM_RATES_URL);
272 if (fetched === null) {
273 fetched = yield* fetchRates(LITELLM_RATES_API_URL, "application/vnd.github.raw+json");
274 }
275 if (fetched === null) {
276 ratesLastFailedAtMs = yield* Clock.currentTimeMillis;
277 if (rates.size > 0) ratesStatus = "cached";
278 return;
279 }
280
281 rates = fetched.parsed;
282 ratesFetchedAtMs = now;
283 ratesLastFailedAtMs = null;
284 ratesStatus = "fresh";
285
286 yield* encodeRatesCache({ fetchedAtMs: now, document: fetched.document }).pipe(
287 Effect.flatMap((contents) => writeCacheFile(ratesCachePath, contents)),
288 Effect.ignoreCause,
289 );
290 });
291
292 const ensureRates = (force: boolean) => ratesLock.withPermit(loadRates(force));
293
294 const refreshRates = ensureRates(true).pipe(
295 Effect.map(pricing),
296 Effect.withSpan("UsageService.refreshRates"),
297 );
298
299 const readSettings = settingsService.getSettings.pipe(
300 Effect.catchCause(
301 (cause) =>
302 new UsageReadError({
303 reason: "scanFailed",
304 detail: "Server settings could not be read.",
305 cause: Cause.squash(cause),
306 }),
307 ),
308 );
309
310 const usageInstances = Effect.fn("UsageService.usageInstances")(function* <Config>(
311 driver: ProviderDriver<Config, unknown, unknown>,
312 settings: ServerSettingsValue,
313 ) {
314 const entries: Array<
315 readonly [ProviderInstanceId, Pick<ProviderInstanceConfig, "config" | "environment">, boolean]
316 > = Object.entries(settings.providerInstances)
317 .filter(([, instance]) => instance.driver === driver.driverKind)
318 .map(([id, instance]) => [ProviderInstanceId.make(id), instance, true] as const);
319 if (!Object.hasOwn(settings.providerInstances, driver.driverKind)) {
320 entries.push([ProviderInstanceId.make(driver.driverKind), {}, false]);
321 }
322 const decodeConfig = Schema.decodeUnknownOption(driver.configSchema);
323 return yield* Effect.forEach(
324 entries,
325 Effect.fnUntraced(function* ([instanceId, instance, configured]) {
326 const instanceConfig: ProviderUsageInstance<Config> = {
327 instanceId,
328 config: Option.getOrUndefined(decodeConfig(instance.config ?? {})),
329 environment: yield* mergeProviderInstanceEnvironment(
330 instance.environment,
331 hostEnvironment,
332 ),
333 configured,
334 };
335 return instanceConfig;
336 }),
337 );
338 });
339
340 const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* (
341 settings: ServerSettingsValue,
342 retentionCutoffMs: number,
343 awaitRefresh: boolean,
344 ) {
345 const dirs: Array<TranscriptSource> = [];
346 const pending: Array<ScannedDir> = [];
347 const seen = new Set<string>();
348 for (const { driver, reader } of transcriptReaders) {
349 const { provider, format } = reader;
350 const directories = yield* Effect.forEach(
351 yield* usageInstances(driver, settings),
352 (instance) => reader.directories(instance, { awaitRefresh }),
353 );
354 for (const { dir: directory, fileName, refreshing } of directories.flat()) {
355 if (refreshing) {
356 const key = `${provider}\0${directory}`;
357 if (seen.has(key)) continue;
358 seen.add(key);
359 pending.push({ provider, dir: directory, volumeId: "", files: [], refreshing });
360 continue;
361 }
362 const sourceKey = provider + "\0" + directory;
363 const previous = sourceCache.get(sourceKey);
364 const dir = yield* fileSystem
365 .realPath(directory)
366 .pipe(Effect.orElseSucceed(() => previous?.dir ?? directory));
367 const currentVolumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir));
368 const hasRetainedHistory = fileCache
369 .entries()
370 .some(
371 ([filePath, entry]) =>
372 entry.provider === provider &&
373 entry.mtimeMs >= retentionCutoffMs &&
374 (entry.records.length + entry.tailRecords.length > 0 || entry.malformedRecords > 0) &&
375 isWithinDirectory(filePath, dir),
376 );
377 const volumeId =
378 previous?.dir === dir && (hasRetainedHistory || !currentVolumeId)
379 ? previous.volumeId || currentVolumeId
380 : currentVolumeId;
381 if (previous?.dir !== dir || previous.volumeId !== volumeId) {
382 sourceCache.set(sourceKey, { dir, volumeId });
383 cacheDirty = true;
384 }
385 const key = `${provider}\0${dir}`;
386 if (seen.has(key)) continue;
387 seen.add(key);
388 dirs.push({
389 provider,
390 format,
391 dir,
392 volumeId,
393 ...(fileName === undefined ? {} : { fileName }),
394 });
395 }
396 }
397 return { dirs, pending };
398 });
399
400 // Share first loads so concurrent readers cannot scan against a still-empty cache.
401 const ensureScanCacheLoaded = yield* Effect.cached(
402 Effect.gen(function* () {
403 const readDocument = (filePath: string) =>
404 fileSystem.readFileString(filePath).pipe(
405 Effect.flatMap((raw) => decodeScanCacheFile(raw)),
406 Effect.catchCause(() => Effect.succeed(null)),
407 );
408 let document = yield* readDocument(scanCachePath);
409 const current = document !== null;
410 for (const fileName of LEGACY_SCAN_CACHE_FILE_NAMES) {
411 if (document !== null) break;
412 document = yield* readDocument(path.join(config.stateDir, fileName));
413 cacheDirty = document !== null;
414 }
415 if (document === null) return;
416 const decoded = decodeScanCache(document, transcriptFormats);
417 for (const [path, entry] of decoded) fileCache.set(path, entry);
418 const saved = (document as { readonly files?: unknown }).files;
419 const complete =
420 typeof saved === "object" && saved !== null && Object.keys(saved).length === decoded.size;
421 const sources = decodeCachedSources(document);
422 if (Option.isSome(sources)) {
423 for (const [key, source] of Object.entries(sources.value.sources))
424 sourceCache.set(key, source);
425 // Coverage describes the saved files, so it needs every one, in this version's format.
426 if (current && complete && fileCache.size > 0) {
427 for (const [key, since] of Object.entries(sources.value.coveredSince ?? {}))
428 coveredSince.set(key, since);
429 }
430 }
431 }),
432 );
433
434 const writeScanCache = makeScanCacheWriter();
435 // Serialize writes so concurrently finished window scans cannot overwrite newer snapshots.
436 const persistLock = yield* Semaphore.make(1);
437
438 // The saved cache supersedes older versions. Temp folders an aborted write left behind are
439 // removed once idle for an hour, so a write in progress from another server survives.
440 let supersededFilesRemoved = false;
441 const removeSupersededCacheFiles = Effect.gen(function* () {
442 for (const fileName of LEGACY_SCAN_CACHE_FILE_NAMES) {
443 yield* fileSystem.remove(path.join(config.stateDir, fileName)).pipe(Effect.ignore);
444 }
445 const tempPrefixes = [SCAN_CACHE_FILE_NAME, ...LEGACY_SCAN_CACHE_FILE_NAMES].map(
446 (fileName) => `${fileName}.`,
447 );
448 const staleBeforeMs = (yield* Clock.currentTimeMillis) - STALE_TEMP_DIRECTORY_MS;
449 for (const name of yield* fileSystem.readDirectory(config.stateDir)) {
450 if (!tempPrefixes.some((prefix) => name.startsWith(prefix))) continue;
451 const tempPath = path.join(config.stateDir, name);
452 yield* fileSystem.stat(tempPath).pipe(
453 Effect.flatMap((info) => {
454 const mtimeMs = Option.getOrUndefined(info.mtime)?.getTime();
455 return info.type === "Directory" && mtimeMs !== undefined && mtimeMs <= staleBeforeMs
456 ? fileSystem.remove(tempPath, { recursive: true })
457 : Effect.void;
458 }),
459 Effect.ignore,
460 );
461 }
462 }).pipe(Effect.ignore);
463
464 // Clear dirty before encoding to retain concurrent changes; restore it on failure for retry.
465 const persistCaches = Effect.gen(function* () {
466 if (cacheDirty) {
467 cacheDirty = false;
468 yield* Effect.sync(() =>
469 writeScanCache(fileCache, {
470 sources: Object.fromEntries(sourceCache),
471 coveredSince: Object.fromEntries(coveredSince),
472 }),
473 ).pipe(
474 Effect.flatMap((contents) => writeCacheFile(scanCachePath, contents)),
475 Effect.andThen(
476 Effect.suspend(() => {
477 if (supersededFilesRemoved) return Effect.void;
478 supersededFilesRemoved = true;
479 return removeSupersededCacheFiles;
480 }),
481 ),
482 Effect.catchCause(() =>
483 Effect.sync(() => {
484 cacheDirty = true;
485 }),
486 ),
487 );
488 }
489 }).pipe(persistLock.withPermit, Effect.withSpan("UsageService.persistCaches"));
490
491 const pendingPersists = new Set<Fiber.Fiber<void>>();
492 const schedulePersist = Effect.forkDetach(persistCaches).pipe(
493 Effect.map((fiber) => {
494 pendingPersists.add(fiber);
495 fiber.addObserver(() => pendingPersists.delete(fiber));
496 }),
497 );
498 const awaitPersisted = Effect.suspend(() => Fiber.awaitAll([...pendingPersists])).pipe(
499 Effect.asVoid,
500 );
501 yield* Effect.addFinalizer(() => awaitPersisted);
502
503 /**
504 * Reads run concurrently, and the caller stores updates in walk order rather
505 * than completion order: saved records of deleted transcripts aggregate in
506 * cache order, where the first copy of a duplicate decides its bucket. Untraced: a scan
507 * reads thousands of files, so scanTranscriptDir records the counts instead.
508 */
509 const readFileRecords = Effect.fnUntraced(function* (
510 filePath: string,
511 size: number,
512 mtimeMs: number,
513 provider: UsageProviderKind,
514 format: TranscriptUsageFormat<unknown>,
515 ): Effect.fn.Return<{
516 readonly records: readonly UsageRecord[];
517 readonly malformedRecords: number;
518 readonly failed?: true;
519 readonly update?: { readonly entry: CachedFile; readonly replaces: CachedFile | undefined };
520 }> {
521 const cached = fileCache.get(filePath);
522 // Provider is part of the identity: if both providers were ever pointed
523 // at one directory, a hit parsed by the other parser must not be reused.
524 if (
525 cached &&
526 cached.size === size &&
527 cached.mtimeMs === mtimeMs &&
528 cached.provider === provider
529 ) {
530 return {
531 records:
532 cached.tailRecords.length === 0
533 ? cached.records
534 : [...cached.records, ...cached.tailRecords],
535 malformedRecords: cached.malformedRecords,
536 };
537 }
538
539 // Only a strictly grown file may resume. Same size with a new mtime, or
540 // a shrunken file, means rewritten content; re-parse it whole.
541 const resumeFrom =
542 cached !== undefined && cached.provider === provider && size > cached.size
543 ? cached.position
544 : undefined;
545
546 const parsed = yield* Effect.promise(() => readTranscriptRecords(filePath, format, resumeFrom));
547 // A read failure is not an empty transcript: caching it under this
548 // (size, mtime) would silently drop the file's usage until it changes.
549 if (parsed === null)
550 return {
551 records: cached?.provider === provider ? [...cached.records, ...cached.tailRecords] : [],
552 malformedRecords: cached?.provider === provider ? cached.malformedRecords : 0,
553 failed: true,
554 };
555
556 const identity = (record: UsageRecord) =>
557 encodeUsageRecordKey([
558 record.provider,
559 record.sessionId,
560 record.timestampMs,
561 record.model,
562 record.rateModel,
563 record.totals,
564 record.reportedCostUsd,
565 record.dedupeKey,
566 ]);
567 if (!parsed.resumed && cached !== undefined && cached.provider === provider) {
568 // A whole-file re-parse re-emits saved requests unresolved. Keep their saved tier
569 // corrections, since the history that resolved them may no longer be available.
570 // Equal requests keep their own corrections by matching occurrences in order.
571 const saved = new Map<string, UsageRecord[]>();
572 for (const previous of [...cached.records, ...cached.tailRecords]) {
573 const corrected = correctedRecord(previous);
574 const key = identity(corrected);
575 const occurrences = saved.get(key);
576 if (occurrences === undefined) saved.set(key, [corrected]);
577 else occurrences.push(corrected);
578 }
579 for (const record of [...parsed.records, ...parsed.tailRecords]) {
580 const corrected = saved.get(identity(record))?.shift();
581 if (
582 record.unresolvedServiceTierTurnId !== undefined &&
583 corrected !== undefined &&
584 corrected.unresolvedServiceTierTurnId === undefined
585 ) {
586 resolvedRecords.set(record, corrected);
587 }
588 }
589 }
590 if (parsed.resumed && cached !== undefined && cached.tailRecords.length > 0) {
591 // Completing a tentative line reparses it. Reuse its saved correction only when the
592 // same usage is emitted again, without putting inferred tiers in the parser state.
593 const reparsed = [...parsed.records, ...parsed.tailRecords];
594 for (const [index, previous] of cached.tailRecords.entries()) {
595 const record = reparsed[index];
596 const corrected = correctedRecord(previous);
597 if (
598 record?.unresolvedServiceTierTurnId !== undefined &&
599 corrected.unresolvedServiceTierTurnId === undefined &&
600 identity(record) === identity(corrected)
601 ) {
602 resolvedRecords.set(record, corrected);
603 }
604 }
605 }
606 const base = parsed.resumed && cached !== undefined ? cached.records : [];
607 const records = dedupeWithinFile([...base, ...parsed.records]);
608 const tailRecords = parsed.tailRecords;
609
610 return {
611 records: tailRecords.length === 0 ? records : [...records, ...tailRecords],
612 malformedRecords: parsed.malformedRecords,
613 update: {
614 entry: {
615 size,
616 mtimeMs,
617 provider,
618 records,
619 tailRecords,
620 malformedRecords: parsed.malformedRecords,
621 position: parsed.position,
622 },
623 replaces: cached,
624 },
625 };
626 });
627
628 interface ScannedDir {
629 readonly provider: UsageProviderKind;
630 readonly dir: string;
631 readonly volumeId: string;
632 readonly hostId?: string;
633 readonly status?: UsageSource["status"];
634 readonly message?: string;
635 readonly files:
636 | readonly {
637 readonly path: string;
638 readonly records: readonly UsageRecord[];
639 readonly malformedRecords: number;
640 }[]
641 | null;
642 /** Every transcript path the walk listed, in walk order, whatever its mtime. */
643 readonly walked?: readonly string[];
644 readonly refreshing?: true;
645 }
646
647 const scanTranscriptDir = Effect.fn("UsageService.scanTranscriptDir")(function* (
648 source: TranscriptSource,
649 windowStartMs: number,
650 fromMemory: boolean,
651 ) {
652 const { provider, format, dir, volumeId, fileName } = source;
653 const exists = yield* fileSystem
654 .exists(dir)
655 .pipe(Effect.catchCause(() => Effect.succeed(false)));
656 if (!exists) {
657 cover(source, windowStartMs);
658 return { provider, dir, volumeId, files: null } satisfies ScannedDir;
659 }
660 // scanSummary aggregates the saved files, as it does for a deleted directory.
661 if (fromMemory) return { provider, dir, volumeId, files: [] } satisfies ScannedDir;
662 const { files, walked, failedPaths } = yield* Effect.promise(() =>
663 listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }),
664 );
665 const read = yield* Effect.forEach(
666 files,
667 (file) =>
668 readFileRecords(file.path, file.size, file.mtimeMs, provider, format).pipe(
669 Effect.map((result) => ({ path: file.path, ...result })),
670 ),
671 { concurrency: TRANSCRIPT_READ_CONCURRENCY },
672 );
673 const parsedFiles = read.map(({ path, records, malformedRecords, update }) => {
674 if (update === undefined) return { path, records, malformedRecords };
675 // A scan of another window may have cached its own read of this file
676 // meanwhile. Then keep whichever read saw the later file, so a slower
677 // scan never replaces newer usage with older.
678 const current = fileCache.get(path);
679 if (
680 current === update.replaces ||
681 current === undefined ||
682 isLaterRead(update.entry, current)
683 ) {
684 fileCache.set(path, {
685 ...update.entry,
686 records: correctedBatch(update.entry.records),
687 tailRecords: correctedBatch(update.entry.tailRecords),
688 });
689 cacheDirty = true;
690 }
691 return { path, records, malformedRecords };
692 });
693 const livePaths = new Set(files.map((file) => file.path));
694 for (const file of read) {
695 if (file.failed || !file.path.endsWith(".jsonl.zst")) continue;
696 const originalPath = file.path.slice(0, -4);
697 // A successfully read archive replaces the original's saved records and diagnostics.
698 if (!livePaths.has(originalPath) && fileCache.delete(originalPath)) cacheDirty = true;
699 }
700 const unread = failedPaths + read.filter((file) => file.failed).length;
701 if (unread === 0) cover(source, windowStartMs);
702 yield* Effect.annotateCurrentSpan({
703 "usage.files": files.length,
704 "usage.files_parsed": read.filter((file) => file.update !== undefined).length,
705 });
706 return {
707 provider,
708 dir,
709 volumeId,
710 files: parsedFiles,
711 walked,
712 ...(unread > 0
713 ? {
714 status: "partial",
715 message: `${unread} transcript path(s) could not be read; usage may be incomplete.`,
716 }
717 : {}),
718 } satisfies ScannedDir;
719 });
720
721 const collectDirs = Effect.fn("UsageService.collectDirs")(function* (
722 { dirs, pending }: Effect.Success<ReturnType<typeof resolveTranscriptDirs>>,
723 windowStartMs: number,
724 settings: ServerSettingsValue,
725 retentionCutoffMs: number,
726 awaitRefresh: boolean,
727 fromMemory: boolean,
728 ) {
729 const scans = Effect.forEach(
730 scanReaders,
731 ({ driver, reader }) =>
732 usageInstances(driver, settings).pipe(
733 Effect.flatMap((instances) =>
734 reader.scan({ instances, settings, windowStartMs, retentionCutoffMs, awaitRefresh }),
735 ),
736 Effect.flatMap((sources) =>
737 Effect.forEach(sources, ({ volumeId, ...source }) =>
738 Effect.map(
739 volumeId === undefined
740 ? Effect.promise(() => readDirectoryVolumeId(source.dir))
741 : Effect.succeed(volumeId),
742 (resolved): ScannedDir => ({
743 ...source,
744 files: source.files?.map((file) => ({ ...file, malformedRecords: 0 })) ?? null,
745 provider: reader.provider,
746 volumeId: resolved,
747 }),
748 ),
749 ),
750 ),
751 Effect.provideContext(readerContext),
752 ),
753 { concurrency: "unbounded" },
754 );
755
756 // Scan independent sources together, directories sequentially to bound open files; keep order, since the first copy decides a duplicate's bucket.
757 const [transcripts, scanDirs] = yield* Effect.all(
758 [Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs, fromMemory)), scans],
759 { concurrency: "unbounded" },
760 );
761 const scanned: readonly ScannedDir[] = [...transcripts, ...pending, ...scanDirs.flat()];
762 return scanned;
763 });
764
765 const scanSummary = Effect.fn("UsageService.scanSummary")(function* (
766 input: UsageSummaryInput,
767 settings: ServerSettingsValue,
768 provisional: boolean,
769 ) {
770 if (input.sinceDay > input.untilDay) {
771 return yield* new UsageReadError({
772 reason: "invalidWindow",
773 detail: `sinceDay '${input.sinceDay}' is after untilDay '${input.untilDay}'`,
774 });
775 }
776
777 let hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null = null;
778 if (input.resolution === "hour") {
779 const sinceTime =
780 input.sinceTime === undefined ? Option.none() : DateTime.make(input.sinceTime);
781 const untilTime =
782 input.untilTime === undefined ? Option.none() : DateTime.make(input.untilTime);
783 if (Option.isNone(sinceTime) || Option.isNone(untilTime)) {
784 return yield* new UsageReadError({
785 reason: "invalidWindow",
786 detail: "Hourly usage requires valid sinceTime and untilTime instants",
787 });
788 }
789 const sinceTimeMs = DateTime.toEpochMillis(sinceTime.value);
790 const untilTimeMs = DateTime.toEpochMillis(untilTime.value);
791 const durationMs = untilTimeMs - sinceTimeMs;
792 if (durationMs <= 0 || durationMs > MAX_HOURLY_WINDOW_MS) {
793 return yield* new UsageReadError({
794 reason: "invalidWindow",
795 detail: "Hourly usage window must be greater than zero and at most 24 hours",
796 });
797 }
798 hourlyWindow = { sinceTimeMs, untilTimeMs };
799 }
800
801 const startedAtMs = yield* Clock.currentTimeMillis;
802 yield* ensureScanCacheLoaded;
803
804 const windowStart = DateTime.make(`${input.sinceDay}T00:00:00Z`);
805 if (Option.isNone(windowStart)) {
806 return yield* new UsageReadError({
807 reason: "invalidWindow",
808 detail: `sinceDay '${input.sinceDay}' is not a valid date`,
809 });
810 }
811 const windowStartMs =
812 (hourlyWindow?.sinceTimeMs ?? DateTime.toEpochMillis(windowStart.value)) - MTIME_SLACK_MS;
813
814 const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000;
815
816 const awaitRefresh = input.awaitRefresh === true;
817 const collect = Effect.gen(function* () {
818 const resolved = yield* resolveTranscriptDirs(settings, retentionCutoffMs, awaitRefresh).pipe(
819 Effect.provideContext(readerContext),
820 );
821 // A window no complete scan has read would show partial totals, so every directory must cover it.
822 // Saved files are keyed by path, so two providers reading one directory keep only one parse.
823 const fromMemory =
824 provisional &&
825 resolved.dirs.length > 0 &&
826 new Set(resolved.dirs.map((source) => source.dir)).size === resolved.dirs.length &&
827 resolved.dirs.every(
828 (source) => (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs,
829 );
830 const [, scanned] = yield* Effect.all(
831 [
832 fromMemory ? loadSavedRates : provisional ? ensureRates(false) : Effect.void,
833 collectDirs(
834 resolved,
835 windowStartMs,
836 settings,
837 retentionCutoffMs,
838 awaitRefresh,
839 fromMemory,
840 ),
841 ],
842 { concurrency: 2 },
843 );
844 return { scannedDirs: scanned, fromMemory };
845 });
846 const [{ scannedDirs, fromMemory }] = yield* Effect.all(
847 [collect, provisional ? Effect.void : ensureRates(false)],
848 { concurrency: 2 },
849 );
850
851 const aggregator = new UsageAggregator({
852 timeZone: input.timeZone,
853 sinceDay: input.sinceDay,
854 untilDay: input.untilDay,
855 resolution: input.resolution ?? "day",
856 ...hourlyWindow,
857 rates,
858 priceOverrides: createOverrideRateTable(settings.usagePriceOverrides),
859 modelAliases: resolveModelAliases(settings.usageModelAliases),
860 ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }),
861 });
862
863 const sources: UsageSource[] = [];
864
865 // Deleted transcripts' saved usage still aggregates and deduplicates normally.
866 const retainedSinceMs = Math.max(windowStartMs, retentionCutoffMs);
867 // Saved paths are files below a scanned root, so their folder decides; folders repeat thousands of times.
868 const withinDir = (dir: string) => {
869 const foldersWithin = new Map<string, boolean>();
870 return (filePath: string) => {
871 const folder = path.dirname(filePath);
872 let within = foldersWithin.get(folder);
873 if (within === undefined) {
874 within = isWithinDirectory(folder, dir);
875 foldersWithin.set(folder, within);
876 }
877 return within;
878 };
879 };
880 const filesByDir = scannedDirs.map(({ provider, dir, files }) => {
881 const retainedFiles = [...(files ?? [])];
882 const livePaths = new Set(retainedFiles.map((file) => file.path));
883 const isRetained = withinDir(dir);
884 for (const [filePath, entry] of fileCache) {
885 if (
886 entry.provider !== provider ||
887 entry.mtimeMs < retainedSinceMs ||
888 livePaths.has(filePath) ||
889 !isRetained(filePath)
890 )
891 continue;
892 retainedFiles.push({
893 path: filePath,
894 records:
895 entry.tailRecords.length === 0
896 ? entry.records
897 : [...entry.records, ...entry.tailRecords],
898 malformedRecords: entry.malformedRecords,
899 });
900 }
901 return retainedFiles;
902 });
903 // Usage two transcripts share counts in the first one, so saved files take the order an exact
904 // scan reads them in: walked files in walk order, then the directory's deleted files.
905 for (const { provider, dir, walked } of fromMemory ? [] : scannedDirs) {
906 if (walked === undefined) continue;
907 const within = withinDir(dir);
908 const saved = [...fileCache.keys()];
909 const members = new Set(
910 saved.filter(
911 (filePath) => fileCache.get(filePath)!.provider === provider && within(filePath),
912 ),
913 );
914 if (members.size < 2) continue;
915 const listed = new Set(walked);
916 const target = [
917 ...walked.filter((filePath) => members.has(filePath)),
918 ...saved.filter((filePath) => members.has(filePath) && !listed.has(filePath)),
919 ];
920 let next = 0;
921 const ordered = saved.map((filePath) => (members.has(filePath) ? target[next++]! : filePath));
922 if (ordered.every((filePath, index) => filePath === saved[index])) continue;
923 const entries = new Map(fileCache);
924 fileCache.clear();
925 for (const filePath of ordered) fileCache.set(filePath, entries.get(filePath)!);
926 cacheDirty = true;
927 }
928 const windowEndMs =
929 hourlyWindow?.untilTimeMs ??
930 Date.parse(`${input.untilDay}T00:00:00Z`) + 86400000 + MTIME_SLACK_MS;
931 let resolvedAny = false;
932 for (const { reader } of transcriptReaders) {
933 if (reader.resolveServiceTiers === undefined || fromMemory) continue;
934 const pending = new Set<UsageRecord>();
935 for (const [index, files] of filesByDir.entries()) {
936 if (scannedDirs[index]?.provider !== reader.provider) continue;
937 for (const file of files) {
938 for (const raw of file.records) {
939 const record = correctedRecord(raw);
940 if (
941 record.provider === reader.provider &&
942 record.unresolvedServiceTierTurnId !== undefined &&
943 record.timestampMs >= windowStartMs &&
944 record.timestampMs < windowEndMs
945 ) {
946 pending.add(record);
947 }
948 }
949 }
950 }
951 if (pending.size === 0) continue;
952 const resolved = yield* reader
953 .resolveServiceTiers([...pending])
954 .pipe(Effect.provideContext(readerContext));
955 for (const [record, speed] of resolved) {
956 const { unresolvedServiceTierTurnId: _turnId, ...rest } = record;
957 resolvedRecords.set(record, { ...rest, speed });
958 resolvedAny = true;
959 }
960 }
961 if (resolvedAny) {
962 for (const [filePath, entry] of fileCache) {
963 const records = correctedBatch(entry.records);
964 const tailRecords = correctedBatch(entry.tailRecords);
965 if (records === entry.records && tailRecords === entry.tailRecords) continue;
966 // Update the current entry by identity; a slow resolver cannot restore an older file.
967 fileCache.set(filePath, { ...entry, records, tailRecords });
968 cacheDirty = true;
969 }
970 }
971 const crossFileSessions = sharedSessions(filesByDir.flat());
972
973 for (const [
974 index,
975 { provider, dir, volumeId, files, status, message, refreshing, hostId: sourceHostId },
976 ] of scannedDirs.entries()) {
977 let scannedFiles = 0;
978 let skippedFiles = 0;
979 let malformedRecords = 0;
980 // Distinct per directory. Buckets carry per-cell session counts, but a
981 // session spans days and models, so clients total this figure instead.
982 const sessionIds = new Set<string>();
983 let counted = false;
984
985 for (const file of filesByDir[index] ?? []) {
986 malformedRecords += file.malformedRecords;
987 if (file.records.length === 0) {
988 skippedFiles += 1;
989 continue;
990 }
991 scannedFiles += 1;
992 const eventOccurrences = new Map<string, number>();
993 for (const rawRecord of file.records) {
994 const record = correctedRecord(rawRecord);
995 let usageRecord = record;
996 if (
997 sharedSessionProviders.has(record.provider) &&
998 crossFileSessions.has(record.sessionId)
999 ) {
1000 // Dedupe moved copies but retain repeated equal events within a file, including second-precision timestamps.
1001 const key = encodeUsageRecordKey([
1002 record.provider,
1003 record.sessionId,
1004 record.timestampMs,
1005 record.model,
1006 record.totals,
1007 ]);
1008 const occurrence = (eventOccurrences.get(key) ?? 0) + 1;
1009 eventOccurrences.set(key, occurrence);
1010 usageRecord = { ...record, dedupeKey: key + ":" + occurrence };
1011 }
1012 if (aggregator.add(usageRecord, dir)) {
1013 counted = true;
1014 if (record.sessionId.length > 0) sessionIds.add(record.sessionId);
1015 }
1016 }
1017 }
1018
1019 const sourceMessage =
1020 message ?? (files === null ? "No transcript directory on this environment." : null);
1021 sources.push({
1022 fingerprint: { hostId: sourceHostId ?? hostId, provider, resolvedHomePath: dir, volumeId },
1023 status:
1024 files === null && scannedFiles === 0
1025 ? "missing"
1026 : malformedRecords > 0
1027 ? status === undefined || status === "ok"
1028 ? "partial"
1029 : status
1030 : (status ?? "ok"),
1031 scannedFiles,
1032 skippedFiles,
1033 malformedRecords,
1034 distinctSessions: sessionIds.size,
1035 message:
1036 malformedRecords > 0
1037 ? [sourceMessage, formatMalformedUsageWarning(malformedRecords)]
1038 .filter(Boolean)
1039 .join(" ")
1040 : sourceMessage,
1041 // A saved answer misses what changed since its scan, so the client must ask again. An
1042 // empty source stays unmarked: it would only flash a loading row.
1043 ...(refreshing || (fromMemory && counted && transcriptFormats.has(provider))
1044 ? { refreshing: true }
1045 : {}),
1046 });
1047 }
1048
1049 const pruned = pruneScanCache(fileCache, retentionCutoffMs);
1050 if (pruned > 0) {
1051 cacheDirty = true;
1052 // Pruned files are no longer saved, so nothing before the cutoff stays covered.
1053 for (const [key, since] of coveredSince) {
1054 if (since < retentionCutoffMs) coveredSince.set(key, retentionCutoffMs);
1055 }
1056 }
1057
1058 const aggregated = aggregator.finish();
1059 const readAt = yield* DateTime.now;
1060 const finishedAtMs = yield* Clock.currentTimeMillis;
1061
1062 const summary = {
1063 contractVersion: USAGE_CONTRACT_VERSION,
1064 readAt: DateTime.formatIso(readAt),
1065 timeZone: input.timeZone,
1066 sinceDay: input.sinceDay,
1067 untilDay: input.untilDay,
1068 buckets: aggregated.buckets,
1069 sources,
1070 pricing: pricing(),
1071 scanDurationMs: Math.max(0, finishedAtMs - startedAtMs),
1072 ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }),
1073 } satisfies UsageSummary;
1074 return { summary, fromMemory };
1075 });
1076
1077 const readWindow = Effect.fn("UsageService.readWindow")(function* (
1078 input: UsageSummaryInput,
1079 settings: ServerSettingsValue,
1080 provisional: boolean,
1081 ) {
1082 const { summary, fromMemory } = yield* scanSummary(input, settings, provisional);
1083 // With nothing marked refreshing the client would keep a saved answer as final.
1084 return fromMemory && !summary.sources.some((source) => source.refreshing === true)
1085 ? (yield* scanSummary(input, settings, false)).summary
1086 : summary;
1087 });
1088
1089 const inflightScans = new Map<string, Deferred.Deferred<UsageSummary, UsageReadError>>();
1090
1091 const scanKey = (
1092 input: UsageSummaryInput,
1093 settings: ServerSettingsValue,
1094 provisional: boolean,
1095 ): string =>
1096 JSON.stringify([
1097 input.timeZone,
1098 input.sinceDay,
1099 input.untilDay,
1100 input.resolution ?? "day",
1101 input.sinceTime ?? null,
1102 input.untilTime ?? null,
1103 settings,
1104 input.awaitRefresh === true,
1105 provisional,
1106 input.modelFilter ?? null,
1107 ]);
1108
1109 const readSummary = Effect.fn("UsageService.readSummary")(function* (
1110 requested: UsageSummaryInput,
1111 ) {
1112 const { modelFilter: requestedFilter, ...unfiltered } = requested;
1113 const modelFilter = canonicalModelFilter(requestedFilter);
1114 const input = modelFilter === undefined ? unfiltered : { ...unfiltered, modelFilter };
1115 const settings = yield* readSettings;
1116 const provisional = input.provisional === true && input.awaitRefresh !== true;
1117 const key = scanKey(input, settings, provisional);
1118 const deferred = yield* Effect.uninterruptible(
1119 Effect.gen(function* () {
1120 const existing = inflightScans.get(key);
1121 if (existing !== undefined) return existing;
1122
1123 // Register and fork atomically so cancellation cannot leave a Deferred without a scan.
1124 const created = Deferred.makeUnsafe<UsageSummary, UsageReadError>();
1125 inflightScans.set(key, created);
1126 // Detach so caller departure cannot cancel shared scans; finished scans warm the cache.
1127 // Register writes before releasing waiters, but start writing after delivering their summary.
1128 yield* readWindow(input, settings, provisional).pipe(
1129 Effect.onExit((exit) =>
1130 Effect.sync(() => inflightScans.delete(key)).pipe(
1131 Effect.andThen(Exit.isSuccess(exit) ? schedulePersist : Effect.void),
1132 Effect.andThen(Deferred.done(created, exit)),
1133 ),
1134 ),
1135 Effect.forkDetach,
1136 );
1137 return created;
1138 }),
1139 );
1140 return yield* Deferred.await(deferred);
1141 });
1142
1143 return { readSummary, refreshRates, awaitPersisted } as const;
1144});
1145
1146export const layer = Layer.effect(UsageService, make);