Skip to content

Commit 4fdeb43

Browse files
maria-rcksmaria-rcks
committed
fix(usage): preserve saved totals after transcript cleanup
1 parent 4749035 commit 4fdeb43

4 files changed

Lines changed: 132 additions & 143 deletions

File tree

‎apps/server/src/usage/UsageService.test.ts‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,9 @@ import * as NodePath from "node:path";
77
import { assert, describe, it } from "@effect/vitest";
88
import * as NodeServices from "@effect/platform-node/NodeServices";
99
import { HostProcessEnvironment } from "@t3tools/shared/hostProcess";
10+
import { mergeUsage } from "@t3tools/shared/usageMerge";
1011
import {
12+
EnvironmentId,
1113
ProviderDriverKind,
1214
ProviderInstanceId,
1315
UsageDay,
@@ -199,6 +201,20 @@ describe("UsageService", () => {
199201
);
200202
const summary = yield* service.readSummary(WINDOW);
201203
assert.strictEqual(totalOutputTokens(summary), 36);
204+
yield* Effect.promise(() =>
205+
NodeFSP.rename(
206+
NodePath.join(codexHome, "sessions", "rollout.jsonl"),
207+
NodePath.join(codexHome, "sessions", "moved.jsonl"),
208+
),
209+
);
210+
const moved = yield* service.readSummary(WINDOW);
211+
assert.deepStrictEqual(moved.buckets, summary.buckets);
212+
yield* Effect.promise(() =>
213+
NodeFSP.rm(NodePath.join(codexHome, "sessions"), { recursive: true }),
214+
);
215+
const removed = yield* service.readSummary(WINDOW);
216+
assert.deepStrictEqual(removed.buckets, summary.buckets);
217+
202218
const sources = summary.sources.filter((source) => source.status === "ok");
203219
assert.strictEqual(sources.length, 4);
204220
assert.strictEqual(
@@ -393,6 +409,75 @@ describe("UsageService", () => {
393409
}).pipe(Effect.scoped),
394410
);
395411

412+
it.live("preserves saved tokens, costs and sessions after transcript cleanup and restart", () =>
413+
Effect.gen(function* () {
414+
const { transcript, settings, home } = yield* setup;
415+
const content = claudeLine(1, 5);
416+
yield* Effect.promise(() => NodeFSP.writeFile(transcript, content));
417+
yield* Effect.gen(function* () {
418+
const service = yield* UsageService.make;
419+
const first = yield* service.readSummary(WINDOW);
420+
assert.strictEqual(totalOutputTokens(first), 5);
421+
assert.isAbove(first.buckets[0]?.costUsd ?? 0, 0);
422+
423+
yield* Effect.promise(() => NodeFSP.rm(transcript));
424+
const deleted = yield* service.readSummary(WINDOW);
425+
assert.deepStrictEqual(deleted.buckets, first.buckets);
426+
assert.deepStrictEqual(deleted.sources, first.sources);
427+
428+
const restarted = yield* UsageService.make;
429+
const restored = yield* restarted.readSummary(WINDOW);
430+
assert.deepStrictEqual(restored.buckets, first.buckets);
431+
assert.deepStrictEqual(restored.sources, first.sources);
432+
433+
// A moved transcript must not count the saved usage twice.
434+
yield* Effect.promise(() => NodeFSP.writeFile(transcript + ".jsonl", content));
435+
const moved = yield* restarted.readSummary(WINDOW);
436+
assert.deepStrictEqual(moved.buckets, first.buckets);
437+
assert.strictEqual(moved.sources[0]?.distinctSessions, 1);
438+
439+
yield* Effect.promise(() =>
440+
NodeFSP.rm(NodePath.join(home, "claude", "projects"), { recursive: true }),
441+
);
442+
const missingRoot = yield* restarted.readSummary(WINDOW);
443+
assert.deepStrictEqual(missingRoot.buckets, first.buckets);
444+
assert.strictEqual(missingRoot.sources[0]?.distinctSessions, 1);
445+
assert.strictEqual(missingRoot.sources[0]?.status, "ok");
446+
const merged = mergeUsage(
447+
[
448+
{
449+
environmentId: EnvironmentId.make("cleanup-test"),
450+
label: "test",
451+
summary: missingRoot,
452+
},
453+
],
454+
missingRoot.contractVersion,
455+
);
456+
assert.strictEqual(merged.outputTokens, 5);
457+
assert.strictEqual(merged.sessions, 1);
458+
assert.strictEqual(merged.costUsd, first.buckets[0]?.costUsd);
459+
460+
const outsideWindow = yield* restarted.readSummary({
461+
...WINDOW,
462+
sinceDay: UsageDay.make("2026-08-02"),
463+
});
464+
assert.deepStrictEqual(outsideWindow.buckets, []);
465+
assert.strictEqual(outsideWindow.sources[0]?.distinctSessions, 0);
466+
}).pipe(
467+
Effect.provide(
468+
serviceLayers({
469+
prefix: "usage-service-cleanup-test",
470+
home,
471+
settings,
472+
ratesDocument: {
473+
"claude-fable-5": { input_cost_per_token: 1e-5, output_cost_per_token: 5e-5 },
474+
},
475+
}),
476+
),
477+
);
478+
}).pipe(Effect.scoped),
479+
);
480+
396481
it.live("does not share an in-flight scan after custom prices change", () =>
397482
Effect.gen(function* () {
398483
const { transcript, settings, home } = yield* setup;

‎apps/server/src/usage/UsageService.ts‎

Lines changed: 41 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,7 @@ const encodeRatesCache = Schema.encodeEffect(
101101
const ScanCacheJson = Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec<unknown>);
102102
const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);
103103
const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson);
104+
const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson);
104105

105106
export class UsageService extends Context.Service<
106107
UsageService,
@@ -366,7 +367,8 @@ export const make = Effect.gen(function* () {
366367
);
367368
// A read failure is not an empty transcript: caching it under this
368369
// (size, mtime) would silently drop the file's usage until it changes.
369-
if (parsed === null) return [];
370+
if (parsed === null)
371+
return cached?.provider === provider ? [...cached.records, ...cached.tailRecords] : [];
370372

371373
// Stored already de-duplicated within the file, which is 99% of all
372374
// duplicates. The aggregator still runs the cross-file dedupe pass. One
@@ -499,33 +501,34 @@ export const make = Effect.gen(function* () {
499501
priceOverrides: createOverrideRateTable(settings.usagePriceOverrides),
500502
});
501503

504+
const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000;
502505
const sources: UsageSource[] = [];
503-
const livePaths = new Set<string>();
504-
const walkedRoots: string[] = [];
505506

506507
for (const { provider, dir, volumeId, files } of scannedDirs) {
507-
if (files === null) {
508-
sources.push({
509-
fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId },
510-
status: "missing",
511-
scannedFiles: 0,
512-
skippedFiles: 0,
513-
malformedRecords: 0,
514-
distinctSessions: 0,
515-
message: "No transcript directory on this environment.",
516-
});
517-
continue;
508+
const retainedFiles = [...(files ?? [])];
509+
const livePaths = new Set(retainedFiles.map((file) => file.path));
510+
// Cleanup may remove transcripts, but the usage we already saved still
511+
// contributes to this source. Keep the normal aggregation and dedupe path.
512+
for (const [filePath, entry] of fileCache) {
513+
const relative = path.relative(dir, filePath);
514+
if (
515+
entry.provider !== provider ||
516+
entry.mtimeMs < retentionCutoffMs ||
517+
livePaths.has(filePath) ||
518+
relative === ".." ||
519+
relative.startsWith(".." + path.sep) ||
520+
path.isAbsolute(relative)
521+
)
522+
continue;
523+
retainedFiles.push({ path: filePath, records: [...entry.records, ...entry.tailRecords] });
518524
}
519-
520-
walkedRoots.push(dir);
521525
let scannedFiles = 0;
522526
let skippedFiles = 0;
523527
// Distinct per directory. Buckets carry per-cell session counts, but a
524528
// session spans days and models, so clients total this figure instead.
525529
const sessionIds = new Set<string>();
526530

527-
for (const file of files) {
528-
livePaths.add(file.path);
531+
for (const file of retainedFiles) {
529532
if (file.records.length === 0) {
530533
skippedFiles += 1;
531534
continue;
@@ -534,29 +537,40 @@ export const make = Effect.gen(function* () {
534537
for (const record of file.records) {
535538
// Only sessions that contributed in-window count: the mtime slack
536539
// admits boundary files whose records fall outside the range.
537-
if (aggregator.add(record) && record.sessionId.length > 0) {
540+
// A moved Codex rollout can coexist with its saved copy. Codex records
541+
// have no provider event id, so identify the same session event here.
542+
const usageRecord =
543+
record.provider === "codex" && record.sessionId.length > 0
544+
? {
545+
...record,
546+
dedupeKey: encodeUsageRecordKey([
547+
record.provider,
548+
record.sessionId,
549+
record.timestampMs,
550+
record.model,
551+
record.totals,
552+
]),
553+
}
554+
: record;
555+
if (aggregator.add(usageRecord) && record.sessionId.length > 0) {
538556
sessionIds.add(record.sessionId);
539557
}
540558
}
541559
}
542560

543561
sources.push({
544562
fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId },
545-
status: "ok",
563+
// Clients exclude missing sources, so saved records remain an available source.
564+
status: files === null && scannedFiles === 0 ? "missing" : "ok",
546565
scannedFiles,
547566
skippedFiles,
548567
malformedRecords: 0,
549568
distinctSessions: sessionIds.size,
550-
message: null,
569+
message: files === null ? "No transcript directory on this environment." : null,
551570
});
552571
}
553572

554-
const pruned = pruneScanCache(fileCache, {
555-
livePaths,
556-
walkedRoots,
557-
windowStartMs,
558-
retentionCutoffMs: startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000,
559-
});
573+
const pruned = pruneScanCache(fileCache, retentionCutoffMs);
560574
if (pruned > 0) cacheDirty = true;
561575
yield* persistScanCache();
562576

‎apps/server/src/usage/usageScanCache.test.ts‎

Lines changed: 3 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -194,88 +194,17 @@ describe("pruneScanCache", () => {
194194
it("drops entries older than retention", () => {
195195
const cache = cacheWith([["/old.jsonl", 500, [record()]]]);
196196

197-
const removed = pruneScanCache(cache, {
198-
livePaths: new Set(),
199-
walkedRoots: ["/"],
200-
windowStartMs: 400,
201-
retentionCutoffMs,
202-
});
197+
const removed = pruneScanCache(cache, retentionCutoffMs);
203198

204199
expect(removed).toBe(1);
205200
expect(cache.size).toBe(0);
206201
});
207202

208-
it("drops in-window entries whose file has disappeared", () => {
203+
it("keeps entries whose file has disappeared", () => {
209204
const cache = cacheWith([["/gone.jsonl", 5000, [record()]]]);
210205

211-
pruneScanCache(cache, {
212-
livePaths: new Set(),
213-
walkedRoots: ["/"],
214-
windowStartMs: 4000,
215-
retentionCutoffMs,
216-
});
217-
218-
expect(cache.size).toBe(0);
219-
});
220-
221-
it("keeps entries outside the walked window that are still within retention", () => {
222-
// Viewing 7 days must not evict the 30-day entries, which that walk never
223-
// looked for and so cannot prove are gone.
224-
const cache = cacheWith([["/older-but-valid.jsonl", 2000, [record()]]]);
225-
226-
const removed = pruneScanCache(cache, {
227-
livePaths: new Set(),
228-
walkedRoots: ["/"],
229-
windowStartMs: 4000,
230-
retentionCutoffMs,
231-
});
232-
233-
expect(removed).toBe(0);
234-
expect(cache.size).toBe(1);
235-
});
236-
237-
it("keeps entries the walk saw", () => {
238-
const cache = cacheWith([["/live.jsonl", 5000, [record()]]]);
239-
240-
pruneScanCache(cache, {
241-
livePaths: new Set(["/live.jsonl"]),
242-
walkedRoots: ["/"],
243-
windowStartMs: 4000,
244-
retentionCutoffMs,
245-
});
246-
247-
expect(cache.size).toBe(1);
248-
});
249-
});
250-
251-
describe("pruneScanCache with an unwalked root", () => {
252-
it("keeps in-window entries for a provider whose directory was not walked", () => {
253-
// A missing provider root or failed settings read leaves livePaths without
254-
// that provider's files. Its warm entries must survive the pass.
255-
const cache = cacheWith([["/codex/sessions/a.jsonl", 5000, [record()]]]);
256-
257-
const removed = pruneScanCache(cache, {
258-
livePaths: new Set(),
259-
walkedRoots: ["/claude/projects"],
260-
windowStartMs: 4000,
261-
retentionCutoffMs: 1000,
262-
});
263-
264-
expect(removed).toBe(0);
265-
expect(cache.size).toBe(1);
266-
});
267-
268-
it("keeps entries under a sibling path that only shares the walked root prefix", () => {
269-
const cache = cacheWith([["/claude/projects-copy/a.jsonl", 5000, [record()]]]);
270-
271-
const removed = pruneScanCache(cache, {
272-
livePaths: new Set(),
273-
walkedRoots: ["/claude/projects"],
274-
windowStartMs: 4000,
275-
retentionCutoffMs: 1000,
276-
});
206+
pruneScanCache(cache, retentionCutoffMs);
277207

278-
expect(removed).toBe(0);
279208
expect(cache.size).toBe(1);
280209
});
281210
});

‎apps/server/src/usage/usageScanCache.ts‎

Lines changed: 3 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -14,9 +14,6 @@
1414
*
1515
* @module usageScanCache
1616
*/
17-
// @effect-diagnostics nodeBuiltinImport:off
18-
import * as NodePath from "node:path";
19-
2017
import type { UsageProviderKind } from "@t3tools/contracts";
2118

2219
import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts";
@@ -295,47 +292,11 @@ function decodeCodexState(value: unknown): CodexScanState | null | undefined {
295292
};
296293
}
297294

298-
export interface PruneOptions {
299-
/** Files the walk just saw. Only meaningful inside the walked window. */
300-
readonly livePaths: ReadonlySet<string>;
301-
/**
302-
* Roots the walk actually completed. Absence from `livePaths` only proves a
303-
* file is gone when its root was walked: a provider whose directory failed to
304-
* resolve this pass must not have its warm entries purged.
305-
*/
306-
readonly walkedRoots: readonly string[];
307-
/** Start of the walked window; entries older than this were not looked for. */
308-
readonly windowStartMs: number;
309-
/** Entries older than this are dropped regardless. */
310-
readonly retentionCutoffMs: number;
311-
}
312-
313-
/**
314-
* Drops aged-out entries, and entries for files that have disappeared.
315-
*
316-
* The walk only covers the requested window, so absence from `livePaths` only
317-
* proves deletion for entries *inside* that window. Pruning everything the walk
318-
* missed would evict the 30-day entries every time someone looked at 7 days.
319-
*
320-
* Replaces an earlier record cap that cleared the whole cache once exceeded,
321-
* which meant a large enough window never warmed up at all.
322-
*/
323-
export function pruneScanCache(cache: ScanCache, options: PruneOptions): number {
295+
/** Keeps saved usage after transcript cleanup, until the reporting retention expires. */
296+
export function pruneScanCache(cache: ScanCache, retentionCutoffMs: number): number {
324297
let removed = 0;
325298
for (const [path, entry] of cache) {
326-
const agedOut = entry.mtimeMs < options.retentionCutoffMs;
327-
const underWalkedRoot = options.walkedRoots.some((root) => {
328-
const relative = NodePath.relative(root, path);
329-
return (
330-
relative === "" ||
331-
(relative !== ".." &&
332-
!relative.startsWith(`..${NodePath.sep}`) &&
333-
!NodePath.isAbsolute(relative))
334-
);
335-
});
336-
const deleted =
337-
underWalkedRoot && entry.mtimeMs >= options.windowStartMs && !options.livePaths.has(path);
338-
if (agedOut || deleted) {
299+
if (entry.mtimeMs < retentionCutoffMs) {
339300
cache.delete(path);
340301
removed += 1;
341302
}

0 commit comments

Comments
 (0)