clover's git

the-knife

1	/**
2	* UsageService - reads every driver's usage history and returns priced usage
3	* buckets.
4	*
5	* Each built-in driver with a `usage` reader contributes its sources, including
6	* work driven outside T3 Code. A `transcripts` reader names JSONL directories
7	* that this service streams itself; a `scan` reader reads its own sources.
8	*
9	* JSONL transcripts are append-only, so parsed records are memoised per file by
10	* `(size, mtime)`. A cold 30-day scan of ~1.4 GB lands around 2-3 seconds; warm
11	* scans only reparse files that changed, and a file that merely grew resumes
12	* from its cached parse position so only the appended bytes are read.
13	*
14	* A scan reader with a slow source (an account API) may answer from its own
15	* cache and mark itself `refreshing` while a background refresh runs;
16	* `awaitRefresh` waits for that refresh instead.
17	*
18	* @module UsageService
19	*/
20	import * as NodeOS from "node:os";
21
22	import {
23	type ProviderInstanceConfig,
24	ProviderInstanceId,
25	USAGE_CONTRACT_VERSION,
26	type ServerSettings as ServerSettingsValue,
27	type UsageProviderKind,
28	type UsageSource,
29	type UsagePricing,
30	type UsageSummary,
31	type UsageSummaryInput,
32	UsageReadError,
33	} from "@t3tools/contracts";
34	import * as HostProcess from "@t3tools/shared/HostProcess";
35	import * as Cause from "effect/Cause";
36	import * as Clock from "effect/Clock";
37	import * as Context from "effect/Context";
38	import * as DateTime from "effect/DateTime";
39	import * as Deferred from "effect/Deferred";
40	import * as Effect from "effect/Effect";
41	import * as Exit from "effect/Exit";
42	import * as Fiber from "effect/Fiber";
43	import * as FileSystem from "effect/FileSystem";
44	import * as Layer from "effect/Layer";
45	import * as Option from "effect/Option";
46	import * as Path from "effect/Path";
47	import * as Schema from "effect/Schema";
48	import * as Semaphore from "effect/Semaphore";
49	import { HttpClient, HttpClientResponse } from "effect/http";
50
51	import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite";
52	import * as ServerConfig from "../config.ts";
53	import * as ServerSettings from "../serverSettings.ts";
54	import { BUILT_IN_USAGE_DRIVERS, type BuiltInUsageReadersEnv } from "../provider/builtInDrivers.ts";
55	import type { ProviderDriver } from "@t3tools/provider-core/server/driver";
56	import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment";
57	import type {
58	ProviderUsageInstance,
59	TranscriptUsageFormat,
60	UsageRecord,
61	} from "@t3tools/provider-core/server/usage";
62	import { resolveModelAliases, UsageAggregator } from "./usageAggregation.ts";
63	import { createOverrideRateTable, parseRateTable, type RateTable } from "./usagePricing.ts";
64	import {
65	listTranscriptFiles,
66	readDirectoryVolumeId,
67	readTranscriptRecords,
68	} from "./usageTranscriptReader.ts";
69	import {
70	decodeScanCache,
71	dedupeWithinFile,
72	LEGACY_SCAN_CACHE_FILE_NAME,
73	makeScanCacheWriter,
74	pruneScanCache,
75	SCAN_CACHE_FILE_NAME,
76	type CachedFile,
77	type ScanCache,
78	} from "./usageScanCache.ts";
79
80	const LITELLM_RATES_URL =
81	"https://raw.githubusercontent.com/BerriAI/litellm/main/model_prices_and_context_window.json";
82
83	/** Rates move rarely; a day-old table keeps the page working offline. */
84	const RATES_TTL_MS = 24 * 60 * 60 * 1000;
85
86	/** An explicit refresh ignores the TTL, but not a table fetched this recently. */
87	const RATES_REFRESH_FLOOR_MS = 60 * 1000;
88
89	/**
90	* Files are filtered by mtime before opening. The slack covers a session whose
91	* last write lands just before local midnight on the window's first day.
92	*/
93	const MTIME_SLACK_MS = 36 * 60 * 60 * 1000;
94	const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000;
95
96	/**
97	* The longest window the UI offers, 90 days, plus its `MTIME_SLACK_MS`, rounded
98	* up. Older entries are pruned.
99	*/
100	const CACHE_RETENTION_DAYS = 92;
101
102	/** Transcripts parsed at once. More gains little once the disk stays busy. */
103	const TRANSCRIPT_READ_CONCURRENCY = 4;
104
105	/** The transcript readers, in driver order. */
106	const transcriptReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) =>
107	driver.usage?.kind === "transcripts" ? [{ driver, reader: driver.usage }] : [],
108	);
109
110	/** The scan readers, in driver order. */
111	const scanReaders = BUILT_IN_USAGE_DRIVERS.flatMap((driver) =>
112	driver.usage?.kind === "scan" ? [{ driver, reader: driver.usage }] : [],
113	);
114
115	/** Transcript formats by provider, for decoding the persisted scan cache. */
116	const transcriptFormats = new Map(
117	transcriptReaders.map(({ reader }) => [reader.provider, reader.format] as const),
118	);
119
120	/** One transcript directory to scan. */
121	interface TranscriptSource {
122	readonly provider: UsageProviderKind;
123	readonly format: TranscriptUsageFormat<unknown>;
124	readonly dir: string;
125	readonly volumeId: string;
126	readonly fileName?: string;
127	}
128
129	/** On-disk shape of the rate snapshot. */
130	const RatesCacheFile = Schema.Struct({
131	fetchedAtMs: Schema.Number,
132	document: Schema.Unknown,
133	});
134	const decodeRatesCache = Schema.decodeUnknownEffect(
135	Schema.fromJsonString(RatesCacheFile as unknown as Schema.Codec<typeof RatesCacheFile.Type>),
136	);
137	const encodeRatesCache = Schema.encodeEffect(
138	Schema.fromJsonString(RatesCacheFile as unknown as Schema.Codec<typeof RatesCacheFile.Type>),
139	);
140
141	/** The scan cache is narrowed by hand in `usageScanCache`, so JSON is enough here. */
142	const ScanCacheJson = Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec<unknown>);
143	const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);
144	const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson);
145	const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String });
146
147	/** Whether `a` read a later state of its file than `b`. Transcripts only grow. */
148	function isLaterRead(a: CachedFile, b: CachedFile): boolean {
149	return a.mtimeMs > b.mtimeMs || (a.mtimeMs === b.mtimeMs && a.size > b.size);
150	}
151
152	/** Providers whose format lets one session's records appear in several files. */
153	const sharedSessionProviders: ReadonlySet<UsageProviderKind> = new Set(
154	transcriptReaders.flatMap(({ reader }) =>
155	reader.format.sharedSessionsAcrossFiles ? [reader.provider] : [],
156	),
157	);
158
159	/**
160	* Sessions of `sharedSessionProviders` with records in more than one file,
161	* such as a rollout that moved after it was read. Only these need cross-file
162	* dedupe keys: within one file the occurrence count already keeps every key
163	* unique, so keying the rest would only build and hash a string for each of
164	* their records.
165	*/
166	function sharedSessions(
167	files: readonly { readonly records: readonly UsageRecord[] }[],
168	): ReadonlySet<string> {
169	const firstFile = new Map<string, number>();
170	const shared = new Set<string>();
171	for (const [index, file] of files.entries()) {
172	let previous = "";
173	for (const { provider, sessionId } of file.records) {
174	if (!sharedSessionProviders.has(provider) || sessionId === previous || sessionId.length === 0)
175	continue;
176	previous = sessionId;
177	const first = firstFile.get(sessionId);
178	if (first === undefined) firstFile.set(sessionId, index);
179	else if (first !== index) shared.add(sessionId);
180	}
181	}
182	return shared;
183	}
184	const decodeCachedSources = Schema.decodeUnknownOption(
185	Schema.Struct({ sources: Schema.Record(Schema.String, CachedSource) }),
186	);
187
188	export class UsageService extends Context.Service<
189	UsageService,
190	{
191	readonly readSummary: (input: UsageSummaryInput) => Effect.Effect<UsageSummary, UsageReadError>;
192	/** Refetches the rate table ahead of its TTL. See `ensureRates`. */
193	readonly refreshRates: Effect.Effect<UsagePricing>;
194	}
195	>()("t3/usage/UsageService") {}
196
197	const EMPTY_PRICING: UsagePricing = {
198	status: "unavailable",
199	source: LITELLM_RATES_URL,
200	fetchedAt: null,
201	knownModels: 0,
202	};
203
204	/** Empty summary, for suites that only need the RPC surface to resolve. */
205	const layerTest = Layer.succeed(
206	UsageService,
207	UsageService.of({
208	readSummary: (input) =>
209	Effect.succeed({
210	contractVersion: USAGE_CONTRACT_VERSION,
211	readAt: "1970-01-01T00:00:00.000Z",
212	timeZone: input.timeZone,
213	sinceDay: input.sinceDay,
214	untilDay: input.untilDay,
215	buckets: [],
216	sources: [],
217	pricing: EMPTY_PRICING,
218	scanDurationMs: 0,
219	}),
220	refreshRates: Effect.succeed(EMPTY_PRICING),
221	}),
222	);
223
224	export const make = Effect.gen(function* () {
225	const fileSystem = yield* FileSystem.FileSystem;
226	const path = yield* Path.Path;
227	const config = yield* ServerConfig.ServerConfig;
228	const settingsService = yield* ServerSettings.ServerSettingsService;
229	const httpClient = yield* HttpClient.HttpClient;
230	const hostEnvironment = yield* HostProcess.Environment;
231	// The readers yield their own services; scans run them against this context.
232	const readerContext = yield* Effect.context<BuiltInUsageReadersEnv>();
233
234	const fileCache: ScanCache = new Map();
235	const sourceCache = new Map<string, typeof CachedSource.Type>();
236	let cacheDirty = false;
237	const isWithinDirectory = (filePath: string, dir: string) => {
238	const relative = path.relative(dir, filePath);
239	return relative !== ".." && !relative.startsWith(".." + path.sep) && !path.isAbsolute(relative);
240	};
241
242	const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json");
243	const scanCachePath = path.join(config.stateDir, SCAN_CACHE_FILE_NAME);
244	const legacyScanCachePath = path.join(config.stateDir, LEGACY_SCAN_CACHE_FILE_NAME);
245	const writeCacheFile = (filePath: string, contents: string) =>
246	writeFileStringAtomically({ filePath, contents }).pipe(
247	Effect.provideService(FileSystem.FileSystem, fileSystem),
248	Effect.provideService(Path.Path, path),
249	);
250	let rates: RateTable = new Map();
251	let ratesFetchedAtMs: number | null = null;
252	let ratesStatus: UsagePricing["status"] = "unavailable";
253	// One fetch at a time. A burst of refreshes from several clients waits on
254	// the first fetch and then sees a table young enough to skip its own.
255	const ratesLock = yield* Semaphore.make(1);
256
257	const pricing = (): UsagePricing => ({
258	status: ratesStatus,
259	source: LITELLM_RATES_URL,
260	fetchedAt:
261	ratesFetchedAtMs === null ? null : DateTime.formatIso(DateTime.makeUnsafe(ratesFetchedAtMs)),
262	knownModels: rates.size,
263	});
264
265	/**
266	* Loads the LiteLLM rate table, preferring a fresh copy and falling back to
267	* the on-disk snapshot. With neither, every model reports as unpriced rather
268	* than the page failing. `force` refetches inside the TTL so a model that
269	* LiteLLM added since the last fetch gets priced now.
270	*/
271	const loadRates = Effect.fn("UsageService.loadRates")(function* (force: boolean) {
272	const now = yield* Clock.currentTimeMillis;
273	const maxAgeMs = force ? RATES_REFRESH_FLOOR_MS : RATES_TTL_MS;
274	if (ratesFetchedAtMs !== null && now - ratesFetchedAtMs < maxAgeMs) return;
275
276	if (ratesFetchedAtMs === null) {
277	const fromDisk = yield* fileSystem.readFileString(ratesCachePath).pipe(
278	Effect.flatMap((raw) => decodeRatesCache(raw)),
279	Effect.catchCause(() => Effect.succeed(null)),
280	);
281	if (fromDisk !== null) {
282	const parsed = parseRateTable(fromDisk.document);
283	if (parsed.size > 0) {
284	rates = parsed;
285	ratesFetchedAtMs = fromDisk.fetchedAtMs;
286	ratesStatus = "cached";
287	if (now - fromDisk.fetchedAtMs < maxAgeMs) return;
288	}
289	}
290	}
291
292	const fetched = yield* httpClient.get(LITELLM_RATES_URL).pipe(
293	Effect.flatMap(HttpClientResponse.filterStatusOk),
294	Effect.flatMap((response) => response.json),
295	Effect.timeout(10_000),
296	Effect.catchCause(() => Effect.succeed(null)),
297	);
298	if (fetched === null) {
299	// The refresh failed; whatever we are serving is now past its TTL and
300	// must not keep claiming to be fresh.
301	if (rates.size > 0) ratesStatus = "cached";
302	return;
303	}
304
305	const parsed = parseRateTable(fetched);
306	if (parsed.size === 0) return;
307
308	rates = parsed;
309	ratesFetchedAtMs = now;
310	ratesStatus = "fresh";
311
312	yield* encodeRatesCache({ fetchedAtMs: now, document: fetched }).pipe(
313	Effect.flatMap((contents) => writeCacheFile(ratesCachePath, contents)),
314	Effect.ignoreCause,
315	);
316	});
317
318	const ensureRates = (force: boolean) => ratesLock.withPermit(loadRates(force));
319
320	const refreshRates = ensureRates(true).pipe(
321	Effect.map(pricing),
322	Effect.withSpan("UsageService.refreshRates"),
323	);
324
325	// A settings failure must not silently discard custom rates or transcript homes.
326	const readSettings = settingsService.getSettings.pipe(
327	Effect.catchCause(
328	(cause) =>
329	new UsageReadError({
330	reason: "scanFailed",
331	detail: "Server settings could not be read.",
332	cause: Cause.squash(cause),
333	}),
334	),
335	);
336
337	/**
338	* Every instance of `driver` with its decoded config, or none when it does
339	* not decode. Disabled accounts still have history. An unconfigured default
340	* slot runs with default config, just as it does in the provider registry.
341	*/
342	const usageInstances = Effect.fn("UsageService.usageInstances")(function* <Config>(
343	driver: ProviderDriver<Config, unknown, unknown>,
344	settings: ServerSettingsValue,
345	) {
346	const entries: Array<
347	readonly [ProviderInstanceId, Pick<ProviderInstanceConfig, "config" | "environment">, boolean]
348	> = Object.entries(settings.providerInstances)
349	.filter(([, instance]) => instance.driver === driver.driverKind)
350	.map(([id, instance]) => [ProviderInstanceId.make(id), instance, true] as const);
351	if (!Object.hasOwn(settings.providerInstances, driver.driverKind)) {
352	entries.push([ProviderInstanceId.make(driver.driverKind), {}, false]);
353	}
354	const decodeConfig = Schema.decodeUnknownOption(driver.configSchema);
355	return yield* Effect.forEach(
356	entries,
357	Effect.fnUntraced(function* ([instanceId, instance, configured]) {
358	const instanceConfig: ProviderUsageInstance<Config> = {
359	instanceId,
360	config: Option.getOrUndefined(decodeConfig(instance.config ?? {})),
361	environment: yield* mergeProviderInstanceEnvironment(
362	instance.environment,
363	hostEnvironment,
364	),
365	configured,
366	};
367	return instanceConfig;
368	}),
369	);
370	});
371
372	/** Resolves every transcript directory the usage readers point at. */
373	const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* (
374	settings: ServerSettingsValue,
375	retentionCutoffMs: number,
376	) {
377	const dirs: Array<TranscriptSource> = [];
378	const seen = new Set<string>();
379	for (const { driver, reader } of transcriptReaders) {
380	const { provider, format } = reader;
381	const directories = yield* Effect.forEach(
382	yield* usageInstances(driver, settings),
383	(instance) => reader.directories(instance),
384	);
385	for (const { dir: directory, fileName } of directories.flat()) {
386	const sourceKey = provider + "\0" + directory;
387	const previous = sourceCache.get(sourceKey);
388	// Keep canonical paths and source fingerprints stable after root cleanup,
389	// including aliases and clients merging pre-cleanup environment summaries.
390	const dir = yield* fileSystem
391	.realPath(directory)
392	.pipe(Effect.orElseSucceed(() => previous?.dir ?? directory));
393	const currentVolumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir));
394	const hasRetainedHistory = fileCache
395	.entries()
396	.some(
397	([filePath, entry]) =>
398	entry.provider === provider &&
399	entry.mtimeMs >= retentionCutoffMs &&
400	entry.records.length + entry.tailRecords.length > 0 &&
401	isWithinDirectory(filePath, dir),
402	);
403	// A recreated directory still reports the retained history under its old identity.
404	const volumeId =
405	previous?.dir === dir && (hasRetainedHistory || !currentVolumeId)
406	? previous.volumeId || currentVolumeId
407	: currentVolumeId;
408	if (previous?.dir !== dir || previous.volumeId !== volumeId) {
409	sourceCache.set(sourceKey, { dir, volumeId });
410	cacheDirty = true;
411	}
412	const key = `${provider}\0${dir}`;
413	if (seen.has(key)) continue;
414	seen.add(key);
415	dirs.push({
416	provider,
417	format,
418	dir,
419	volumeId,
420	...(fileName === undefined ? {} : { fileName }),
421	});
422	}
423	}
424	return dirs;
425	});
426
427	/**
428	* Loads the persisted scan cache exactly once per process.
429	*
430	* `Effect.cached` makes concurrent first readers await the same load rather
431	* than each seeing a "loaded" flag set before the read finished and cold
432	* scanning against an empty cache.
433	*/
434	const ensureScanCacheLoaded = yield* Effect.cached(
435	Effect.gen(function* () {
436	const readDocument = (filePath: string) =>
437	fileSystem.readFileString(filePath).pipe(
438	Effect.flatMap((raw) => decodeScanCacheFile(raw)),
439	Effect.catchCause(() => Effect.succeed(null)),
440	);
441	let document = yield* readDocument(scanCachePath);
442	if (document === null) {
443	document = yield* readDocument(legacyScanCachePath);
444	// Write the migrated cache to its own file on the next scan.
445	cacheDirty = document !== null;
446	}
447	if (document === null) return;
448	for (const [path, entry] of decodeScanCache(document, transcriptFormats)) {
449	fileCache.set(path, entry);
450	}
451	const sources = decodeCachedSources(document);
452	if (Option.isSome(sources)) {
453	for (const [key, source] of Object.entries(sources.value.sources))
454	sourceCache.set(key, source);
455	}
456	}),
457	);
458
459	const writeScanCache = makeScanCacheWriter();
460	// Scans with different windows can finish together; serializing the writes
461	// keeps an older snapshot from landing after a newer one.
462	const persistLock = yield* Semaphore.make(1);
463
464	// The dirty flag is cleared before encoding, so a change while the write is
465	// in flight marks it dirty again. A failed write restores the flag, so the
466	// next persist retries instead of leaving disk stale. A cache we cannot
467	// write is a slower next start, not a failed read.
468	const persistCaches = Effect.gen(function* () {
469	if (cacheDirty) {
470	cacheDirty = false;
471	yield* Effect.sync(() =>
472	writeScanCache(fileCache, { sources: Object.fromEntries(sourceCache) }),
473	).pipe(
474	Effect.flatMap((contents) => writeCacheFile(scanCachePath, contents)),
475	Effect.catchCause(() =>
476	Effect.sync(() => {
477	cacheDirty = true;
478	}),
479	),
480	);
481	}
482	}).pipe(persistLock.withPermit, Effect.withSpan("UsageService.persistCaches"));
483
484	const pendingPersists = new Set<Fiber.Fiber<void>>();
485	/** Writes dirty caches in the background, after the summary that dirtied them answers. */
486	const schedulePersist = Effect.forkDetach(persistCaches).pipe(
487	Effect.map((fiber) => {
488	pendingPersists.add(fiber);
489	fiber.addObserver(() => pendingPersists.delete(fiber));
490	}),
491	);
492	/** Waits for every write scheduled so far, as a restart would need. */
493	const awaitPersisted = Effect.suspend(() => Fiber.awaitAll([...pendingPersists])).pipe(
494	Effect.asVoid,
495	);
496	// A write still running when the service shuts down finishes first, so the
497	// next start does not lose the last scan.
498	yield* Effect.addFinalizer(() => awaitPersisted);
499
500	/**
501	* Parses one transcript, reusing the cached result when it is unchanged.
502	*
503	* A file that only grew re-parses from the cached position, so an actively
504	* written multi-hundred-megabyte rollout costs its appended bytes per scan
505	* rather than a full re-read. The reader verifies the position's guard bytes
506	* and silently restarts from byte 0 when they no longer match.
507	*
508	* A fresh parse comes back as `update` for the caller to cache, with the
509	* entry it was built from. Reads run concurrently, and the caller stores
510	* updates in walk order rather than completion order: saved records of
511	* deleted transcripts aggregate in cache order, where the first copy of a
512	* duplicate wins.
513	*/
514	const readFileRecords = (
515	filePath: string,
516	size: number,
517	mtimeMs: number,
518	provider: UsageProviderKind,
519	format: TranscriptUsageFormat<unknown>,
520	): Effect.Effect<{
521	readonly records: readonly UsageRecord[];
522	readonly failed?: true;
523	readonly update?: { readonly entry: CachedFile; readonly replaces: CachedFile | undefined };
524	}> =>
525	Effect.gen(function* () {
526	const cached = fileCache.get(filePath);
527	// Provider is part of the identity: if both providers were ever pointed
528	// at one directory, a hit parsed by the other parser must not be reused.
529	if (
530	cached &&
531	cached.size === size &&
532	cached.mtimeMs === mtimeMs &&
533	cached.provider === provider
534	) {
535	return {
536	records:
537	cached.tailRecords.length === 0
538	? cached.records
539	: [...cached.records, ...cached.tailRecords],
540	};
541	}
542
543	// Only a strictly grown file may resume. Same size with a new mtime, or
544	// a shrunken file, means rewritten content; re-parse it whole.
545	const resumeFrom =
546	cached !== undefined && cached.provider === provider && size > cached.size
547	? cached.position
548	: undefined;
549
550	const parsed = yield* Effect.promise(() =>
551	readTranscriptRecords(filePath, format, resumeFrom),
552	);
553	// A read failure is not an empty transcript: caching it under this
554	// (size, mtime) would silently drop the file's usage until it changes.
555	if (parsed === null)
556	return {
557	records: cached?.provider === provider ? [...cached.records, ...cached.tailRecords] : [],
558	failed: true,
559	};
560
561	// Stored already de-duplicated within the file, which is 99% of all
562	// duplicates. The aggregator still runs the cross-file dedupe pass. One
563	// seen set spans the cached base, the new lines, and the tail so a
564	// resumed parse dedupes exactly like a full one.
565	const base = parsed.resumed && cached !== undefined ? cached.records : [];
566	const seen = new Set<string>();
567	const records = dedupeWithinFile([...base, ...parsed.records], seen);
568	const tailRecords = dedupeWithinFile(parsed.tailRecords, seen);
569
570	return {
571	records: tailRecords.length === 0 ? records : [...records, ...tailRecords],
572	update: {
573	entry: { size, mtimeMs, provider, records, tailRecords, position: parsed.position },
574	replaces: cached,
575	},
576	};
577	});
578
579	/** One provider directory's walk and parse, before rates are involved. */
580	interface ScannedDir {
581	readonly provider: UsageProviderKind;
582	readonly dir: string;
583	readonly volumeId: string;
584	readonly hostId?: string;
585	readonly status?: UsageSource["status"];
586	readonly message?: string;
587	readonly action?: UsageSource["action"];
588	/** Parsed records per file, or `null` when the directory does not exist. */
589	readonly files:
590	| readonly { readonly path: string; readonly records: readonly UsageRecord[] }[]
591	| null;
592	/** Answered from a cache while a refresh runs. */
593	readonly refreshing?: true;
594	}
595
596	const scanTranscriptDir = Effect.fn("UsageService.scanTranscriptDir")(function* (
597	source: TranscriptSource,
598	windowStartMs: number,
599	) {
600	const { provider, format, dir, volumeId, fileName } = source;
601	const exists = yield* fileSystem
602	.exists(dir)
603	.pipe(Effect.catchCause(() => Effect.succeed(false)));
604	if (!exists) return { provider, dir, volumeId, files: null } satisfies ScannedDir;
605	const { files, failedPaths } = yield* Effect.promise(() =>
606	listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }),
607	);
608	// A cold parse waits on disk reads, so a few files in flight read
609	// close to twice as fast. Results keep walk order.
610	const read = yield* Effect.forEach(
611	files,
612	(file) =>
613	readFileRecords(file.path, file.size, file.mtimeMs, provider, format).pipe(
614	Effect.map((result) => ({ path: file.path, ...result })),
615	),
616	{ concurrency: TRANSCRIPT_READ_CONCURRENCY },
617	);
618	const parsedFiles = read.map(({ path, records, update }) => {
619	if (update === undefined) return { path, records };
620	// A scan of another window may have cached its own read of this file
621	// meanwhile. Then keep whichever read saw the later file, so a slower
622	// scan never replaces newer usage with older.
623	const current = fileCache.get(path);
624	if (
625	current === update.replaces ||
626	current === undefined ||
627	!isLaterRead(current, update.entry)
628	) {
629	fileCache.set(path, update.entry);
630	cacheDirty = true;
631	}
632	return { path, records };
633	});
634	// Unread files keep their cached usage, but the total may be short.
635	const unread = failedPaths + read.filter((file) => file.failed).length;
636	return {
637	provider,
638	dir,
639	volumeId,
640	files: parsedFiles,
641	...(unread > 0
642	? {
643	status: "partial",
644	message: `${unread} transcript path(s) could not be read; usage may be incomplete.`,
645	}
646	: {}),
647	} satisfies ScannedDir;
648	});
649
650	const collectDirs = Effect.fn("UsageService.collectDirs")(function* (
651	windowStartMs: number,
652	settings: ServerSettingsValue,
653	retentionCutoffMs: number,
654	awaitRefresh: boolean,
655	) {
656	const dirs = yield* resolveTranscriptDirs(settings, retentionCutoffMs).pipe(
657	Effect.provideContext(readerContext),
658	);
659
660	const scans = Effect.forEach(
661	scanReaders,
662	({ driver, reader }) =>
663	usageInstances(driver, settings)
664	.pipe(
665	Effect.flatMap((instances) =>
666	reader.scan({ instances, settings, windowStartMs, retentionCutoffMs, awaitRefresh }),
667	),
668	)
669	.pipe(
670	Effect.flatMap((sources) =>
671	Effect.forEach(sources, ({ volumeId, ...source }) =>
672	Effect.map(
673	volumeId === undefined
674	? Effect.promise(() => readDirectoryVolumeId(source.dir))
675	: Effect.succeed(volumeId),
676	(resolved): ScannedDir => ({
677	...source,
678	provider: reader.provider,
679	volumeId: resolved,
680	}),
681	),
682	),
683	),
684	Effect.provideContext(readerContext),
685	),
686	{ concurrency: "unbounded" },
687	);
688
689	// Independent sources scan together. Transcript directories go one at a
690	// time, so open files stay at `TRANSCRIPT_READ_CONCURRENCY`. The result
691	// keeps this order, since aggregation keeps the first copy of a duplicate.
692	const [transcripts, scanDirs] = yield* Effect.all(
693	[Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs)), scans],
694	{ concurrency: "unbounded" },
695	);
696	const scanned: readonly ScannedDir[] = [...transcripts, ...scanDirs.flat()];
697	return scanned;
698	});
699
700	const scanSummary = Effect.fn("UsageService.scanSummary")(function* (
701	input: UsageSummaryInput,
702	settings: ServerSettingsValue,
703	) {
704	if (input.sinceDay > input.untilDay) {
705	return yield* new UsageReadError({
706	reason: "invalidWindow",
707	detail: `sinceDay '${input.sinceDay}' is after untilDay '${input.untilDay}'`,
708	});
709	}
710
711	let hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null = null;
712	if (input.resolution === "hour") {
713	const sinceTime =
714	input.sinceTime === undefined ? Option.none() : DateTime.make(input.sinceTime);
715	const untilTime =
716	input.untilTime === undefined ? Option.none() : DateTime.make(input.untilTime);
717	if (Option.isNone(sinceTime) || Option.isNone(untilTime)) {
718	return yield* new UsageReadError({
719	reason: "invalidWindow",
720	detail: "Hourly usage requires valid sinceTime and untilTime instants",
721	});
722	}
723	const sinceTimeMs = DateTime.toEpochMillis(sinceTime.value);
724	const untilTimeMs = DateTime.toEpochMillis(untilTime.value);
725	const durationMs = untilTimeMs - sinceTimeMs;
726	if (durationMs <= 0 || durationMs > MAX_HOURLY_WINDOW_MS) {
727	return yield* new UsageReadError({
728	reason: "invalidWindow",
729	detail: "Hourly usage window must be greater than zero and at most 24 hours",
730	});
731	}
732	hourlyWindow = { sinceTimeMs, untilTimeMs };
733	}
734
735	const startedAtMs = yield* Clock.currentTimeMillis;
736	yield* ensureScanCacheLoaded;
737
738	const hostId = NodeOS.hostname();
739	const windowStart = DateTime.make(`${input.sinceDay}T00:00:00Z`);
740	if (Option.isNone(windowStart)) {
741	return yield* new UsageReadError({
742	reason: "invalidWindow",
743	detail: `sinceDay '${input.sinceDay}' is not a valid date`,
744	});
745	}
746	const windowStartMs =
747	(hourlyWindow?.sinceTimeMs ?? DateTime.toEpochMillis(windowStart.value)) - MTIME_SLACK_MS;
748
749	const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000;
750
751	// Pricing only matters once records are aggregated, so the rate table
752	// loads while transcripts stream instead of gating them: a cold rates
753	// fetch on a slow network no longer delays the scan by its own timeout.
754	const [, scannedDirs] = yield* Effect.all(
755	[
756	ensureRates(false),
757	collectDirs(windowStartMs, settings, retentionCutoffMs, input.awaitRefresh === true),
758	],
759	{ concurrency: 2 },
760	);
761
762	const aggregator = new UsageAggregator({
763	timeZone: input.timeZone,
764	sinceDay: input.sinceDay,
765	untilDay: input.untilDay,
766	resolution: input.resolution ?? "day",
767	...hourlyWindow,
768	rates,
769	priceOverrides: createOverrideRateTable(settings.usagePriceOverrides),
770	modelAliases: resolveModelAliases(settings.usageModelAliases),
771	});
772
773	const sources: UsageSource[] = [];
774
775	// Cleanup may remove transcripts, but the usage we already saved still
776	// contributes to its source through the normal aggregation and dedupe
777	// path. Like the walk, skip files last written before the window: they
778	// cannot hold records inside it.
779	const retainedSinceMs = Math.max(windowStartMs, retentionCutoffMs);
780	const filesByDir = scannedDirs.map(({ provider, dir, files }) => {
781	const retainedFiles = [...(files ?? [])];
782	const livePaths = new Set(retainedFiles.map((file) => file.path));
783	for (const [filePath, entry] of fileCache) {
784	if (
785	entry.provider !== provider ||
786	entry.mtimeMs < retainedSinceMs ||
787	livePaths.has(filePath) ||
788	!isWithinDirectory(filePath, dir)
789	)
790	continue;
791	retainedFiles.push({ path: filePath, records: [...entry.records, ...entry.tailRecords] });
792	}
793	return retainedFiles;
794	});
795	const crossFileSessions = sharedSessions(filesByDir.flat());
796
797	for (const [
798	index,
799	{ provider, dir, volumeId, files, status, message, action, refreshing, hostId: sourceHostId },
800	] of scannedDirs.entries()) {
801	let scannedFiles = 0;
802	let skippedFiles = 0;
803	// Distinct per directory. Buckets carry per-cell session counts, but a
804	// session spans days and models, so clients total this figure instead.
805	const sessionIds = new Set<string>();
806
807	for (const file of filesByDir[index] ?? []) {
808	if (file.records.length === 0) {
809	skippedFiles += 1;
810	continue;
811	}
812	scannedFiles += 1;
813	const eventOccurrences = new Map<string, number>();
814	for (const record of file.records) {
815	let usageRecord = record;
816	if (
817	sharedSessionProviders.has(record.provider) &&
818	crossFileSessions.has(record.sessionId)
819	) {
820	// Match moved copies without collapsing repeated equal events
821	// within one file (timestamps can have only second precision).
822	// Only sessions seen in several files can have a copy to match.
823	const key = encodeUsageRecordKey([
824	record.provider,
825	record.sessionId,
826	record.timestampMs,
827	record.model,
828	record.totals,
829	]);
830	const occurrence = (eventOccurrences.get(key) ?? 0) + 1;
831	eventOccurrences.set(key, occurrence);
832	usageRecord = { ...record, dedupeKey: key + ":" + occurrence };
833	}
834	// Only sessions contributing in-window count; the mtime slack can
835	// admit boundary files whose records fall outside the range.
836	if (aggregator.add(usageRecord, dir) && record.sessionId.length > 0) {
837	sessionIds.add(record.sessionId);
838	}
839	}
840	}
841
842	sources.push({
843	fingerprint: { hostId: sourceHostId ?? hostId, provider, resolvedHomePath: dir, volumeId },
844	// Clients exclude missing sources, so saved records remain an available source.
845	status: files === null && scannedFiles === 0 ? "missing" : (status ?? "ok"),
846	scannedFiles,
847	skippedFiles,
848	malformedRecords: 0,
849	distinctSessions: sessionIds.size,
850	message:
851	message ?? (files === null ? "No transcript directory on this environment." : null),
852	...(action ? { action } : {}),
853	...(refreshing ? { refreshing } : {}),
854	});
855	}
856
857	const pruned = pruneScanCache(fileCache, retentionCutoffMs);
858	if (pruned > 0) cacheDirty = true;
859
860	const aggregated = aggregator.finish();
861	const readAt = yield* DateTime.now;
862	const finishedAtMs = yield* Clock.currentTimeMillis;
863
864	return {
865	contractVersion: USAGE_CONTRACT_VERSION,
866	readAt: DateTime.formatIso(readAt),
867	timeZone: input.timeZone,
868	sinceDay: input.sinceDay,
869	untilDay: input.untilDay,
870	buckets: aggregated.buckets,
871	sources,
872	pricing: pricing(),
873	scanDurationMs: Math.max(0, finishedAtMs - startedAtMs),
874	} satisfies UsageSummary;
875	});
876
877	/**
878	* In-flight scans by window and settings, so concurrent identical requests (the usage
879	* page open on two clients at once) share one scan instead of racing over
880	* the same corpus twice. Readers read settings of their own, so a scan is
881	* shared only under the same settings snapshot.
882	*/
883	const inflightScans = new Map<string, Deferred.Deferred<UsageSummary, UsageReadError>>();
884
885	const scanKey = (input: UsageSummaryInput, settings: ServerSettingsValue): string =>
886	JSON.stringify([
887	input.timeZone,
888	input.sinceDay,
889	input.untilDay,
890	input.resolution ?? "day",
891	input.sinceTime ?? null,
892	input.untilTime ?? null,
893	settings,
894	// A waiting read must never share a scan that answers with `refreshing`.
895	input.awaitRefresh === true,
896	]);
897
898	const readSummary = Effect.fn("UsageService.readSummary")(function* (input: UsageSummaryInput) {
899	const settings = yield* readSettings;
900	const key = scanKey(input, settings);
901	const deferred = yield* Effect.uninterruptible(
902	Effect.gen(function* () {
903	const existing = inflightScans.get(key);
904	if (existing !== undefined) return existing;
905
906	// Enrollment and detached-fiber creation must be atomic. Otherwise a
907	// canceled first caller can leave a Deferred with no scan to finish it.
908	const created = Deferred.makeUnsafe<UsageSummary, UsageReadError>();
909	inflightScans.set(key, created);
910	// Detached so one departing client cannot tear the scan out from under
911	// the fibers awaiting it; a finished scan warms the cache either way.
912	// The cache write is registered before the waiters resume, so they
913	// can await it, but its fiber starts after they have the summary.
914	yield* scanSummary(input, settings).pipe(
915	Effect.onExit((exit) =>
916	Effect.sync(() => inflightScans.delete(key)).pipe(
917	Effect.andThen(Exit.isSuccess(exit) ? schedulePersist : Effect.void),
918	Effect.andThen(Deferred.done(created, exit)),
919	),
920	),
921	Effect.forkDetach,
922	);
923	return created;
924	}),
925	);
926	// Waiting stays interruptible. The detached scan continues for other
927	// callers and still warms the cache if this caller leaves.
928	return yield* Deferred.await(deferred);
929	});
930
931	// `awaitPersisted` is outside the service interface: tests use it to restart
932	// against what a previous instance wrote.
933	return { readSummary, refreshRates, awaitPersisted } as const;
934	});
935
936	export const layer = Layer.effect(UsageService, make);