diff --git a/.agents/upstream-review.md b/.agents/upstream-review.md index ac1894ec4b..ca2eab2980 100644 --- a/.agents/upstream-review.md +++ b/.agents/upstream-review.md @@ -219,3 +219,5 @@ The review cursor stays unchanged; unrelated sources in this range remain unclas | PR cache reuse and quota efficiency / `6d1d549441be84f19696ab59ed7e2fbf305280d4` | `f4600d77dd7c2fa9f10e8f4500882e427fcc7e26` (#11888), `d612d12b8bb278af97b0029f8d48b2403a5f9ee6` (#12168) | Adopted applicable behavior: scoped cache revisions, canonical detail reuse, deduplicated refreshes and bounded quota probes. Preserve Pylon environment fences and observed GraphQL consumption. Exclude account-router-specific hooks and Forgejo tests: those dependencies are absent in Pylon; their independent source decisions remain open. No cross-environment credential routing introduced. | Cycle [#625](https://github.com/pylon-code/pylon/issues/625), [PR #626](https://github.com/pylon-code/pylon/pull/626); direct adversarial self-review, focused backend/client regressions and scoped types/lint. | | Authenticated GitHub PR media / `6d1d549441be84f19696ab59ed7e2fbf305280d4` | `32e8b2584556c0c55ce0d1d8f72b506cb771d60d` (#11706) | Adopted with Pylon path/context guards retained. Signed assets use the active server GitHub CLI account per request; bounded GitHub/CDN redirects strip credentials off credentialed hosts, stream ranges, constrain MIME and sandbox SVG. Older servers retain public fallback. GitHub Enterprise media unsupported. | Cycle [#625](https://github.com/pylon-code/pylon/issues/625), [PR #629](https://github.com/pylon-code/pylon/pull/629); direct adversarial self-review, credential/redirect/range regressions, scoped checks and browser evidence. | + +| Retained usage totals and newest shared scan / `7ed3f711caa36485001538bbe1cb0017b84aa7fe` | `d4d5d12e8ba086cfbf79ca3adeb4156b46ead665` (#10315), `675869d2b0f9e059d5950b80390c0cad1cb1590e` (#12304) | Adopted with Pylon adaptation: retain saved usage after transcript cleanup or moves and across restart; claim a shared fingerprint from the newest scan with a stable ID tie-break. Configured and standalone Antigravity homes retain their profile identity, partial status and per-directory SQLite limits. Stale source identities expire when no current or retained records use them. Existing provider-level wire buckets cannot split costs by individual home when two environments each own distinct and shared homes; that cross-environment overlap remains outside this port and needs source-attributed buckets. The 90-day retention bound remains. Cursor unchanged. | Cycle [#736](https://github.com/pylon-code/pylon/issues/736), [PR #762](https://github.com/pylon-code/pylon/pull/762); 46 focused tests and scoped server checks. | diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index ec1c88ccfd..7b0dcb44df 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -8,7 +8,9 @@ import * as NodeSqlite from "node:sqlite"; import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { HostProcessEnvironment } from "@t3tools/shared/hostProcess"; +import { mergeUsage } from "@t3tools/shared/usageMerge"; import { + EnvironmentId, ProviderDriverKind, ProviderInstanceId, ServerProviderInstancesMutationId, @@ -148,14 +150,15 @@ describe("UsageService", () => { [ { type: "session_meta", payload: { id: "codex-account-session" } }, { type: "turn_context", payload: { model: "gpt-5.6-sol" } }, - { + // A-B-A at one timestamp must preserve both equal A events. + ...[11, 12, 11].map((outputTokens) => ({ type: "event_msg", timestamp: "2026-08-01T10:00:00Z", payload: { type: "token_count", - info: { last_token_usage: { input_tokens: 10, output_tokens: 11 } }, + info: { last_token_usage: { input_tokens: 10, output_tokens: outputTokens } }, }, - }, + })), ] .map((line) => encodeUnknownJsonString(line)) .join("\n") + "\n", @@ -215,7 +218,21 @@ describe("UsageService", () => { ), ); const summary = yield* service.readSummary(WINDOW); - assert.strictEqual(totalOutputTokens(summary), 36); + assert.strictEqual(totalOutputTokens(summary), 59); + yield* Effect.promise(() => + NodeFSP.rename( + NodePath.join(codexHome, "sessions", "rollout.jsonl"), + NodePath.join(codexHome, "sessions", "moved.jsonl"), + ), + ); + const moved = yield* service.readSummary(WINDOW); + assert.deepStrictEqual(moved.buckets, summary.buckets); + yield* Effect.promise(() => + NodeFSP.rm(NodePath.join(codexHome, "sessions"), { recursive: true }), + ); + const removed = yield* service.readSummary(WINDOW); + assert.deepStrictEqual(removed.buckets, summary.buckets); + const sources = summary.sources.filter((source) => source.status === "ok"); assert.strictEqual(sources.length, 4); assert.strictEqual( @@ -458,6 +475,128 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live("preserves saved tokens, costs and sessions after transcript cleanup and restart", () => + Effect.gen(function* () { + const { transcript, settings, home } = yield* setup; + const alias = NodePath.join(home, "claude-alias"); + yield* Effect.promise(() => + NodeFSP.symlink(NodePath.join(home, "claude"), alias, "junction"), + ); + const content = claudeLine(1, 5); + yield* Effect.promise(() => NodeFSP.writeFile(transcript, content)); + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + const first = yield* service.readSummary(WINDOW); + assert.strictEqual(totalOutputTokens(first), 5); + assert.isAbove(first.buckets[0]?.costUsd ?? 0, 0); + + yield* Effect.promise(() => NodeFSP.rm(transcript)); + const deleted = yield* service.readSummary(WINDOW); + assert.deepStrictEqual(deleted.buckets, first.buckets); + assert.deepStrictEqual(deleted.sources, first.sources); + + const restarted = yield* UsageService.make; + const restored = yield* restarted.readSummary(WINDOW); + assert.deepStrictEqual(restored.buckets, first.buckets); + assert.deepStrictEqual(restored.sources, first.sources); + + // A moved transcript must not count the saved usage twice. + yield* Effect.promise(() => NodeFSP.writeFile(transcript + ".jsonl", content)); + const moved = yield* restarted.readSummary(WINDOW); + assert.deepStrictEqual(moved.buckets, first.buckets); + assert.strictEqual(moved.sources[0]?.distinctSessions, 1); + + const replacementProjects = NodePath.join(home, "replacement-projects"); + yield* Effect.promise(() => NodeFSP.mkdir(replacementProjects)); + yield* Effect.promise(() => + NodeFSP.rm(NodePath.join(home, "claude", "projects"), { recursive: true }), + ); + const afterRootCleanup = yield* UsageService.make; + const missingRoot = yield* afterRootCleanup.readSummary(WINDOW); + assert.deepStrictEqual(missingRoot.buckets, first.buckets); + assert.strictEqual(missingRoot.sources[0]?.distinctSessions, 1); + assert.strictEqual(missingRoot.sources[0]?.status, "ok"); + assert.deepStrictEqual(missingRoot.sources[0]?.fingerprint, first.sources[0]?.fingerprint); + yield* Effect.promise(async () => { + const projects = NodePath.join(home, "claude", "projects"); + await NodeFSP.rename(replacementProjects, projects); + await NodeFSP.writeFile(NodePath.join(projects, "new.jsonl"), claudeLine(2, 7)); + }); + const recreated = yield* afterRootCleanup.readSummary(WINDOW); + assert.strictEqual(totalOutputTokens(recreated), 12); + assert.deepStrictEqual(recreated.sources[0]?.fingerprint, first.sources[0]?.fingerprint); + + const merged = mergeUsage( + [ + { + environmentId: EnvironmentId.make("cleanup-test"), + label: "test", + summary: recreated, + }, + { + environmentId: EnvironmentId.make("other-environment"), + label: "before cleanup", + summary: first, + }, + ], + missingRoot.contractVersion, + ); + assert.strictEqual(merged.outputTokens, 12); + assert.strictEqual(merged.sessions, 1); + assert.strictEqual(merged.costUsd, recreated.buckets[0]?.costUsd); + + const outsideWindow = yield* restarted.readSummary({ + ...WINDOW, + sinceDay: UsageDay.make("2026-08-02"), + }); + assert.deepStrictEqual(outsideWindow.buckets, []); + assert.strictEqual(outsideWindow.sources[0]?.distinctSessions, 0); + }).pipe( + Effect.provide( + serviceLayers({ + prefix: "usage-service-cleanup-test", + home, + settings: { providers: { ...settings.providers, claudeAgent: { homePath: alias } } }, + ratesDocument: { + "claude-fable-5": { input_cost_per_token: 1e-5, output_cost_per_token: 5e-5 }, + }, + }), + ), + ); + }).pipe(Effect.scoped), + ); + + it.live("drops stale source identities after an empty account home is replaced", () => + Effect.gen(function* () { + const { home, settings } = yield* setup; + const replacement = NodePath.join(home, "replacement-claude"); + yield* Effect.promise(() => + NodeFSP.mkdir(NodePath.join(replacement, "projects"), { recursive: true }), + ); + yield* Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const settingsService = yield* ServerSettings.ServerSettingsService; + const service = yield* UsageService.make; + yield* service.readSummary(WINDOW); + const cachePath = NodePath.join(config.stateDir, "usage-scan-cache.json"); + const oldDir = NodePath.join(home, "claude", "projects"); + assert.include(yield* Effect.promise(() => NodeFSP.readFile(cachePath, "utf8")), oldDir); + + yield* settingsService.updateSettings({ + providers: { claudeAgent: { homePath: replacement } }, + }); + yield* service.readSummary(WINDOW); + const persisted = yield* Effect.promise(() => NodeFSP.readFile(cachePath, "utf8")); + assert.notInclude(persisted, oldDir); + assert.include(persisted, NodePath.join(replacement, "projects")); + }).pipe( + Effect.provide( + serviceLayers({ prefix: "usage-service-source-prune-test", home, settings }), + ), + ); + }).pipe(Effect.scoped), + ); + it.live("does not share an in-flight scan after custom prices change", () => Effect.gen(function* () { const { transcript, settings, home } = yield* setup; @@ -769,6 +908,120 @@ describe("UsageService", () => { ), ); + it.live("retains Antigravity usage after conversation cleanup and service restart", () => + Effect.gen(function* () { + yield* setup; + const config = yield* ServerConfig.ServerConfig; + const instanceId = ProviderInstanceId.make("antigravity-retained"); + const profileDir = resolveAntigravityProfileDirectory(config.stateDir, instanceId); + const convDir = NodePath.join(profileDir, "antigravity-acp", "conversations"); + yield* Effect.promise(() => NodeFSP.mkdir(convDir, { recursive: true })); + const dbPath = NodePath.join(convDir, "retained.db"); + seedAntigravityDb(dbPath, [ + { + idx: 0, + blob: encodeSyntheticGenMetadataBlob({ + modelName: "gemini-2.5-pro", + timestampSeconds: 1785578400n, + inputTokens: 100, + outputTokens: 40, + }), + }, + ]); + + const service = yield* UsageService.make; + const first = yield* service.readSummary(WINDOW); + yield* Effect.promise(() => NodeFSP.rm(convDir, { recursive: true })); + const afterCleanup = yield* service.readSummary(WINDOW); + const restarted = yield* UsageService.make; + const afterRestart = yield* restarted.readSummary(WINDOW); + + for (const [phase, summary] of [ + ["cleanup", afterCleanup], + ["restart", afterRestart], + ] as const) { + assert.deepStrictEqual(summary.buckets, first.buckets, phase); + const source = summary.sources.find( + (item) => + item.fingerprint.provider === "antigravity" && + item.fingerprint.resolvedHomePath === convDir, + ); + assert.isDefined(source); + assert.strictEqual(source.status, "ok"); + assert.strictEqual(source.scannedFiles, 1); + } + }).pipe( + Effect.provide( + serviceLayers({ + prefix: "usage-service-antigravity-retained-test", + home: "", + settings: { + providers: {}, + providerInstances: { + [ProviderInstanceId.make("antigravity-retained")]: { + driver: "antigravity", + enabled: true, + }, + }, + }, + }), + ), + Effect.scoped, + ), + ); + + it.live( + "retains standalone Antigravity history after its directory is removed and service restarts", + () => + Effect.gen(function* () { + yield* setup; + const config = yield* ServerConfig.ServerConfig; + const convDir = NodePath.join(config.stateDir, "standalone-antigravity", "conversations"); + yield* Effect.promise(() => NodeFSP.mkdir(convDir, { recursive: true })); + seedAntigravityDb(NodePath.join(convDir, "session.db"), [ + { + idx: 0, + blob: encodeSyntheticGenMetadataBlob({ + modelName: "gemini-2.5-pro", + timestampSeconds: 1785578400n, + inputTokens: 100, + outputTokens: 40, + }), + }, + ]); + + yield* Effect.gen(function* () { + const firstService = yield* UsageService.make; + const first = yield* firstService.readSummary(WINDOW); + assert.isTrue(first.buckets.some((bucket) => bucket.provider === "antigravity")); + yield* Effect.promise(() => NodeFSP.rm(convDir, { recursive: true })); + + for (const summary of [ + yield* firstService.readSummary(WINDOW), + yield* (yield* UsageService.make).readSummary(WINDOW), + ]) { + assert.deepStrictEqual(summary.buckets, first.buckets); + const source = summary.sources.find( + (item) => + item.fingerprint.provider === "antigravity" && + item.fingerprint.resolvedHomePath === convDir, + ); + assert.isDefined(source); + assert.strictEqual(source.status, "ok"); + } + }).pipe(Effect.provideService(UsageService.StandaloneAntigravityConversations, convDir)); + }).pipe( + Effect.provide( + serviceLayers({ + prefix: "usage-service-standalone-antigravity-retained-test", + home: "", + settings: { providers: {}, providerInstances: {} }, + }), + ), + Effect.scoped, + ), + ); + it.live( "reports honest missing status when configured Antigravity instance has no conversations dir on disk", () => diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 122b89436e..8d63a91b9d 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -14,6 +14,8 @@ * @module UsageService */ import * as NodeOS from "node:os"; +// @effect-diagnostics nodeBuiltinImport:off - Context.Reference defaults cannot request Path. +import * as NodePath from "node:path"; import { ClaudeSettings, @@ -90,6 +92,18 @@ const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000; /** Longest window the UI offers, plus slack. Older entries are pruned. */ const CACHE_RETENTION_DAYS = 90; +const MAX_ANTIGRAVITY_FILES_PER_DIR = 500; +const MAX_ANTIGRAVITY_BYTES_PER_DIR = 256 * 1024 * 1024; +const MAX_ANTIGRAVITY_RECORDS_PER_DIR = 50_000; + +/** The standalone Antigravity home belongs to this server environment. */ +export const StandaloneAntigravityConversations = Context.Reference( + "StandaloneAntigravityConversations", + { + defaultValue: () => + NodePath.join(NodeOS.homedir(), ".gemini", "antigravity-cli", "conversations"), + }, +); const decodeCodexSettings = Schema.decodeOption(CodexSettings); const decodeClaudeSettings = Schema.decodeOption(ClaudeSettings); @@ -110,6 +124,11 @@ const encodeRatesCache = Schema.encodeEffect( const ScanCacheJson = Schema.fromJsonString(Schema.Unknown as unknown as Schema.Codec); const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson); const encodeScanCacheFile = Schema.encodeEffect(ScanCacheJson); +const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson); +const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String }); +const decodeCachedSources = Schema.decodeUnknownOption( + Schema.Struct({ sources: Schema.Record(Schema.String, CachedSource) }), +); export class UsageService extends Context.Service< UsageService, @@ -156,7 +175,12 @@ export const make = Effect.gen(function* () { const hostEnvironment = yield* HostProcessEnvironment; const fileCache: ScanCache = new Map(); + const sourceCache = new Map(); let cacheDirty = false; + const isWithinDirectory = (filePath: string, dir: string) => { + const relative = path.relative(dir, filePath); + return relative !== ".." && !relative.startsWith(".." + path.sep) && !path.isAbsolute(relative); + }; const ratesCachePath = path.join(config.stateDir, "usage-model-rates.json"); const scanCachePath = path.join(config.stateDir, "usage-scan-cache.json"); @@ -250,9 +274,16 @@ export const make = Effect.gen(function* () { /** Resolves the transcript directory for each provider. */ const resolveTranscriptDirs = Effect.fn("UsageService.resolveTranscriptDirs")(function* ( settings: ServerSettingsValue, + retentionCutoffMs: number, ) { - const dirs: Array<{ provider: UsageProviderKind; dir: string; fileName?: string }> = []; + const dirs: Array<{ + provider: UsageProviderKind; + dir: string; + volumeId: string; + fileName?: string; + }> = []; const seen = new Set(); + const activeSourceKeys = new Set(); for (const driver of ["claudeAgent", "codex", "grok"] as const) { // Disabled accounts still have history. Explicit default slots replace // the legacy settings, just as they do in the provider registry. @@ -289,14 +320,42 @@ export const make = Effect.gen(function* () { ); } const directory = path.resolve(home, provider === "claude" ? "projects" : "sessions"); - // Account aliases and Codex auth overlays can share the same history. + const sourceKey = provider + "\0" + directory; + const previous = sourceCache.get(sourceKey); + // Keep canonical paths and source fingerprints stable after root cleanup, + // including aliases and clients merging pre-cleanup environment summaries. const dir = yield* fileSystem .realPath(directory) - .pipe(Effect.orElseSucceed(() => directory)); + .pipe(Effect.orElseSucceed(() => previous?.dir ?? directory)); + const currentVolumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir)); + const hasRetainedHistory = fileCache + .entries() + .some( + ([filePath, entry]) => + entry.provider === provider && + entry.mtimeMs >= retentionCutoffMs && + entry.records.length + entry.tailRecords.length > 0 && + isWithinDirectory(filePath, dir), + ); + // A recreated directory still reports the retained history under its old identity. + const volumeId = + previous?.dir === dir && (hasRetainedHistory || !currentVolumeId) + ? previous.volumeId || currentVolumeId + : currentVolumeId; + if (previous?.dir !== dir || previous.volumeId !== volumeId) { + sourceCache.set(sourceKey, { dir, volumeId }); + cacheDirty = true; + } const key = `${provider}\0${dir}`; + activeSourceKeys.add(sourceKey); if (seen.has(key)) continue; seen.add(key); - dirs.push({ provider, dir, ...(provider === "grok" ? { fileName: "updates.jsonl" } : {}) }); + dirs.push({ + provider, + dir, + volumeId, + ...(provider === "grok" ? { fileName: "updates.jsonl" } : {}), + }); } } // Antigravity profile roots: resolve configured instances @@ -324,21 +383,69 @@ export const make = Effect.gen(function* () { } // Standalone ~/.gemini/antigravity-cli/conversations when present - const standaloneDir = path.join( - NodeOS.homedir(), - ".gemini", - "antigravity-cli", - "conversations", - ); + const standaloneDir = yield* StandaloneAntigravityConversations; const standaloneExists = yield* fileSystem .exists(standaloneDir) .pipe(Effect.catchCause(() => Effect.succeed(false))); - if (standaloneExists) { + const hasStandaloneHistory = fileCache + .entries() + .some( + ([filePath, entry]) => + entry.provider === "antigravity" && + entry.mtimeMs >= retentionCutoffMs && + entry.records.length + entry.tailRecords.length > 0 && + isWithinDirectory(filePath, standaloneDir), + ); + if (standaloneExists || hasStandaloneHistory) { antigravityRoots.add(standaloneDir); } for (const root of antigravityRoots) { - dirs.push({ provider: "antigravity" as const, dir: root }); + const directory = path.resolve(root); + const sourceKey = "antigravity\0" + directory; + const previous = sourceCache.get(sourceKey); + // Pylon's Antigravity profile fingerprint is the configured profile + // path, not its realpath. Keep it stable across profile cleanup. + const dir = directory; + const currentVolumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir)); + const hasRetainedHistory = fileCache + .entries() + .some( + ([filePath, entry]) => + entry.provider === "antigravity" && + entry.mtimeMs >= retentionCutoffMs && + entry.records.length + entry.tailRecords.length > 0 && + isWithinDirectory(filePath, dir), + ); + const volumeId = + previous?.dir === dir && (hasRetainedHistory || !currentVolumeId) + ? previous.volumeId || currentVolumeId + : currentVolumeId; + if (previous?.dir !== dir || previous.volumeId !== volumeId) { + sourceCache.set(sourceKey, { dir, volumeId }); + cacheDirty = true; + } + activeSourceKeys.add(sourceKey); + dirs.push({ provider: "antigravity", dir, volumeId }); + } + + // Old configured homes can disappear from settings. Keep their identity + // only while an in-retention cached transcript could use it again. + for (const [sourceKey, source] of sourceCache) { + if (activeSourceKeys.has(sourceKey)) continue; + const provider = sourceKey.split("\0", 1)[0]; + const hasRetainedHistory = fileCache + .entries() + .some( + ([filePath, entry]) => + entry.provider === provider && + entry.mtimeMs >= retentionCutoffMs && + entry.records.length + entry.tailRecords.length > 0 && + isWithinDirectory(filePath, source.dir), + ); + if (hasRetainedHistory) continue; + sourceCache.delete(sourceKey); + cacheDirty = true; } return dirs; @@ -359,6 +466,11 @@ export const make = Effect.gen(function* () { ); if (document === null) return; for (const [path, entry] of decodeScanCache(document)) fileCache.set(path, entry); + const sources = decodeCachedSources(document); + if (Option.isSome(sources)) { + for (const [key, source] of Object.entries(sources.value.sources)) + sourceCache.set(key, source); + } }), ); @@ -366,7 +478,10 @@ export const make = Effect.gen(function* () { if (!cacheDirty) return; // Cleared only after the write lands, so a failed persist is retried on // the next scan instead of leaving disk permanently stale. - yield* encodeScanCacheFile(encodeScanCache(fileCache)).pipe( + yield* encodeScanCacheFile({ + ...encodeScanCache(fileCache), + sources: Object.fromEntries(sourceCache), + }).pipe( Effect.flatMap((serialized) => fileSystem.writeFileString(scanCachePath, serialized)), Effect.map(() => { cacheDirty = false; @@ -417,7 +532,8 @@ export const make = Effect.gen(function* () { ); // A read failure is not an empty transcript: caching it under this // (size, mtime) would silently drop the file's usage until it changes. - if (parsed === null) return []; + if (parsed === null) + return cached?.provider === provider ? [...cached.records, ...cached.tailRecords] : []; // Stored already de-duplicated within the file, which is 99% of all // duplicates. The aggregator still runs the cross-file dedupe pass. One @@ -452,24 +568,25 @@ export const make = Effect.gen(function* () { readonly status?: "ok" | "missing" | "partial" | "failed"; readonly malformedRecords?: number; readonly message?: string | null; + readonly nativeBudget?: { + readonly files: number; + readonly bytes: number; + readonly records: number; + }; } const collectDirs = Effect.fn("UsageService.collectDirs")(function* ( windowStartMs: number, settings: ServerSettingsValue, + retentionCutoffMs: number, ) { // The home resolvers ask for `Path` themselves; satisfy them from the // instance we already hold so the scan stays context-free. - const dirs = yield* resolveTranscriptDirs(settings).pipe( + const dirs = yield* resolveTranscriptDirs(settings, retentionCutoffMs).pipe( Effect.provideService(Path.Path, path), ); const scanned: ScannedDir[] = []; - // Bound native SQLite work across all instance roots, including warm-cache records. - let nativeFiles = 0; - let nativeBytes = 0; - let nativeRecords = 0; - for (const { provider, dir, fileName } of dirs) { - const volumeId = yield* Effect.promise(() => readDirectoryVolumeId(dir)); + for (const { provider, dir, volumeId, fileName } of dirs) { const exists = yield* fileSystem .exists(dir) .pipe(Effect.catchCause(() => Effect.succeed(false))); @@ -479,6 +596,10 @@ export const make = Effect.gen(function* () { } if (provider === "antigravity") { + // Native SQLite limits apply to each profile directory independently. + let nativeFiles = 0; + let nativeBytes = 0; + let nativeRecords = 0; const dirEntries = yield* fileSystem .readDirectory(dir) .pipe(Effect.catchCause(() => Effect.succeed(null))); @@ -497,10 +618,6 @@ export const make = Effect.gen(function* () { const dbFileNames = dirEntries.filter((name) => name.endsWith(".db")).sort(); - const MAX_ANTIGRAVITY_FILES_PER_DIR = 500; - const MAX_ANTIGRAVITY_BYTES_PER_DIR = 256 * 1024 * 1024; - const MAX_ANTIGRAVITY_RECORDS_PER_DIR = 50_000; - const parsedFiles: { path: string; records: readonly UsageRecord[] }[] = []; let dirMalformedRows = 0; let dirTruncated = false; @@ -683,6 +800,7 @@ export const make = Effect.gen(function* () { status, malformedRecords: dirMalformedRows, message, + nativeBudget: { files: nativeFiles, bytes: nativeBytes, records: nativeRecords }, }); continue; } @@ -749,11 +867,13 @@ export const make = Effect.gen(function* () { const windowStartMs = (hourlyWindow?.sinceTimeMs ?? DateTime.toEpochMillis(windowStart.value)) - MTIME_SLACK_MS; + const retentionCutoffMs = startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000; + // Pricing only matters once records are aggregated, so the rate table // loads while transcripts stream instead of gating them: a cold rates // fetch on a slow network no longer delays the scan by its own timeout. const [, scannedDirs] = yield* Effect.all( - [ensureRates(false), collectDirs(windowStartMs, settings)], + [ensureRates(false), collectDirs(windowStartMs, settings, retentionCutoffMs)], { concurrency: 2 }, ); @@ -768,8 +888,6 @@ export const make = Effect.gen(function* () { }); const sources: UsageSource[] = []; - const livePaths = new Set(); - const walkedRoots: string[] = []; for (const { provider, @@ -779,38 +897,72 @@ export const make = Effect.gen(function* () { status, malformedRecords, message, + nativeBudget, } of scannedDirs) { - if (files === null) { - sources.push({ - fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId }, - status: "missing", - scannedFiles: 0, - skippedFiles: 0, - malformedRecords: 0, - distinctSessions: 0, - message: "No transcript directory on this environment.", - }); - continue; + const retainedFiles = [...(files ?? [])]; + let retainedTruncated = false; + let retainedNativeFiles = nativeBudget?.files ?? 0; + let retainedNativeBytes = nativeBudget?.bytes ?? 0; + let retainedNativeRecords = nativeBudget?.records ?? 0; + const livePaths = new Set(retainedFiles.map((file) => file.path)); + // Cleanup may remove transcripts, but the usage we already saved still + // contributes to this source. Keep the normal aggregation and dedupe path. + for (const [filePath, entry] of fileCache) { + if ( + entry.provider !== provider || + entry.mtimeMs < retentionCutoffMs || + livePaths.has(filePath) || + !isWithinDirectory(filePath, dir) + ) + continue; + if (provider === "antigravity") { + const recordCount = entry.records.length + entry.tailRecords.length; + if ( + retainedNativeFiles + 1 > MAX_ANTIGRAVITY_FILES_PER_DIR || + retainedNativeBytes + entry.size > MAX_ANTIGRAVITY_BYTES_PER_DIR || + retainedNativeRecords + recordCount > MAX_ANTIGRAVITY_RECORDS_PER_DIR + ) { + retainedTruncated = true; + continue; + } + retainedNativeFiles++; + retainedNativeBytes += entry.size; + retainedNativeRecords += recordCount; + } + retainedFiles.push({ path: filePath, records: [...entry.records, ...entry.tailRecords] }); } - - walkedRoots.push(dir); let scannedFiles = 0; let skippedFiles = 0; // 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(); - for (const file of files) { - livePaths.add(file.path); + for (const file of retainedFiles) { if (file.records.length === 0) { skippedFiles += 1; continue; } scannedFiles += 1; + const codexEventOccurrences = new Map(); for (const record of file.records) { - // Only sessions that contributed in-window count: the mtime slack - // admits boundary files whose records fall outside the range. - if (aggregator.add(record) && record.sessionId.length > 0) { + let usageRecord = record; + if (record.provider === "codex" && record.sessionId.length > 0) { + // Match moved rollout copies without collapsing repeated equal events + // within one rollout (timestamps can have only second precision). + const key = encodeUsageRecordKey([ + record.provider, + record.sessionId, + record.timestampMs, + record.model, + record.totals, + ]); + const occurrence = (codexEventOccurrences.get(key) ?? 0) + 1; + codexEventOccurrences.set(key, occurrence); + usageRecord = { ...record, dedupeKey: key + ":" + occurrence }; + } + // Only sessions contributing in-window count; the mtime slack can + // admit boundary files whose records fall outside the range. + if (aggregator.add(usageRecord) && record.sessionId.length > 0) { sessionIds.add(record.sessionId); } } @@ -818,21 +970,27 @@ export const make = Effect.gen(function* () { sources.push({ fingerprint: { hostId, provider, resolvedHomePath: dir, volumeId }, - status: status ?? "ok", + // Saved records remain available after cleanup. Preserve Pylon's native + // partial/failed scan status when an Antigravity read was incomplete. + status: retainedTruncated + ? "partial" + : files === null + ? scannedFiles === 0 + ? "missing" + : "ok" + : status === "failed" && scannedFiles > 0 + ? "partial" + : (status ?? "ok"), scannedFiles, skippedFiles, malformedRecords: malformedRecords ?? 0, distinctSessions: sessionIds.size, - message: message ?? null, + message: + files === null ? "No transcript directory on this environment." : (message ?? null), }); } - const pruned = pruneScanCache(fileCache, { - livePaths, - walkedRoots, - windowStartMs, - retentionCutoffMs: startedAtMs - CACHE_RETENTION_DAYS * 24 * 60 * 60 * 1000, - }); + const pruned = pruneScanCache(fileCache, retentionCutoffMs); if (pruned > 0) cacheDirty = true; yield* persistScanCache(); diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index fdb0aabafa..cc1bdbcc16 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -194,88 +194,17 @@ describe("pruneScanCache", () => { it("drops entries older than retention", () => { const cache = cacheWith([["/old.jsonl", 500, [record()]]]); - const removed = pruneScanCache(cache, { - livePaths: new Set(), - walkedRoots: ["/"], - windowStartMs: 400, - retentionCutoffMs, - }); + const removed = pruneScanCache(cache, retentionCutoffMs); expect(removed).toBe(1); expect(cache.size).toBe(0); }); - it("drops in-window entries whose file has disappeared", () => { + it("keeps entries whose file has disappeared", () => { const cache = cacheWith([["/gone.jsonl", 5000, [record()]]]); - pruneScanCache(cache, { - livePaths: new Set(), - walkedRoots: ["/"], - windowStartMs: 4000, - retentionCutoffMs, - }); - - expect(cache.size).toBe(0); - }); - - it("keeps entries outside the walked window that are still within retention", () => { - // Viewing 7 days must not evict the 30-day entries, which that walk never - // looked for and so cannot prove are gone. - const cache = cacheWith([["/older-but-valid.jsonl", 2000, [record()]]]); - - const removed = pruneScanCache(cache, { - livePaths: new Set(), - walkedRoots: ["/"], - windowStartMs: 4000, - retentionCutoffMs, - }); - - expect(removed).toBe(0); - expect(cache.size).toBe(1); - }); - - it("keeps entries the walk saw", () => { - const cache = cacheWith([["/live.jsonl", 5000, [record()]]]); - - pruneScanCache(cache, { - livePaths: new Set(["/live.jsonl"]), - walkedRoots: ["/"], - windowStartMs: 4000, - retentionCutoffMs, - }); - - expect(cache.size).toBe(1); - }); -}); - -describe("pruneScanCache with an unwalked root", () => { - it("keeps in-window entries for a provider whose directory was not walked", () => { - // A missing provider root or failed settings read leaves livePaths without - // that provider's files. Its warm entries must survive the pass. - const cache = cacheWith([["/codex/sessions/a.jsonl", 5000, [record()]]]); - - const removed = pruneScanCache(cache, { - livePaths: new Set(), - walkedRoots: ["/claude/projects"], - windowStartMs: 4000, - retentionCutoffMs: 1000, - }); - - expect(removed).toBe(0); - expect(cache.size).toBe(1); - }); - - it("keeps entries under a sibling path that only shares the walked root prefix", () => { - const cache = cacheWith([["/claude/projects-copy/a.jsonl", 5000, [record()]]]); - - const removed = pruneScanCache(cache, { - livePaths: new Set(), - walkedRoots: ["/claude/projects"], - windowStartMs: 4000, - retentionCutoffMs: 1000, - }); + pruneScanCache(cache, retentionCutoffMs); - expect(removed).toBe(0); expect(cache.size).toBe(1); }); }); diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 04f9597f42..8439d5ca74 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -14,9 +14,6 @@ * * @module usageScanCache */ -// @effect-diagnostics nodeBuiltinImport:off -import * as NodePath from "node:path"; - import type { UsageProviderKind } from "@t3tools/contracts"; import { GUARD_LENGTH, type TranscriptParsePosition } from "./usageTranscriptReader.ts"; @@ -302,47 +299,11 @@ function decodeCodexState(value: unknown): CodexScanState | null | undefined { }; } -export interface PruneOptions { - /** Files the walk just saw. Only meaningful inside the walked window. */ - readonly livePaths: ReadonlySet; - /** - * Roots the walk actually completed. Absence from `livePaths` only proves a - * file is gone when its root was walked: a provider whose directory failed to - * resolve this pass must not have its warm entries purged. - */ - readonly walkedRoots: readonly string[]; - /** Start of the walked window; entries older than this were not looked for. */ - readonly windowStartMs: number; - /** Entries older than this are dropped regardless. */ - readonly retentionCutoffMs: number; -} - -/** - * Drops aged-out entries, and entries for files that have disappeared. - * - * The walk only covers the requested window, so absence from `livePaths` only - * proves deletion for entries *inside* that window. Pruning everything the walk - * missed would evict the 30-day entries every time someone looked at 7 days. - * - * Replaces an earlier record cap that cleared the whole cache once exceeded, - * which meant a large enough window never warmed up at all. - */ -export function pruneScanCache(cache: ScanCache, options: PruneOptions): number { +/** Keeps saved usage after transcript cleanup, until the reporting retention expires. */ +export function pruneScanCache(cache: ScanCache, retentionCutoffMs: number): number { let removed = 0; for (const [path, entry] of cache) { - const agedOut = entry.mtimeMs < options.retentionCutoffMs; - const underWalkedRoot = options.walkedRoots.some((root) => { - const relative = NodePath.relative(root, path); - return ( - relative === "" || - (relative !== ".." && - !relative.startsWith(`..${NodePath.sep}`) && - !NodePath.isAbsolute(relative)) - ); - }); - const deleted = - underWalkedRoot && entry.mtimeMs >= options.windowStartMs && !options.livePaths.has(path); - if (agedOut || deleted) { + if (entry.mtimeMs < retentionCutoffMs) { cache.delete(path); removed += 1; } diff --git a/packages/shared/src/usageMerge.test.ts b/packages/shared/src/usageMerge.test.ts index 3da3865022..9c1346b3ce 100644 --- a/packages/shared/src/usageMerge.test.ts +++ b/packages/shared/src/usageMerge.test.ts @@ -147,6 +147,52 @@ describe("mergeUsage", () => { ).toEqual({ claude: 1, codex: 1 }); }); + it("uses the newest scan when environments share the same transcript directory", () => { + const source = { provider: "claude" as const, hostId: "mac", homePath: "/home/theo/.claude" }; + const environments = [ + environment("env-a", summary([bucket({ costUsd: 4, records: 2 })], [source])), + environment("env-b", { + ...summary([bucket()], [source]), + readAt: "2026-08-07T01:00:00.000Z", + }), + ]; + + for (const ordered of [environments, environments.toReversed()]) { + const merged = mergeUsage(ordered, USAGE_CONTRACT_VERSION); + expect(merged.costUsd).toBe(10); + expect(merged.records).toBe(5); + expect(merged.sessions).toBe(1); + expect(merged.contributingEnvironments).toEqual(["env-b"]); + expect(merged.duplicateSources).toEqual(["env-a: /home/theo/.claude"]); + } + }); + + it("uses stable environment ids for equal or invalid scan timestamps", () => { + const source = { provider: "claude" as const, hostId: "mac", homePath: "/shared/.claude" }; + const invalid = environment("env-a", { + ...summary([bucket({ costUsd: 4 })], [source]), + readAt: "invalid", + }); + const equallyNew = [ + environment("env-b", summary([bucket({ costUsd: 10 })], [source])), + environment("env-c", summary([bucket({ costUsd: 20 })], [source])), + ]; + for (const ordered of [[invalid, ...equallyNew], [...equallyNew, invalid].toReversed()]) { + const merged = mergeUsage(ordered, USAGE_CONTRACT_VERSION); + expect(merged.costUsd).toBe(10); + expect(merged.contributingEnvironments).toEqual(["env-b"]); + } + const bothInvalid = mergeUsage( + [ + environment("env-z", { ...summary([bucket({ costUsd: 9 })], [source]), readAt: "bad" }), + invalid, + ], + USAGE_CONTRACT_VERSION, + ); + expect(bothInvalid.costUsd).toBe(4); + expect(bothInvalid.contributingEnvironments).toEqual(["env-a"]); + }); + it("excludes an environment reporting an older contract version", () => { const merged = mergeUsage( [ diff --git a/packages/shared/src/usageMerge.ts b/packages/shared/src/usageMerge.ts index e0cb0510eb..fee8e9c956 100644 --- a/packages/shared/src/usageMerge.ts +++ b/packages/shared/src/usageMerge.ts @@ -118,9 +118,9 @@ function fingerprintKey(fingerprint: UsageSourceFingerprint): string { * * Several environments on one machine (worktree servers, for instance) resolve * the same provider home and would otherwise double count every token. The - * first environment in a stable order claims a fingerprint; the rest have that - * provider's buckets dropped. Environments are sorted by id so the winner does - * not change between renders. + * most recently read summary claims a fingerprint; the rest have that provider's + * buckets dropped. Environment ids break ties so the winner is stable when + * summaries have the same read time. */ function claimSources(environments: readonly EnvironmentUsage[]): { readonly ownerByFingerprint: ReadonlyMap; @@ -129,7 +129,11 @@ function claimSources(environments: readonly EnvironmentUsage[]): { const ownerByFingerprint = new Map(); const duplicates: string[] = []; - const ordered = [...environments].sort((a, b) => a.environmentId.localeCompare(b.environmentId)); + const ordered = [...environments].sort( + (a, b) => + (Date.parse(b.summary.readAt) || 0) - (Date.parse(a.summary.readAt) || 0) || + a.environmentId.localeCompare(b.environmentId), + ); for (const environment of ordered) { for (const source of environment.summary.sources) {