From 8954990255e12c3f6dbfd30f86df79488e563e59 Mon Sep 17 00:00:00 2001 From: Bil0000 <62337003+Bil0000@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:21:01 +0100 Subject: [PATCH 1/8] perf(usage): the Usage page shows figures at once instead of after a full transcript scan Every usage read walked and stat'ed all provider transcripts before it answered: 2 to 11 seconds per period on a machine with 23,000 files. A client with nothing to show now asks for a provisional answer. The server gives it from the scan it already holds in memory, when a complete earlier scan covers the window, and marks the sources as refreshing. The existing follow-up read then runs the exact scan, so the final numbers are unchanged. The exact scan also lists sibling directories together and keeps the serial walk order, which first-copy dedupe depends on. --- apps/server/src/usage/UsageService.test.ts | 102 +++++++++ apps/server/src/usage/UsageService.ts | 202 ++++++++++++++++-- .../src/usage/usageTranscriptReader.test.ts | 24 +++ .../server/src/usage/usageTranscriptReader.ts | 49 +++-- packages/client-runtime/src/state/server.ts | 18 +- .../src/state/serverUsage.test.ts | 10 +- packages/contracts/src/usage.ts | 2 + 7 files changed, 367 insertions(+), 40 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index a3775c6aebe17..2d062eb98f3b9 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2259,6 +2259,74 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live( + "answers a provisional read from the last scan, then the exact read adds what changed", + () => + Effect.gen(function* () { + const { transcript, settings, home } = yield* setup; + yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5))); + const provisional = { ...WINDOW, provisional: true }; + const exact = { ...WINDOW, awaitRefresh: true }; + const claude = (summary: UsageSummary) => + summary.sources.find((source) => source.fingerprint.provider === "claude"); + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + // A scan that missed a path covers nothing, so the next answer is a full scan too. + const loop = NodePath.join(NodePath.dirname(transcript), "loop.jsonl"); + yield* Effect.promise(() => NodeFSP.symlink(loop, loop, "junction")); + assert.strictEqual(claude(yield* service.readSummary(provisional))?.status, "partial"); + assert.isUndefined(claude(yield* service.readSummary(provisional))?.refreshing); + yield* Effect.promise(() => NodeFSP.rm(loop)); + // Nothing saved covers the window yet, so the first answer is a full scan. + const first = yield* service.readSummary(provisional); + assert.strictEqual(totalOutputTokens(first), 5); + assert.isUndefined(claude(first)?.refreshing); + + yield* Effect.promise(() => NodeFSP.appendFile(transcript, claudeLine(2, 7))); + const saved = yield* service.readSummary(provisional); + assert.strictEqual(totalOutputTokens(saved), 5); + assert.isTrue(claude(saved)?.refreshing); + assert.isTrue( + saved.sources.every((source) => source === claude(saved) || !source.refreshing), + ); + const refreshed = yield* service.readSummary(exact); + assert.strictEqual(totalOutputTokens(refreshed), 12); + assert.isUndefined(claude(refreshed)?.refreshing); + assert.deepStrictEqual( + (yield* service.readSummary(provisional)).buckets, + refreshed.buckets, + ); + yield* service.awaitPersisted; + + // A restarted server has no earlier answer, only the persisted scan. + yield* Effect.promise(() => NodeFSP.appendFile(transcript, claudeLine(3, 11))); + const restarted = yield* UsageService.make; + const restored = yield* restarted.readSummary(provisional); + assert.deepStrictEqual(restored.buckets, refreshed.buckets); + assert.isTrue(claude(restored)?.refreshing); + // A window that starts before every scan could only show part of its usage from memory. + const wider = yield* restarted.readSummary({ + ...provisional, + sinceDay: UsageDay.make("2026-07-01"), + }); + assert.strictEqual(totalOutputTokens(wider), 23); + assert.isUndefined(claude(wider)?.refreshing); + assert.strictEqual(totalOutputTokens(yield* restarted.readSummary(exact)), 23); + // With no saved usage to show, the answer is exact: nothing would tell the client to ask again. + const empty = yield* restarted.readSummary({ + ...provisional, + modelFilter: { include: [{ model: "absent" }] }, + }); + assert.deepStrictEqual(empty.buckets, []); + assert.isTrue(empty.sources.every((source) => !source.refreshing)); + yield* restarted.awaitPersisted; + }).pipe( + Effect.scoped, + Effect.provide(layerService({ prefix: "usage-service-provisional", home, settings })), + ); + }).pipe(Effect.scoped), + ); + it.live("filters original model IDs before aliases, pricing, buckets and session counts", () => Effect.gen(function* () { const { transcript, settings, home } = yield* setup; @@ -2951,6 +3019,40 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live("credits the same copy of a duplicate from the last scan as an exact read does", () => + Effect.gen(function* () { + yield* TestClock.setTime(Date.parse("2026-08-15T00:00:00Z")); + const { transcript, settings, home } = yield* setup; + const dir = NodePath.dirname(transcript); + const [first = "", second = ""] = yield* Effect.promise(async () => { + await NodeFSP.writeFile(NodePath.join(dir, "a.jsonl"), ""); + await NodeFSP.writeFile(NodePath.join(dir, "b.jsonl"), ""); + return (await NodeFSP.readdir(dir)).map((name) => NodePath.join(dir, name)); + }); + const forked = (line: string) => line.replace('"session-1"', '"session-2"'); + yield* Effect.promise(async () => { + await NodeFSP.writeFile(first, claudeLine(1, 5)); + await NodeFSP.utimes(first, BEFORE_NARROW_WINDOW, BEFORE_NARROW_WINDOW); + await NodeFSP.writeFile(second, forked(claudeLine(1, 5)) + forked(claudeLine(2, 7))); + }); + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + // The narrow window saves the second copy before the wide one reads the first. + yield* service.readSummary(NARROW_WINDOW); + const exact = yield* service.readSummary(WINDOW); + assert.strictEqual(exact.buckets[0]?.sessions, 2); + yield* service.awaitPersisted; + const restarted = yield* UsageService.make; + const saved = yield* restarted.readSummary({ ...WINDOW, provisional: true }); + assert.deepStrictEqual(saved.buckets, exact.buckets); + yield* Effect.promise(() => Promise.all([NodeFSP.rm(first), NodeFSP.rm(second)])); + assert.deepStrictEqual((yield* restarted.readSummary(WINDOW)).buckets, exact.buckets); + }).pipe( + Effect.provide(layerService({ prefix: "usage-service-saved-order-test", home, settings })), + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + it.live.skipIf(HostProcess.Platform.defaultValue() === "win32")( "keeps a newer cached read when a slower scan of another window finishes later", () => diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 6b6762c089ba6..e809909940bcc 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -136,7 +136,10 @@ function sharedSessions( return shared; } const decodeCachedSources = Schema.decodeUnknownOption( - Schema.Struct({ sources: Schema.Record(Schema.String, CachedSource) }), + Schema.Struct({ + sources: Schema.Record(Schema.String, CachedSource), + coveredSince: Schema.optional(Schema.Record(Schema.String, Schema.Number)), + }), ); export class UsageService extends Context.Service< @@ -164,6 +167,10 @@ export const make = Effect.gen(function* () { const correctedBatch = (records: readonly UsageRecord[]) => records.some((record) => resolvedRecords.has(record)) ? records.map(correctedRecord) : records; const sourceCache = new Map(); + // Earliest window start a full scan read each transcript directory from. A window inside it + // has all its files in fileCache except what changed since, so it can be answered from memory. + const coveredSince = new Map(); + const coverageKey = (source: TranscriptSource) => `${source.provider}\0${source.dir}`; let cacheDirty = false; const isWithinDirectory = (filePath: string, dir: string) => { const relative = path.relative(dir, filePath); @@ -234,7 +241,10 @@ export const make = Effect.gen(function* () { Effect.catchCause(() => Effect.succeed(null)), ); - const loadRates = Effect.fn("UsageService.loadRates")(function* (force: boolean) { + const loadRates = Effect.fn("UsageService.loadRates")(function* ( + force: boolean, + savedOnly = false, + ) { const now = yield* Clock.currentTimeMillis; const maxAgeMs = force ? RATES_REFRESH_FLOOR_MS : RATES_TTL_MS; if (ratesFetchedAtMs !== null && now - ratesFetchedAtMs < maxAgeMs) return; @@ -255,6 +265,8 @@ export const make = Effect.gen(function* () { } } + // Saved rates of any age price a provisional answer; only an exact read waits on the network. + if (savedOnly && rates.size > 0) return; if (ratesLastFailedAtMs !== null && now - ratesLastFailedAtMs < RATES_REFRESH_FLOOR_MS) return; let fetched = yield* fetchRates(LITELLM_RATES_URL); @@ -278,7 +290,8 @@ export const make = Effect.gen(function* () { ); }); - const ensureRates = (force: boolean) => ratesLock.withPermit(loadRates(force)); + const ensureRates = (force: boolean, savedOnly = false) => + ratesLock.withPermit(loadRates(force, savedOnly)); const refreshRates = ensureRates(true).pipe( Effect.map(pricing), @@ -399,6 +412,11 @@ export const make = Effect.gen(function* () { if (Option.isSome(sources)) { for (const [key, source] of Object.entries(sources.value.sources)) sourceCache.set(key, source); + // Coverage describes the saved files, so a rejected document keeps none of it. + if (fileCache.size > 0) { + for (const [key, since] of Object.entries(sources.value.coveredSince ?? {})) + coveredSince.set(key, since); + } } }), ); @@ -412,7 +430,10 @@ export const make = Effect.gen(function* () { if (cacheDirty) { cacheDirty = false; yield* Effect.sync(() => - writeScanCache(fileCache, { sources: Object.fromEntries(sourceCache) }), + writeScanCache(fileCache, { + sources: Object.fromEntries(sourceCache), + coveredSince: Object.fromEntries(coveredSince), + }), ).pipe( Effect.flatMap((contents) => writeCacheFile(scanCachePath, contents)), Effect.catchCause(() => @@ -559,12 +580,15 @@ export const make = Effect.gen(function* () { const scanTranscriptDir = Effect.fn("UsageService.scanTranscriptDir")(function* ( source: TranscriptSource, windowStartMs: number, + fromMemory: boolean, ) { const { provider, format, dir, volumeId, fileName } = source; const exists = yield* fileSystem .exists(dir) .pipe(Effect.catchCause(() => Effect.succeed(false))); if (!exists) return { provider, dir, volumeId, files: null } satisfies ScannedDir; + // No walk: scanSummary aggregates this directory's saved files, as it does for deleted ones. + if (fromMemory) return { provider, dir, volumeId, files: [] } satisfies ScannedDir; const { files, failedPaths } = yield* Effect.promise(() => listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }), ); @@ -597,6 +621,12 @@ export const make = Effect.gen(function* () { return { path, records, malformedRecords }; }); const unread = failedPaths + read.filter((file) => file.failed).length; + const covered = coveredSince.get(coverageKey(source)); + // A read that missed paths covers nothing: memory would answer for files it never held. + if (unread === 0 && (covered === undefined || covered > windowStartMs)) { + coveredSince.set(coverageKey(source), windowStartMs); + cacheDirty = true; + } yield* Effect.annotateCurrentSpan({ "usage.files": files.length, "usage.files_parsed": read.filter((file) => file.update !== undefined).length, @@ -620,10 +650,27 @@ export const make = Effect.gen(function* () { settings: ServerSettingsValue, retentionCutoffMs: number, awaitRefresh: boolean, + provisional: boolean, ) { const dirs = yield* resolveTranscriptDirs(settings, retentionCutoffMs).pipe( Effect.provideContext(readerContext), ); + // A window no full scan has read would show partial totals, so every directory must cover it. + const coverage = provisional + ? yield* Effect.forEach(dirs, (source) => + fileSystem.exists(source.dir).pipe( + Effect.catchCause(() => Effect.succeed(false)), + Effect.map((exists) => + !exists + ? "missing" + : (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs + ? "covered" + : "uncovered", + ), + ), + ) + : []; + const fromMemory = coverage.includes("covered") && !coverage.includes("uncovered"); const scans = Effect.forEach( scanReaders, @@ -654,16 +701,44 @@ export const make = Effect.gen(function* () { // Scan independent sources together, directories sequentially to bound open files; preserve order for first-copy dedupe. const [transcripts, scanDirs] = yield* Effect.all( - [Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs)), scans], + [Effect.forEach(dirs, (dir) => scanTranscriptDir(dir, windowStartMs, fromMemory)), scans], { concurrency: "unbounded" }, ); const scanned: readonly ScannedDir[] = [...transcripts, ...scanDirs.flat()]; - return scanned; + return { scanned, fromMemory, unread: transcripts.some((dir) => "status" in dir) }; }); + /** + * A saved answer misses whatever changed since its scan, so it is never final: its transcript + * sources with usage are marked refreshing, which tells the client to ask again with + * awaitRefresh. Null when none has usage: there is nothing to show early, and marking an + * empty source would only flash a loading row. + */ + const asSaved = (summary: UsageSummary): UsageSummary | null => { + const marked = new Set( + summary.sources.filter( + ({ fingerprint }) => + transcriptFormats.has(fingerprint.provider) && + summary.buckets.some( + (bucket) => + bucket.provider === fingerprint.provider && + bucket.sourcePath === fingerprint.resolvedHomePath, + ), + ), + ); + if (marked.size === 0) return null; + return { + ...summary, + sources: summary.sources.map((source) => + marked.has(source) ? { ...source, refreshing: true } : source, + ), + }; + }; + const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, + provisional: boolean, ) { if (input.sinceDay > input.untilDay) { return yield* new UsageReadError({ @@ -711,13 +786,21 @@ export const make = Effect.gen(function* () { const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000; - const [, scannedDirs] = yield* Effect.all( + const [, { scanned: scannedDirs, fromMemory, unread }] = yield* Effect.all( [ - ensureRates(false), - collectDirs(windowStartMs, settings, retentionCutoffMs, input.awaitRefresh === true), + // Loaded rates skip the lock, which an exact read holds through a slow fetch. + provisional && rates.size > 0 ? Effect.void : ensureRates(false, provisional), + collectDirs( + windowStartMs, + settings, + retentionCutoffMs, + input.awaitRefresh === true, + provisional, + ), ], { concurrency: 2 }, ); + if (provisional && !fromMemory) yield* ensureRates(false); const aggregator = new UsageAggregator({ timeZone: input.timeZone, @@ -738,28 +821,65 @@ export const make = Effect.gen(function* () { const filesByDir = scannedDirs.map(({ provider, dir, files }) => { const retainedFiles = [...(files ?? [])]; const livePaths = new Set(retainedFiles.map((file) => file.path)); + // Saved paths are files below a scanned root, so their folder decides; folders repeat thousands of times. + const foldersWithin = new Map(); + const isRetained = (filePath: string) => { + const folder = path.dirname(filePath); + let within = foldersWithin.get(folder); + if (within === undefined) { + within = isWithinDirectory(folder, dir); + foldersWithin.set(folder, within); + } + return within; + }; for (const [filePath, entry] of fileCache) { if ( entry.provider !== provider || entry.mtimeMs < retainedSinceMs || livePaths.has(filePath) || - !isWithinDirectory(filePath, dir) + !isRetained(filePath) ) continue; retainedFiles.push({ path: filePath, - records: [...entry.records, ...entry.tailRecords], + records: + entry.tailRecords.length === 0 + ? entry.records + : [...entry.records, ...entry.tailRecords], malformedRecords: entry.malformedRecords, }); } return retainedFiles; }); + // Usage two transcripts share counts in the first one, so saved files keep the order of the last walk. + for (const files of fromMemory ? [] : filesByDir) { + const saved = files.flatMap((file) => (fileCache.has(file.path) ? [file.path] : [])); + if (saved.length < 2) continue; + const members = new Set(saved); + const expected = saved.values(); + if ( + fileCache + .keys() + .filter((filePath) => members.has(filePath)) + .every((filePath) => filePath === expected.next().value) + ) + continue; + // Reorder within the slots these files hold, so files this walk did not list keep their place. + const entries = new Map(fileCache); + const walked = saved.values(); + fileCache.clear(); + for (const filePath of entries.keys()) { + const slot = members.has(filePath) ? walked.next().value! : filePath; + fileCache.set(slot, entries.get(slot)!); + } + cacheDirty = true; + } const windowEndMs = hourlyWindow?.untilTimeMs ?? Date.parse(`${input.untilDay}T00:00:00Z`) + 86400000 + MTIME_SLACK_MS; let resolvedAny = false; for (const { reader } of transcriptReaders) { - if (reader.resolveServiceTiers === undefined) continue; + if (reader.resolveServiceTiers === undefined || fromMemory) continue; const pending = new Set(); for (const [index, files] of filesByDir.entries()) { if (scannedDirs[index]?.provider !== reader.provider) continue; @@ -870,13 +990,19 @@ export const make = Effect.gen(function* () { } const pruned = pruneScanCache(fileCache, retentionCutoffMs); - if (pruned > 0) cacheDirty = true; + if (pruned > 0) { + cacheDirty = true; + // Pruned files are gone from memory, so nothing before the cutoff is covered any more. + for (const [key, since] of coveredSince) { + if (since < retentionCutoffMs) coveredSince.set(key, retentionCutoffMs); + } + } const aggregated = aggregator.finish(); const readAt = yield* DateTime.now; const finishedAtMs = yield* Clock.currentTimeMillis; - return { + const summary: UsageSummary = { contractVersion: USAGE_CONTRACT_VERSION, readAt: DateTime.formatIso(readAt), timeZone: input.timeZone, @@ -887,12 +1013,30 @@ export const make = Effect.gen(function* () { pricing: pricing(), scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), - } satisfies UsageSummary; + }; + return { summary, fromMemory, unread }; + }); + + const readWindow = Effect.fn("UsageService.readWindow")(function* ( + input: UsageSummaryInput, + settings: ServerSettingsValue, + provisional: boolean, + ) { + const scan = yield* scanSummary(input, settings, provisional); + if (!scan.fromMemory) return scan; + const saved = asSaved(scan.summary); + return saved === null + ? yield* scanSummary(input, settings, false) + : { ...scan, summary: saved }; }); const inflightScans = new Map>(); - const scanKey = (input: UsageSummaryInput, settings: ServerSettingsValue): string => + // The last exact answer per window and settings; a provisional read returns it without aggregating. + const answered = new Map(); + const ANSWERED_WINDOWS = 32; + + const windowKey = (input: UsageSummaryInput, settings: ServerSettingsValue): string => JSON.stringify([ input.timeZone, input.sinceDay, @@ -901,7 +1045,6 @@ export const make = Effect.gen(function* () { input.sinceTime ?? null, input.untilTime ?? null, settings, - input.awaitRefresh === true, input.modelFilter ?? null, ]); @@ -912,7 +1055,12 @@ export const make = Effect.gen(function* () { const modelFilter = canonicalModelFilter(requestedFilter); const input = modelFilter === undefined ? unfiltered : { ...unfiltered, modelFilter }; const settings = yield* readSettings; - const key = scanKey(input, settings); + const window = windowKey(input, settings); + const provisional = input.provisional === true && input.awaitRefresh !== true; + const last = provisional ? answered.get(window) : undefined; + const saved = last === undefined ? null : asSaved(last); + if (saved !== null) return saved; + const key = `${window}\0${input.awaitRefresh === true}\0${provisional}`; const deferred = yield* Effect.uninterruptible( Effect.gen(function* () { const existing = inflightScans.get(key); @@ -923,7 +1071,23 @@ export const make = Effect.gen(function* () { inflightScans.set(key, created); // Detach so caller departure cannot cancel shared scans; finished scans warm the cache. // Register writes before releasing waiters, but start writing after delivering their summary. - yield* scanSummary(input, settings).pipe( + yield* readWindow(input, settings, provisional).pipe( + Effect.map(({ summary, unread }) => { + // An hourly window moves every minute, so its key never repeats. A scan that + // could not read every path is not kept: its answer may miss files. + if ( + input.resolution === "hour" || + unread || + summary.sources.some((source) => source.refreshing === true) + ) + return summary; + answered.delete(window); + answered.set(window, summary); + if (answered.size > ANSWERED_WINDOWS) { + answered.delete(answered.keys().next().value!); + } + return summary; + }), Effect.onExit((exit) => Effect.sync(() => inflightScans.delete(key)).pipe( Effect.andThen(Exit.isSuccess(exit) ? schedulePersist : Effect.void), diff --git a/apps/server/src/usage/usageTranscriptReader.test.ts b/apps/server/src/usage/usageTranscriptReader.test.ts index 105884d81ccfb..f4cb72ec1e25c 100644 --- a/apps/server/src/usage/usageTranscriptReader.test.ts +++ b/apps/server/src/usage/usageTranscriptReader.test.ts @@ -99,6 +99,30 @@ describe("compressed transcripts", () => { assert.isAbove(listed.files.find((file) => file.path === compressed)?.size ?? 0, 0); }); + it("lists files in directory order when more directories wait than it reads at once", async () => { + for (let index = 0; index < 40; index += 1) { + const nested = NodePath.join(dir, `project-${index}`, "sessions"); + await NodeFSP.mkdir(nested, { recursive: true }); + await NodeFSP.writeFile(NodePath.join(nested, "a.jsonl"), ""); + await NodeFSP.writeFile(NodePath.join(nested, "..", "b.jsonl"), ""); + } + // Usage that two transcripts share counts in the first one listed, so the order is part of the result. + const inOrder = async (folder: string): Promise => { + const paths: string[] = []; + for (const entry of await NodeFSP.readdir(folder, { withFileTypes: true })) { + const child = NodePath.join(folder, entry.name); + if (entry.isDirectory()) paths.push(...(await inOrder(child))); + else paths.push(child); + } + return paths; + }; + const listed = await listTranscriptFiles(dir, 0); + assert.deepStrictEqual( + listed.files.map((file) => file.path), + await inOrder(dir), + ); + }); + it.each([ { parser: "native", streamingThresholdBytes: undefined }, { parser: "projected", streamingThresholdBytes: 1024 }, diff --git a/apps/server/src/usage/usageTranscriptReader.ts b/apps/server/src/usage/usageTranscriptReader.ts index 38b80d36c8665..e20de671b75dd 100644 --- a/apps/server/src/usage/usageTranscriptReader.ts +++ b/apps/server/src/usage/usageTranscriptReader.ts @@ -1,8 +1,10 @@ // @effect-diagnostics nodeBuiltinImport:off // Cold windows span ~1.4 GB/1,500 files; node:fs buffer streaming is ~10x cheaper than materializing files. +import * as NodeFS from "node:fs"; import * as NodeFSP from "node:fs/promises"; import * as NodePath from "node:path"; import * as NodeStringDecoder from "node:string_decoder"; +import * as NodeUtil from "node:util"; import * as NodeZlib from "node:zlib"; import type { @@ -43,6 +45,9 @@ export const GUARD_LENGTH = 64; const STREAMING_THRESHOLD_BYTES = 8 * 1024 * 1024; const NEWLINE = 0x0a; const STAT_CONCURRENCY = 32; +const WALK_CONCURRENCY = 16; +// Measured about 3x faster than the promise API's stat across 20,000 transcripts. +const statFile = NodeUtil.promisify(NodeFS.stat); const CARRIAGE_RETURN = 0x0d; function selectUsageFields(fields: SelectedFields) { @@ -73,36 +78,48 @@ export async function listTranscriptFiles( options?: { readonly fileName?: string }, ): Promise<{ readonly files: readonly TranscriptFile[]; readonly failedPaths: number }> { const fileName = options?.fileName; - const candidates: string[] = []; let failedPaths = 0; - const walk = async (dir: string): Promise => { + let reading = 0; + const waiting: Array<() => void> = []; + // Sibling directories read together; each returns its own list so the result keeps readdir order. + const walk = async (dir: string): Promise => { + if (reading < WALK_CONCURRENCY) reading += 1; + else await new Promise((resume) => waiting.push(resume)); let entries; try { entries = await NodeFSP.readdir(dir, { withFileTypes: true }); } catch { failedPaths += 1; - return; - } - for (const entry of entries) { - const child = NodePath.join(dir, entry.name); - if (entry.isDirectory()) await walk(child); - else if ( - fileName !== undefined - ? entry.name === fileName - : entry.name.endsWith(".jsonl") || entry.name.endsWith(".jsonl.zst") - ) { - candidates.push(child); - } + return []; + } finally { + // A waiting directory takes over this slot, so the count only drops when none waits. + const next = waiting.shift(); + if (next === undefined) reading -= 1; + else next(); } + const found = await Promise.all( + entries.map((entry) => { + const child = NodePath.join(dir, entry.name); + if (entry.isDirectory()) return walk(child); + return ( + fileName !== undefined + ? entry.name === fileName + : entry.name.endsWith(".jsonl") || entry.name.endsWith(".jsonl.zst") + ) + ? [child] + : []; + }), + ); + return found.flat(); }; - await walk(root); + const candidates = await walk(root); const found: Array = Array.from({ length: candidates.length }); const queue = candidates.entries(); const statQueued = async (): Promise => { for (const [index, path] of queue) { try { - const stats = await NodeFSP.stat(path); + const stats = await statFile(path); if (stats.mtimeMs >= sinceMs) { found[index] = { path, size: stats.size, mtimeMs: stats.mtimeMs }; } diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 6b757ba8d15cf..4b5060be36c08 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1340,14 +1340,19 @@ export function createServerEnvironmentAtoms( tag: WS_METHODS.scheduledTasksSubscribe, }), // A cold transcript scan is measured in seconds, so keep the result around - // long enough that switching windows or re-rendering does not rescan. + // long enough that switching windows or re-rendering does not rescan. With + // nothing to show yet, ask for the server's last scan first: it answers at + // once and marks its sources refreshing, which starts the full read below. usageSummary: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:usage-summary", tag: WS_METHODS.serverGetUsageSummary, staleTimeMs: 60_000, refreshTrigger: ({ environmentId }) => usageScanSettingsAtom(environmentId), - execute: (input, emit) => - request(WS_METHODS.serverGetUsageSummary, input).pipe( + execute: (input, emit, previous) => + request( + WS_METHODS.serverGetUsageSummary, + Option.isSome(previous) ? input : { ...input, provisional: true }, + ).pipe( Effect.filterOrElse( (summary) => !summary.sources.some((source) => source.refreshing === true), (summary) => @@ -1358,7 +1363,12 @@ export function createServerEnvironmentAtoms( Effect.catch((error) => Effect.logWarning("Could not refresh slow usage sources.").pipe( Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), - Effect.as(summary), + // A saved answer is not final, so read the transcripts in full instead. + Effect.andThen( + Option.isSome(previous) + ? Effect.succeed(summary) + : request(WS_METHODS.serverGetUsageSummary, input), + ), ), ), ), diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index b16550c7c487c..552897a36344c 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -295,7 +295,15 @@ it.effect("keeps the cached usage summary when the awaited refresh fails", () => cursorSource(true), ]); expect(AsyncResult.isSuccess(harness.registry.get(summary))).toBe(true); - expect(harness.requests()).toBe(2); + // The first read takes the server's saved answer, which is not final: when the + // follow-up fails, a full read replaces it. + expect( + harness.inputs().map(({ provisional, awaitRefresh }) => ({ provisional, awaitRefresh })), + ).toEqual([ + { provisional: true, awaitRefresh: undefined }, + { provisional: undefined, awaitRefresh: true }, + { provisional: undefined, awaitRefresh: undefined }, + ]); unmount(); }), ), diff --git a/packages/contracts/src/usage.ts b/packages/contracts/src/usage.ts index b473a7264b1a4..d5de29bb8a2fa 100644 --- a/packages/contracts/src/usage.ts +++ b/packages/contracts/src/usage.ts @@ -133,6 +133,8 @@ export const UsageSummaryInput = Schema.Struct({ sinceTime: Schema.optional(TrimmedNonEmptyString), untilTime: Schema.optional(TrimmedNonEmptyString), awaitRefresh: Schema.optional(Schema.Boolean), + /** Accept an immediate answer from the last scan; its sources are marked refreshing. */ + provisional: Schema.optional(Schema.Boolean), modelFilter: Schema.optional(UsageModelFilter), }); export type UsageSummaryInput = typeof UsageSummaryInput.Type; From a32cedbb0e0d8cd25bd9234a515aaabe1a53c28b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 10:45:18 -0800 Subject: [PATCH 2/8] test(usage): trim provisional usage tests to the behaviours they prove Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.test.ts | 50 +++++----------------- 1 file changed, 11 insertions(+), 39 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 24c200220a043..a885adbffe6c9 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2410,59 +2410,34 @@ describe("UsageService", () => { const { transcript, settings, home } = yield* setup; yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 5))); const provisional = { ...WINDOW, provisional: true }; - const exact = { ...WINDOW, awaitRefresh: true }; const claude = (summary: UsageSummary) => summary.sources.find((source) => source.fingerprint.provider === "claude"); yield* Effect.gen(function* () { + // With nothing persisted, the first answer is a full scan. const service = yield* UsageService.make; - // A scan that missed a path covers nothing, so the next answer is a full scan too. const loop = NodePath.join(NodePath.dirname(transcript), "loop.jsonl"); yield* Effect.promise(() => NodeFSP.symlink(loop, loop, "junction")); assert.strictEqual(claude(yield* service.readSummary(provisional))?.status, "partial"); - assert.isUndefined(claude(yield* service.readSummary(provisional))?.refreshing); yield* Effect.promise(() => NodeFSP.rm(loop)); - // Nothing saved covers the window yet, so the first answer is a full scan. - const first = yield* service.readSummary(provisional); - assert.strictEqual(totalOutputTokens(first), 5); - assert.isUndefined(claude(first)?.refreshing); + // A scan that missed a path covers nothing, so this is a full scan too. + assert.isUndefined(claude(yield* service.readSummary(provisional))?.refreshing); + yield* service.awaitPersisted; yield* Effect.promise(() => NodeFSP.appendFile(transcript, claudeLine(2, 7))); - const saved = yield* service.readSummary(provisional); + const restarted = yield* UsageService.make; + const saved = yield* restarted.readSummary(provisional); assert.strictEqual(totalOutputTokens(saved), 5); assert.isTrue(claude(saved)?.refreshing); - assert.isTrue( - saved.sources.every((source) => source === claude(saved) || !source.refreshing), - ); - const refreshed = yield* service.readSummary(exact); - assert.strictEqual(totalOutputTokens(refreshed), 12); - assert.isUndefined(claude(refreshed)?.refreshing); - assert.deepStrictEqual( - (yield* service.readSummary(provisional)).buckets, - refreshed.buckets, - ); - yield* service.awaitPersisted; - - // A restarted server has no earlier answer, only the persisted scan. - yield* Effect.promise(() => NodeFSP.appendFile(transcript, claudeLine(3, 11))); - const restarted = yield* UsageService.make; - const restored = yield* restarted.readSummary(provisional); - assert.deepStrictEqual(restored.buckets, refreshed.buckets); - assert.isTrue(claude(restored)?.refreshing); - // A window that starts before every scan could only show part of its usage from memory. + // A window that starts before the saved scan could only show part of its usage. const wider = yield* restarted.readSummary({ ...provisional, sinceDay: UsageDay.make("2026-07-01"), }); - assert.strictEqual(totalOutputTokens(wider), 23); + assert.strictEqual(totalOutputTokens(wider), 12); assert.isUndefined(claude(wider)?.refreshing); - assert.strictEqual(totalOutputTokens(yield* restarted.readSummary(exact)), 23); - // With no saved usage to show, the answer is exact: nothing would tell the client to ask again. - const empty = yield* restarted.readSummary({ - ...provisional, - modelFilter: { include: [{ model: "absent" }] }, - }); - assert.deepStrictEqual(empty.buckets, []); - assert.isTrue(empty.sources.every((source) => !source.refreshing)); + const refreshed = yield* restarted.readSummary({ ...WINDOW, awaitRefresh: true }); + assert.strictEqual(totalOutputTokens(refreshed), 12); + assert.isUndefined(claude(refreshed)?.refreshing); yield* restarted.awaitPersisted; }).pipe( Effect.scoped, @@ -3247,13 +3222,10 @@ describe("UsageService", () => { // The narrow window saves the second copy before the wide one reads the first. yield* service.readSummary(NARROW_WINDOW); const exact = yield* service.readSummary(WINDOW); - assert.strictEqual(exact.buckets[0]?.sessions, 2); yield* service.awaitPersisted; const restarted = yield* UsageService.make; const saved = yield* restarted.readSummary({ ...WINDOW, provisional: true }); assert.deepStrictEqual(saved.buckets, exact.buckets); - yield* Effect.promise(() => Promise.all([NodeFSP.rm(first), NodeFSP.rm(second)])); - assert.deepStrictEqual((yield* restarted.readSummary(WINDOW)).buckets, exact.buckets); }).pipe( Effect.provide(layerService({ prefix: "usage-service-saved-order-test", home, settings })), ); From f6674dbf9fe47dc49aaedbaaad9648bf75cdf4cd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 10:53:32 -0800 Subject: [PATCH 3/8] refactor(usage): answer provisional reads only from the saved scan Drop the in-memory map of the last 32 exact summaries, so a provisional read always goes through directory coverage. Decide coverage before the scan starts: a covered read prices with the saved rates and never takes the rates lock, anything else is a normal exact scan. A missing directory now counts as covered, which removes the existence probe and the three-way coverage state. Sources are marked refreshing while they are built instead of in a second pass over the summary. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.ts | 233 +++++++++----------------- 1 file changed, 80 insertions(+), 153 deletions(-) diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 52e19b7bcfb3b..24b4c21deb8f4 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -168,11 +168,15 @@ export const make = Effect.gen(function* () { const correctedBatch = (records: readonly UsageRecord[]) => records.some((record) => resolvedRecords.has(record)) ? records.map(correctedRecord) : records; const sourceCache = new Map(); - // Earliest window start a full scan read each transcript directory from. A window inside it - // has all its files in fileCache except what changed since, so it can be answered from memory. + // Earliest window start a complete scan read each transcript directory from. const coveredSince = new Map(); const coverageKey = (source: TranscriptSource) => `${source.provider}\0${source.dir}`; let cacheDirty = false; + const cover = (source: TranscriptSource, windowStartMs: number) => { + if ((coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs) return; + coveredSince.set(coverageKey(source), windowStartMs); + cacheDirty = true; + }; const isWithinDirectory = (filePath: string, dir: string) => { const relative = path.relative(dir, filePath); return relative !== ".." && !relative.startsWith(".." + path.sep) && !path.isAbsolute(relative); @@ -242,32 +246,26 @@ export const make = Effect.gen(function* () { Effect.catchCause(() => Effect.succeed(null)), ); - const loadRates = Effect.fn("UsageService.loadRates")(function* ( - force: boolean, - savedOnly = false, - ) { + // Outside ratesLock, which a slow fetch holds: a provisional answer prices with these. + const loadSavedRates = yield* Effect.cached( + fileSystem.readFileString(ratesCachePath).pipe( + Effect.flatMap((raw) => decodeRatesCache(raw)), + Effect.map((saved) => { + const parsed = parseRateTable(saved.document); + if (parsed.size === 0) return; + rates = parsed; + ratesFetchedAtMs = saved.fetchedAtMs; + ratesStatus = "cached"; + }), + Effect.ignoreCause, + ), + ); + + const loadRates = Effect.fn("UsageService.loadRates")(function* (force: boolean) { const now = yield* Clock.currentTimeMillis; const maxAgeMs = force ? RATES_REFRESH_FLOOR_MS : RATES_TTL_MS; + yield* loadSavedRates; 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; - } - } - } - - // Saved rates of any age price a provisional answer; only an exact read waits on the network. - if (savedOnly && rates.size > 0) return; if (ratesLastFailedAtMs !== null && now - ratesLastFailedAtMs < RATES_REFRESH_FLOOR_MS) return; let fetched = yield* fetchRates(LITELLM_RATES_URL); @@ -291,8 +289,7 @@ export const make = Effect.gen(function* () { ); }); - const ensureRates = (force: boolean, savedOnly = false) => - ratesLock.withPermit(loadRates(force, savedOnly)); + const ensureRates = (force: boolean) => ratesLock.withPermit(loadRates(force)); const refreshRates = ensureRates(true).pipe( Effect.map(pricing), @@ -628,8 +625,11 @@ export const make = Effect.gen(function* () { const exists = yield* fileSystem .exists(dir) .pipe(Effect.catchCause(() => Effect.succeed(false))); - if (!exists) return { provider, dir, volumeId, files: null } satisfies ScannedDir; - // No walk: scanSummary aggregates this directory's saved files, as it does for deleted ones. + if (!exists) { + cover(source, windowStartMs); + return { provider, dir, volumeId, files: null } satisfies ScannedDir; + } + // scanSummary aggregates the saved files, as it does for a deleted directory. if (fromMemory) return { provider, dir, volumeId, files: [] } satisfies ScannedDir; const { files, failedPaths } = yield* Effect.promise(() => listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }), @@ -663,12 +663,7 @@ export const make = Effect.gen(function* () { return { path, records, malformedRecords }; }); const unread = failedPaths + read.filter((file) => file.failed).length; - const covered = coveredSince.get(coverageKey(source)); - // A read that missed paths covers nothing: memory would answer for files it never held. - if (unread === 0 && (covered === undefined || covered > windowStartMs)) { - coveredSince.set(coverageKey(source), windowStartMs); - cacheDirty = true; - } + if (unread === 0) cover(source, windowStartMs); yield* Effect.annotateCurrentSpan({ "usage.files": files.length, "usage.files_parsed": read.filter((file) => file.update !== undefined).length, @@ -688,34 +683,13 @@ export const make = Effect.gen(function* () { }); const collectDirs = Effect.fn("UsageService.collectDirs")(function* ( + { dirs, pending }: Effect.Success>, windowStartMs: number, settings: ServerSettingsValue, retentionCutoffMs: number, awaitRefresh: boolean, - provisional: boolean, + fromMemory: boolean, ) { - const { dirs, pending } = yield* resolveTranscriptDirs( - settings, - retentionCutoffMs, - awaitRefresh, - ).pipe(Effect.provideContext(readerContext)); - // A window no full scan has read would show partial totals, so every directory must cover it. - const coverage = provisional - ? yield* Effect.forEach(dirs, (source) => - fileSystem.exists(source.dir).pipe( - Effect.catchCause(() => Effect.succeed(false)), - Effect.map((exists) => - !exists - ? "missing" - : (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs - ? "covered" - : "uncovered", - ), - ), - ) - : []; - const fromMemory = coverage.includes("covered") && !coverage.includes("uncovered"); - const scans = Effect.forEach( scanReaders, ({ driver, reader }) => @@ -749,36 +723,9 @@ export const make = Effect.gen(function* () { { concurrency: "unbounded" }, ); const scanned: readonly ScannedDir[] = [...transcripts, ...pending, ...scanDirs.flat()]; - return { scanned, fromMemory, unread: transcripts.some((dir) => "status" in dir) }; + return scanned; }); - /** - * A saved answer misses whatever changed since its scan, so it is never final: its transcript - * sources with usage are marked refreshing, which tells the client to ask again with - * awaitRefresh. Null when none has usage: there is nothing to show early, and marking an - * empty source would only flash a loading row. - */ - const asSaved = (summary: UsageSummary): UsageSummary | null => { - const marked = new Set( - summary.sources.filter( - ({ fingerprint }) => - transcriptFormats.has(fingerprint.provider) && - summary.buckets.some( - (bucket) => - bucket.provider === fingerprint.provider && - bucket.sourcePath === fingerprint.resolvedHomePath, - ), - ), - ); - if (marked.size === 0) return null; - return { - ...summary, - sources: summary.sources.map((source) => - marked.has(source) ? { ...source, refreshing: true } : source, - ), - }; - }; - const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, @@ -830,21 +777,23 @@ export const make = Effect.gen(function* () { const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000; - const [, { scanned: scannedDirs, fromMemory, unread }] = yield* Effect.all( + const awaitRefresh = input.awaitRefresh === true; + const resolved = yield* resolveTranscriptDirs(settings, retentionCutoffMs, awaitRefresh).pipe( + Effect.provideContext(readerContext), + ); + // A window no complete scan has read would show partial totals, so every directory must cover it. + const fromMemory = + provisional && + resolved.dirs.every( + (source) => (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs, + ); + const [, scannedDirs] = yield* Effect.all( [ - // Loaded rates skip the lock, which an exact read holds through a slow fetch. - provisional && rates.size > 0 ? Effect.void : ensureRates(false, provisional), - collectDirs( - windowStartMs, - settings, - retentionCutoffMs, - input.awaitRefresh === true, - provisional, - ), + fromMemory ? loadSavedRates : ensureRates(false), + collectDirs(resolved, windowStartMs, settings, retentionCutoffMs, awaitRefresh, fromMemory), ], { concurrency: 2 }, ); - if (provisional && !fromMemory) yield* ensureRates(false); const aggregator = new UsageAggregator({ timeZone: input.timeZone, @@ -897,25 +846,16 @@ export const make = Effect.gen(function* () { }); // Usage two transcripts share counts in the first one, so saved files keep the order of the last walk. for (const files of fromMemory ? [] : filesByDir) { - const saved = files.flatMap((file) => (fileCache.has(file.path) ? [file.path] : [])); - if (saved.length < 2) continue; - const members = new Set(saved); - const expected = saved.values(); - if ( - fileCache - .keys() - .filter((filePath) => members.has(filePath)) - .every((filePath) => filePath === expected.next().value) - ) - continue; - // Reorder within the slots these files hold, so files this walk did not list keep their place. + const walked = files.flatMap((file) => (fileCache.has(file.path) ? [file.path] : [])); + const members = new Set(walked); + const saved = [...fileCache.keys()]; + let next = 0; + // Only the slots these files hold change, so files this walk did not list keep their place. + const ordered = saved.map((filePath) => (members.has(filePath) ? walked[next++]! : filePath)); + if (ordered.every((filePath, index) => filePath === saved[index])) continue; const entries = new Map(fileCache); - const walked = saved.values(); fileCache.clear(); - for (const filePath of entries.keys()) { - const slot = members.has(filePath) ? walked.next().value! : filePath; - fileCache.set(slot, entries.get(slot)!); - } + for (const filePath of ordered) fileCache.set(filePath, entries.get(filePath)!); cacheDirty = true; } const windowEndMs = @@ -973,6 +913,7 @@ export const make = Effect.gen(function* () { // Distinct per directory. Buckets carry per-cell session counts, but a // session spans days and models, so clients total this figure instead. const sessionIds = new Set(); + let counted = false; for (const file of filesByDir[index] ?? []) { malformedRecords += file.malformedRecords; @@ -1001,8 +942,9 @@ export const make = Effect.gen(function* () { eventOccurrences.set(key, occurrence); usageRecord = { ...record, dedupeKey: key + ":" + occurrence }; } - if (aggregator.add(usageRecord, dir) && record.sessionId.length > 0) { - sessionIds.add(record.sessionId); + if (aggregator.add(usageRecord, dir)) { + counted = true; + if (record.sessionId.length > 0) sessionIds.add(record.sessionId); } } } @@ -1029,14 +971,18 @@ export const make = Effect.gen(function* () { .filter(Boolean) .join(" ") : sourceMessage, - ...(refreshing ? { refreshing } : {}), + // A saved answer misses what changed since its scan, so the client must ask again. An + // empty source stays unmarked: it would only flash a loading row. + ...(refreshing || (fromMemory && counted && transcriptFormats.has(provider)) + ? { refreshing: true } + : {}), }); } const pruned = pruneScanCache(fileCache, retentionCutoffMs); if (pruned > 0) { cacheDirty = true; - // Pruned files are gone from memory, so nothing before the cutoff is covered any more. + // Pruned files are no longer saved, so nothing before the cutoff stays covered. for (const [key, since] of coveredSince) { if (since < retentionCutoffMs) coveredSince.set(key, retentionCutoffMs); } @@ -1046,7 +992,7 @@ export const make = Effect.gen(function* () { const readAt = yield* DateTime.now; const finishedAtMs = yield* Clock.currentTimeMillis; - const summary: UsageSummary = { + const summary = { contractVersion: USAGE_CONTRACT_VERSION, readAt: DateTime.formatIso(readAt), timeZone: input.timeZone, @@ -1057,8 +1003,8 @@ export const make = Effect.gen(function* () { pricing: pricing(), scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), - }; - return { summary, fromMemory, unread }; + } satisfies UsageSummary; + return { summary, fromMemory }; }); const readWindow = Effect.fn("UsageService.readWindow")(function* ( @@ -1066,21 +1012,20 @@ export const make = Effect.gen(function* () { settings: ServerSettingsValue, provisional: boolean, ) { - const scan = yield* scanSummary(input, settings, provisional); - if (!scan.fromMemory) return scan; - const saved = asSaved(scan.summary); - return saved === null - ? yield* scanSummary(input, settings, false) - : { ...scan, summary: saved }; + const { summary, fromMemory } = yield* scanSummary(input, settings, provisional); + // With nothing marked refreshing the client would keep a saved answer as final. + return fromMemory && !summary.sources.some((source) => source.refreshing === true) + ? (yield* scanSummary(input, settings, false)).summary + : summary; }); const inflightScans = new Map>(); - // The last exact answer per window and settings; a provisional read returns it without aggregating. - const answered = new Map(); - const ANSWERED_WINDOWS = 32; - - const windowKey = (input: UsageSummaryInput, settings: ServerSettingsValue): string => + const scanKey = ( + input: UsageSummaryInput, + settings: ServerSettingsValue, + provisional: boolean, + ): string => JSON.stringify([ input.timeZone, input.sinceDay, @@ -1089,6 +1034,8 @@ export const make = Effect.gen(function* () { input.sinceTime ?? null, input.untilTime ?? null, settings, + input.awaitRefresh === true, + provisional, input.modelFilter ?? null, ]); @@ -1099,12 +1046,8 @@ export const make = Effect.gen(function* () { const modelFilter = canonicalModelFilter(requestedFilter); const input = modelFilter === undefined ? unfiltered : { ...unfiltered, modelFilter }; const settings = yield* readSettings; - const window = windowKey(input, settings); const provisional = input.provisional === true && input.awaitRefresh !== true; - const last = provisional ? answered.get(window) : undefined; - const saved = last === undefined ? null : asSaved(last); - if (saved !== null) return saved; - const key = `${window}\0${input.awaitRefresh === true}\0${provisional}`; + const key = scanKey(input, settings, provisional); const deferred = yield* Effect.uninterruptible( Effect.gen(function* () { const existing = inflightScans.get(key); @@ -1116,22 +1059,6 @@ export const make = Effect.gen(function* () { // Detach so caller departure cannot cancel shared scans; finished scans warm the cache. // Register writes before releasing waiters, but start writing after delivering their summary. yield* readWindow(input, settings, provisional).pipe( - Effect.map(({ summary, unread }) => { - // An hourly window moves every minute, so its key never repeats. A scan that - // could not read every path is not kept: its answer may miss files. - if ( - input.resolution === "hour" || - unread || - summary.sources.some((source) => source.refreshing === true) - ) - return summary; - answered.delete(window); - answered.set(window, summary); - if (answered.size > ANSWERED_WINDOWS) { - answered.delete(answered.keys().next().value!); - } - return summary; - }), Effect.onExit((exit) => Effect.sync(() => inflightScans.delete(key)).pipe( Effect.andThen(Exit.isSuccess(exit) ? schedulePersist : Effect.void), From e905be07466f3492cdb410fdce4441450416a18c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 11:10:13 -0800 Subject: [PATCH 4/8] fix(usage): exact transcript figures no longer wait on slow sources after a saved answer - After a provisional answer the client reads plainly first and waits for slow sources (Cursor, Muse discovery) only if that answer still has them refreshing. - Saved files are ordered against the full walk, including files outside the scanned window, so a narrow scan cannot leave duplicates credited differently. - Coverage is trusted only from the current cache file. - Exact reads fetch rates alongside directory resolution again. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.test.ts | 35 +++++++++++ apps/server/src/usage/UsageService.ts | 58 +++++++++++++------ .../server/src/usage/usageTranscriptReader.ts | 8 ++- packages/client-runtime/src/state/server.ts | 48 ++++++++------- .../src/state/serverUsage.test.ts | 5 +- 5 files changed, 107 insertions(+), 47 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index a885adbffe6c9..0be78eaa59dd2 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -249,6 +249,7 @@ const layerService = (input: { ); const BEFORE_NARROW_WINDOW = Date.parse("2026-08-01T10:00:00Z") / 1000; +const NARROW_WINDOW_MTIME = Date.parse("2026-09-01T10:00:00Z") / 1000; const NARROW_WINDOW: UsageSummaryInput = { timeZone: "UTC", sinceDay: UsageDay.make("2026-09-01"), @@ -2438,6 +2439,15 @@ describe("UsageService", () => { const refreshed = yield* restarted.readSummary({ ...WINDOW, awaitRefresh: true }); assert.strictEqual(totalOutputTokens(refreshed), 12); assert.isUndefined(claude(refreshed)?.refreshing); + // Saved usage with nothing to show marks nothing refreshing, so it must not be the answer. + yield* Effect.promise(() => + NodeFSP.appendFile(transcript, claudeLine(3, 11, "claude-other")), + ); + const unseen = yield* restarted.readSummary({ + ...provisional, + modelFilter: { include: [{ model: "claude-other" }] }, + }); + assert.strictEqual(totalOutputTokens(unseen), 11); yield* restarted.awaitPersisted; }).pipe( Effect.scoped, @@ -3229,6 +3239,31 @@ describe("UsageService", () => { }).pipe( Effect.provide(layerService({ prefix: "usage-service-saved-order-test", home, settings })), ); + + // A narrow scan that first saves the earlier copy still puts it before a later file it skipped. + yield* Effect.promise(async () => { + await NodeFSP.writeFile(first, forked(claudeLine(1, 5)) + forked(claudeLine(2, 7))); + await NodeFSP.utimes( + first, + Date.parse("2026-07-01T00:00:00Z") / 1000, + Date.parse("2026-07-01T00:00:00Z") / 1000, + ); + await NodeFSP.writeFile(second, claudeLine(1, 5)); + await NodeFSP.utimes(second, BEFORE_NARROW_WINDOW, BEFORE_NARROW_WINDOW); + }); + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + yield* service.readSummary(WINDOW); + yield* Effect.promise(() => + NodeFSP.utimes(first, NARROW_WINDOW_MTIME, NARROW_WINDOW_MTIME), + ); + yield* service.readSummary(NARROW_WINDOW); + const saved = yield* service.readSummary({ ...WINDOW, provisional: true }); + const exact = yield* service.readSummary(WINDOW); + assert.deepStrictEqual(saved.buckets, exact.buckets); + }).pipe( + Effect.provide(layerService({ prefix: "usage-service-narrow-order-test", home, settings })), + ); }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 24b4c21deb8f4..4991a49fe8b4a 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -406,6 +406,7 @@ export const make = Effect.gen(function* () { Effect.catchCause(() => Effect.succeed(null)), ); let document = yield* readDocument(scanCachePath); + const current = document !== null; for (const fileName of LEGACY_SCAN_CACHE_FILE_NAMES) { if (document !== null) break; document = yield* readDocument(path.join(config.stateDir, fileName)); @@ -419,8 +420,8 @@ export const make = Effect.gen(function* () { if (Option.isSome(sources)) { for (const [key, source] of Object.entries(sources.value.sources)) sourceCache.set(key, source); - // Coverage describes the saved files, so a rejected document keeps none of it. - if (fileCache.size > 0) { + // Coverage describes the saved files; an older version's files may parse differently. + if (current && fileCache.size > 0) { for (const [key, since] of Object.entries(sources.value.coveredSince ?? {})) coveredSince.set(key, since); } @@ -613,6 +614,8 @@ export const make = Effect.gen(function* () { readonly malformedRecords: number; }[] | null; + /** Every transcript path the walk listed, in walk order, whatever its mtime. */ + readonly walked?: readonly string[]; readonly refreshing?: true; } @@ -631,7 +634,7 @@ export const make = Effect.gen(function* () { } // scanSummary aggregates the saved files, as it does for a deleted directory. if (fromMemory) return { provider, dir, volumeId, files: [] } satisfies ScannedDir; - const { files, failedPaths } = yield* Effect.promise(() => + const { files, walked, failedPaths } = yield* Effect.promise(() => listTranscriptFiles(dir, windowStartMs, fileName === undefined ? undefined : { fileName }), ); const read = yield* Effect.forEach( @@ -673,6 +676,7 @@ export const make = Effect.gen(function* () { dir, volumeId, files: parsedFiles, + walked, ...(unread > 0 ? { status: "partial", @@ -778,8 +782,14 @@ export const make = Effect.gen(function* () { const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000; const awaitRefresh = input.awaitRefresh === true; - const resolved = yield* resolveTranscriptDirs(settings, retentionCutoffMs, awaitRefresh).pipe( - Effect.provideContext(readerContext), + const [resolved] = yield* Effect.all( + [ + resolveTranscriptDirs(settings, retentionCutoffMs, awaitRefresh).pipe( + Effect.provideContext(readerContext), + ), + provisional ? Effect.void : ensureRates(false), + ], + { concurrency: 2 }, ); // A window no complete scan has read would show partial totals, so every directory must cover it. const fromMemory = @@ -789,7 +799,7 @@ export const make = Effect.gen(function* () { ); const [, scannedDirs] = yield* Effect.all( [ - fromMemory ? loadSavedRates : ensureRates(false), + fromMemory ? loadSavedRates : provisional ? ensureRates(false) : Effect.void, collectDirs(resolved, windowStartMs, settings, retentionCutoffMs, awaitRefresh, fromMemory), ], { concurrency: 2 }, @@ -811,12 +821,10 @@ export const make = Effect.gen(function* () { // Deleted transcripts' saved usage still aggregates and deduplicates normally. const retainedSinceMs = Math.max(windowStartMs, retentionCutoffMs); - const filesByDir = scannedDirs.map(({ provider, dir, files }) => { - const retainedFiles = [...(files ?? [])]; - const livePaths = new Set(retainedFiles.map((file) => file.path)); - // Saved paths are files below a scanned root, so their folder decides; folders repeat thousands of times. + // Saved paths are files below a scanned root, so their folder decides; folders repeat thousands of times. + const withinDir = (dir: string) => { const foldersWithin = new Map(); - const isRetained = (filePath: string) => { + return (filePath: string) => { const folder = path.dirname(filePath); let within = foldersWithin.get(folder); if (within === undefined) { @@ -825,6 +833,11 @@ export const make = Effect.gen(function* () { } return within; }; + }; + const filesByDir = scannedDirs.map(({ provider, dir, files }) => { + const retainedFiles = [...(files ?? [])]; + const livePaths = new Set(retainedFiles.map((file) => file.path)); + const isRetained = withinDir(dir); for (const [filePath, entry] of fileCache) { if ( entry.provider !== provider || @@ -844,14 +857,25 @@ export const make = Effect.gen(function* () { } return retainedFiles; }); - // Usage two transcripts share counts in the first one, so saved files keep the order of the last walk. - for (const files of fromMemory ? [] : filesByDir) { - const walked = files.flatMap((file) => (fileCache.has(file.path) ? [file.path] : [])); - const members = new Set(walked); + // Usage two transcripts share counts in the first one, so saved files take the order an exact + // scan reads them in: walked files in walk order, then the directory's deleted files. + for (const { provider, dir, walked } of fromMemory ? [] : scannedDirs) { + if (walked === undefined) continue; + const within = withinDir(dir); const saved = [...fileCache.keys()]; + const members = new Set( + saved.filter( + (filePath) => fileCache.get(filePath)!.provider === provider && within(filePath), + ), + ); + if (members.size < 2) continue; + const listed = new Set(walked); + const target = [ + ...walked.filter((filePath) => members.has(filePath)), + ...saved.filter((filePath) => members.has(filePath) && !listed.has(filePath)), + ]; let next = 0; - // Only the slots these files hold change, so files this walk did not list keep their place. - const ordered = saved.map((filePath) => (members.has(filePath) ? walked[next++]! : filePath)); + const ordered = saved.map((filePath) => (members.has(filePath) ? target[next++]! : filePath)); if (ordered.every((filePath, index) => filePath === saved[index])) continue; const entries = new Map(fileCache); fileCache.clear(); diff --git a/apps/server/src/usage/usageTranscriptReader.ts b/apps/server/src/usage/usageTranscriptReader.ts index e20de671b75dd..d80c51964a20c 100644 --- a/apps/server/src/usage/usageTranscriptReader.ts +++ b/apps/server/src/usage/usageTranscriptReader.ts @@ -76,7 +76,11 @@ export async function listTranscriptFiles( root: string, sinceMs: number, options?: { readonly fileName?: string }, -): Promise<{ readonly files: readonly TranscriptFile[]; readonly failedPaths: number }> { +): Promise<{ + readonly files: readonly TranscriptFile[]; + readonly walked: readonly string[]; + readonly failedPaths: number; +}> { const fileName = options?.fileName; let failedPaths = 0; let reading = 0; @@ -133,7 +137,7 @@ export async function listTranscriptFiles( await Promise.all( Array.from({ length: Math.min(STAT_CONCURRENCY, candidates.length) }, statQueued), ); - return { files: found.filter((file) => file !== undefined), failedPaths }; + return { files: found.filter((file) => file !== undefined), walked: candidates, failedPaths }; } export async function readDirectoryVolumeId(path: string): Promise { diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 622dc46f3159b..9c3ea794cfe76 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -9,6 +9,7 @@ import { type ServerSelfUpdateProgressEvent, type ServerSelfUpdateResult, shouldRecommendDatabaseCompaction, + type UsageSummary, WS_METHODS, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -1498,39 +1499,36 @@ export function createServerEnvironmentAtoms( }), // A cold transcript scan is measured in seconds, so keep the result around // long enough that switching windows or re-rendering does not rescan. With - // nothing to show yet, ask for the server's last scan first: it answers at - // once and marks its sources refreshing, which starts the full read below. + // nothing to show yet, the server's last scan answers first, then a plain + // read, then a read that waits for slow sources. usageSummary: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:usage-summary", tag: WS_METHODS.serverGetUsageSummary, staleTimeMs: 60_000, refreshTrigger: ({ environmentId }) => usageScanSettingsAtom(environmentId), - execute: (input, emit, previous) => - request( - WS_METHODS.serverGetUsageSummary, - Option.isSome(previous) ? input : { ...input, provisional: true }, - ).pipe( - Effect.filterOrElse( - (summary) => !summary.sources.some((source) => source.refreshing === true), - (summary) => - emit(summary).pipe( - Effect.andThen( - request(WS_METHODS.serverGetUsageSummary, { ...input, awaitRefresh: true }), - ), - Effect.catch((error) => - Effect.logWarning("Could not refresh slow usage sources.").pipe( - Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), - // A saved answer is not final, so read the transcripts in full instead. - Effect.andThen( - Option.isSome(previous) - ? Effect.succeed(summary) - : request(WS_METHODS.serverGetUsageSummary, input), - ), - ), + execute: (input, emit, previous) => { + const settled = (summary: UsageSummary) => + !summary.sources.some((source) => source.refreshing === true); + const read = request(WS_METHODS.serverGetUsageSummary, input).pipe( + Effect.filterOrElse(settled, (summary) => + emit(summary).pipe( + Effect.andThen( + request(WS_METHODS.serverGetUsageSummary, { ...input, awaitRefresh: true }), + ), + Effect.catch((error) => + Effect.logWarning("Could not refresh slow usage sources.").pipe( + Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), + Effect.as(summary), ), ), + ), ), - ), + ); + if (Option.isSome(previous)) return read; + return request(WS_METHODS.serverGetUsageSummary, { ...input, provisional: true }).pipe( + Effect.filterOrElse(settled, (saved) => emit(saved).pipe(Effect.andThen(read))), + ); + }, }), resourceTelemetry: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:server:resource-telemetry", diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index 552897a36344c..0bfcae19c6641 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -295,14 +295,13 @@ it.effect("keeps the cached usage summary when the awaited refresh fails", () => cursorSource(true), ]); expect(AsyncResult.isSuccess(harness.registry.get(summary))).toBe(true); - // The first read takes the server's saved answer, which is not final: when the - // follow-up fails, a full read replaces it. + // The saved answer is followed by a plain read before the one that waits for Cursor. expect( harness.inputs().map(({ provisional, awaitRefresh }) => ({ provisional, awaitRefresh })), ).toEqual([ { provisional: true, awaitRefresh: undefined }, - { provisional: undefined, awaitRefresh: true }, { provisional: undefined, awaitRefresh: undefined }, + { provisional: undefined, awaitRefresh: true }, ]); unmount(); }), From 712c1d69b71c5405f2d04ad940ccd0db735cba89 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 11:24:26 -0800 Subject: [PATCH 5/8] fix(usage): transcript scans no longer wait for a rates fetch Exact reads load rates alongside the whole resolve-and-scan branch again, and a read with no transcript directories is never answered from memory. The gated custom-price test now serves real rates, so a scan blocked behind the fetch fails it instead of passing after the fetch times out. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.test.ts | 14 ++++++- apps/server/src/usage/UsageService.ts | 47 +++++++++++++--------- 2 files changed, 40 insertions(+), 21 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 0be78eaa59dd2..995691668d79b 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -3235,6 +3235,7 @@ describe("UsageService", () => { yield* service.awaitPersisted; const restarted = yield* UsageService.make; const saved = yield* restarted.readSummary({ ...WINDOW, provisional: true }); + assert.isTrue(saved.sources.some((source) => source.refreshing === true)); assert.deepStrictEqual(saved.buckets, exact.buckets); }).pipe( Effect.provide(layerService({ prefix: "usage-service-saved-order-test", home, settings })), @@ -3259,6 +3260,7 @@ describe("UsageService", () => { ); yield* service.readSummary(NARROW_WINDOW); const saved = yield* service.readSummary({ ...WINDOW, provisional: true }); + assert.isTrue(saved.sources.some((source) => source.refreshing === true)); const exact = yield* service.readSummary(WINDOW); assert.deepStrictEqual(saved.buckets, exact.buckets); }).pipe( @@ -3408,7 +3410,14 @@ describe("UsageService", () => { HttpClient.HttpClient, HttpClient.make((request) => Deferred.await(releaseRates).pipe( - Effect.as(HttpClientResponse.fromWeb(request, Response.json({}))), + Effect.as( + HttpClientResponse.fromWeb( + request, + Response.json({ + "example-model": { input_cost_per_token: 1e-6, output_cost_per_token: 1e-5 }, + }), + ), + ), ), ), ), @@ -3427,7 +3436,8 @@ describe("UsageService", () => { const original = yield* Fiber.join(first); const updated = yield* Fiber.join(second); - assert.strictEqual(original.buckets[0]?.costUsd, 0); + // Fetched rates price the first read, so scans must start while the fetch is pending. + assert.isAbove(original.buckets[0]?.costUsd ?? 0, 0); assert.closeTo(updated.buckets[0]?.costUsd ?? -1, 0.00006, 1e-12); }).pipe( Effect.provide(layerService({ prefix: "usage-service-price-race-test", home, settings })), diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 4991a49fe8b4a..40411138a6848 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -782,26 +782,35 @@ export const make = Effect.gen(function* () { const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000; const awaitRefresh = input.awaitRefresh === true; - const [resolved] = yield* Effect.all( - [ - resolveTranscriptDirs(settings, retentionCutoffMs, awaitRefresh).pipe( - Effect.provideContext(readerContext), - ), - provisional ? Effect.void : ensureRates(false), - ], - { concurrency: 2 }, - ); - // A window no complete scan has read would show partial totals, so every directory must cover it. - const fromMemory = - provisional && - resolved.dirs.every( - (source) => (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs, + const collect = Effect.gen(function* () { + const resolved = yield* resolveTranscriptDirs(settings, retentionCutoffMs, awaitRefresh).pipe( + Effect.provideContext(readerContext), ); - const [, scannedDirs] = yield* Effect.all( - [ - fromMemory ? loadSavedRates : provisional ? ensureRates(false) : Effect.void, - collectDirs(resolved, windowStartMs, settings, retentionCutoffMs, awaitRefresh, fromMemory), - ], + // A window no complete scan has read would show partial totals, so every directory must cover it. + const fromMemory = + provisional && + resolved.dirs.length > 0 && + resolved.dirs.every( + (source) => (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs, + ); + const [, scanned] = yield* Effect.all( + [ + fromMemory ? loadSavedRates : provisional ? ensureRates(false) : Effect.void, + collectDirs( + resolved, + windowStartMs, + settings, + retentionCutoffMs, + awaitRefresh, + fromMemory, + ), + ], + { concurrency: 2 }, + ); + return { scannedDirs: scanned, fromMemory }; + }); + const [{ scannedDirs, fromMemory }] = yield* Effect.all( + [collect, provisional ? Effect.void : ensureRates(false)], { concurrency: 2 }, ); From d19927fdf3edfa82105977cc208f19829b953a34 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 11:33:49 -0800 Subject: [PATCH 6/8] test(client-runtime): a usage refresh with figures shown skips the saved answer Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/state/serverUsage.test.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index 0bfcae19c6641..7baf02d9f1898 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -303,6 +303,15 @@ it.effect("keeps the cached usage summary when the awaited refresh fails", () => { provisional: undefined, awaitRefresh: undefined }, { provisional: undefined, awaitRefresh: true }, ]); + // With figures on screen, a refresh reads them exactly instead of asking for saved ones. + yield* harness.updateSettings({ + ...DEFAULT_SERVER_SETTINGS, + usagePriceOverrides: { + "custom-model": { inputCostPerMillionTokens: 5, outputCostPerMillionTokens: 10 }, + }, + }); + yield* waitForCost(harness.registry, summary, 5); + expect(harness.inputs()[3]?.provisional).toBeUndefined(); unmount(); }), ), From e754e7f6e6981b16cf61731cee4f53d425f0ac5b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 11:37:16 -0800 Subject: [PATCH 7/8] test(client-runtime): the usage fallback test tells the saved answer from the plain one Co-Authored-By: Claude Opus 5.5 --- packages/client-runtime/src/state/serverUsage.test.ts | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index 7baf02d9f1898..dd55f3e1e6fb5 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -232,7 +232,13 @@ const makeRefreshingHarness = Effect.fn("ServerUsageTest.makeRefreshingHarness") input.awaitRefresh === true ? Deferred.succeed(followUpStarted, undefined).pipe(Effect.andThen(followUp)) : Effect.void, - (input) => [cursorSource(input.awaitRefresh !== true)], + // The saved answer differs from the plain one, so a test can tell which was kept. + (input) => [ + { + ...cursorSource(input.awaitRefresh !== true), + ...(input.provisional === true ? { scannedFiles: 0 } : {}), + }, + ], ); return { ...harness, followUpStarted }; }); From 7e3e1297539499647b2d5eb531e052eb4acf1b5d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 12:23:50 -0800 Subject: [PATCH 8/8] fix(usage): saved usage answers need a fully readable cache and separate directories - Coverage is restored only when every saved file decoded, so a partly corrupt cache no longer answers with part of its usage. - Two providers reading the same directory never answer from memory, since the path-keyed cache holds only one provider's parse of each file. - A failed saved-answer request falls back to the plain read. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.test.ts | 17 +++++++++++++++++ apps/server/src/usage/UsageService.ts | 14 +++++++++----- packages/client-runtime/src/state/server.ts | 7 ++++++- .../src/state/serverUsage.test.ts | 19 +++++++++++++++++++ 4 files changed, 51 insertions(+), 6 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index b2ba2bf52bca2..281884dcc3bc2 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2495,6 +2495,23 @@ describe("UsageService", () => { }); assert.strictEqual(totalOutputTokens(unseen), 11); yield* restarted.awaitPersisted; + + // A saved scan with an entry this server cannot read covers nothing. + const { stateDir } = yield* ServerConfig.ServerConfig; + const cachePath = NodePath.join(stateDir, SCAN_CACHE_FILE_NAME); + yield* Effect.promise(async () => { + const document = decodeUnknownJsonString(await NodeFSP.readFile(cachePath, "utf8")) as { + files: Record>; + }; + const [entry] = Object.values(document.files); + document.files[NodePath.join(NodePath.dirname(transcript), "broken.jsonl")] = { + ...entry, + r: "broken", + }; + await NodeFSP.writeFile(cachePath, JSON.stringify(document)); + }); + const reloaded = yield* UsageService.make; + assert.isUndefined(claude(yield* reloaded.readSummary(provisional))?.refreshing); }).pipe( Effect.scoped, Effect.provide(layerService({ prefix: "usage-service-provisional", home, settings })), diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 5dfb0a88ca8e6..cd04e4bc4c1c9 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -413,15 +413,17 @@ export const make = Effect.gen(function* () { cacheDirty = document !== null; } if (document === null) return; - for (const [path, entry] of decodeScanCache(document, transcriptFormats)) { - fileCache.set(path, entry); - } + const decoded = decodeScanCache(document, transcriptFormats); + for (const [path, entry] of decoded) fileCache.set(path, entry); + const saved = (document as { readonly files?: unknown }).files; + const complete = + typeof saved === "object" && saved !== null && Object.keys(saved).length === decoded.size; const sources = decodeCachedSources(document); if (Option.isSome(sources)) { for (const [key, source] of Object.entries(sources.value.sources)) sourceCache.set(key, source); - // Coverage describes the saved files; an older version's files may parse differently. - if (current && fileCache.size > 0) { + // Coverage describes the saved files, so it needs every one, in this version's format. + if (current && complete && fileCache.size > 0) { for (const [key, since] of Object.entries(sources.value.coveredSince ?? {})) coveredSince.set(key, since); } @@ -794,9 +796,11 @@ export const make = Effect.gen(function* () { Effect.provideContext(readerContext), ); // A window no complete scan has read would show partial totals, so every directory must cover it. + // Saved files are keyed by path, so two providers reading one directory keep only one parse. const fromMemory = provisional && resolved.dirs.length > 0 && + new Set(resolved.dirs.map((source) => source.dir)).size === resolved.dirs.length && resolved.dirs.every( (source) => (coveredSince.get(coverageKey(source)) ?? Infinity) <= windowStartMs, ); diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 9c3ea794cfe76..6162df0809c83 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1526,7 +1526,12 @@ export function createServerEnvironmentAtoms( ); if (Option.isSome(previous)) return read; return request(WS_METHODS.serverGetUsageSummary, { ...input, provisional: true }).pipe( - Effect.filterOrElse(settled, (saved) => emit(saved).pipe(Effect.andThen(read))), + Effect.matchEffect({ + // The saved answer only speeds things up; without it the plain read still answers. + onFailure: () => read, + onSuccess: (saved) => + settled(saved) ? Effect.succeed(saved) : emit(saved).pipe(Effect.andThen(read)), + }), ); }, }), diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index dd55f3e1e6fb5..9d12d1114f896 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -322,3 +322,22 @@ it.effect("keeps the cached usage summary when the awaited refresh fails", () => }), ), ); + +it.effect("reads usage in full when the saved answer cannot be read", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness((_request, input) => + input.provisional === true + ? Effect.fail( + new UsageReadError({ reason: "scanFailed", detail: "saved scan unreadable" }), + ) + : Effect.void, + ); + const summary = harness.summary(); + const unmount = harness.registry.mount(summary); + yield* settledSummary(harness.registry, summary); + expect(harness.inputs().map(({ provisional }) => provisional)).toEqual([true, undefined]); + unmount(); + }), + ), +);