fix(usage): aggregate complete ledger incrementally

docs-site/src/content/docs/reference/management-api.md

@@ -124,7 +124,7 @@ See [Combos](/guides/combos/) for target strategies, cooldowns, aliases, and rou
 | `GET /api/debug/usage-logs` | Read bounded usage-debug entries | — |
 | `GET /api/debug/injection-logs` | Read bounded guidance-injection debug entries | — |
 | `GET /api/claude/inbound-debug` | Read Claude inbound debug state and entries | — |
-| `GET /api/usage` | Summarize usage by range and client surface; Codex responses also include an `accounts` breakdown keyed by stable non-PII log labels | Returns an `error: "read_failed"` summary if storage cannot be read |
+| `GET /api/usage` | Stream the complete usage ledger into compact aggregates, then incrementally fold verified appends; summarize by range and client surface, with a Codex `accounts` breakdown keyed by stable non-PII log labels | Returns an `error: "read_failed"` summary if storage cannot be read |
 | `GET /api/storage` | Scan Codex storage usage by bucket | Returns an `error: "scan_failed"` payload on scan failure |
 | `POST /api/storage/cleanup/preview` | Preview archived-session cleanup and return a binding digest | 400 `invalid_json` or `invalid_percent` |
 | `POST /api/storage/cleanup` | Quarantine or permanently remove the previewed archived set | 400 invalid input; 409 stale/busy/referenced state; 500 filesystem/database failure |
@@ -135,6 +135,25 @@ See [Combos](/guides/combos/) for target strategies, cooldowns, aliases, and rou
 | `POST /api/storage/cleanup-policy/run` | Start a manual cleanup-policy run | 409 `already_running`; 500 `cleanup_failed` |
 | `GET /api/storage/cleanup-policy/test-stream` | Test-only policy stream hook | 404 `not_found` when unavailable |
 
+`GET /api/usage` reads `~/.opencodex/usage.jsonl` from the beginning through the current ledger
+snapshot on a cold start. It processes fixed 1 MiB chunks and retains compact aggregate state rather
+than every normalized request row. Later refreshes validate the previous line boundary and fold only
+newly appended complete rows. Concurrent callers share the same refresh. Range and surface predicates
+are applied to the complete aggregate, so the former read-byte window and parsed-row cap cannot omit
+an earlier file prefix from 7-day, 30-day, or all-history totals. `managementUsageMaxReadBytes` remains
+accepted for compatibility with bounded legacy readers, but changing it no longer expands or reduces
+the history summarized by this endpoint.
+
+The runtime ledger is append-only. Replacing or truncating it, or changing local pricing/time-zone
+inputs, triggers a complete rebuild. If you manually edit an older row in place while the proxy is
+running, restart the proxy (or replace the file) before relying on the new total; incremental refreshes
+verify the append boundary, not every previously aggregated byte.
+
+The response still includes `historyTruncated`, `truncatedPrefixBytes`, `entriesTruncated`, and
+`entriesDropped` so older clients can consume the same wire shape. A successful whole-ledger scan
+reports `false`, `0`, `false`, and `0`, respectively. These are legacy compatibility fields, not a
+signal that the endpoint read only a configured-size tail.
+
 For `GET /api/usage?range=30d&surface=codex`, `accounts` contains one row per observed Codex
 pool label. Each row reports `accountLogLabel`, token totals, `usageCoverageRatio`, and an optional
 `estimatedCostUsd` based on the currently configured display pricing. Active user `modelCosts`

gui/src/pages/use-dashboard-data.ts

@@ -270,7 +270,7 @@ export function useDashboardData(apiBase: string) {
     (signal) => fetchDashboardUsage(apiBase, signal),
     // 30d usage is documented ~5s cold; this shared key has four subscribers, so
     // every one of them carries the same raised deadline (mount-order independent).
-    { enabled: overviewReady, deadlineMs: 60_000 },
+    { enabled: overviewReady, pollMs: 60_000, deadlineMs: 60_000 },
   );
 
   const diagnosticsPoll = useKeyedClientResource(

gui/tests/dashboard-contracts.test.ts

@@ -53,7 +53,7 @@ test("Dashboard usage polling cannot delay core health and settings", async () =
   expect(hook).toContain("fetchDashboardUsage(apiBase, signal)");
   expect(hook).toContain("fetchDashboardSidecars");
   expect(hook).toContain("fetchDashboardOverview");
-  expect(hook).not.toMatch(/usageSummary30dResourceKey\(apiBase\)[\s\S]*pollMs: 60_000/);
+  expect(hook).toMatch(/usageSummary30dResourceKey\(apiBase\)[\s\S]*pollMs: 60_000/);
 });
 
 test("Dashboard interactive controls load independently of health/providers", async () => {

src/config.ts

@@ -1010,7 +1010,9 @@ const configSchema = z.object({
   // A malformed present client block must remain diagnosable from raw config and
   // fail closed through src/client/state.ts; unrelated provider state still loads.
   client: clientConnectionSchema.optional().catch(undefined),
-  managementUsageMaxReadBytes: z.number().int().positive().default(64 * 1024 * 1024),
+  managementUsageMaxReadBytes: z.number().int().positive().default(64 * 1024 * 1024).describe(
+    "Deprecated compatibility limit for bounded legacy usage readers; GET /api/usage always aggregates the complete ledger",
+  ),
   // Invalid hand edits disable only this opt-in circuit. Live writes remain strict.
   upstreamHostCircuitThreshold: z.number().int()
     .min(0)

src/lib/app-owned-memory-stores.ts

@@ -38,6 +38,10 @@ import {
   discardRetainedUsageSnapshot,
   retainedUsageSnapshotStats,
 } from "../usage/log";
+import {
+  discardRetainedUsageAggregate,
+  usageAggregateRetainedStats,
+} from "../server/management/usage-aggregate-cache";
 import {
   cursorBlobRetainedStoreSnapshot,
   evictOldestCursorBlobForBudget,
@@ -61,18 +65,33 @@ function ringSnapshot(metrics: { entries: number; bytes: number; oldestAt: numbe
   };
 }
 
-/** The retained usage tail is a single all-or-nothing entry: evicting it drops the whole tail. */
+/** Legacy parsed tail and streaming aggregate share one stable public store id. */
 function usageSnapshotRetainedStoreSnapshot(): RetainedStoreSnapshot {
-  const stats = retainedUsageSnapshotStats();
+  const legacy = retainedUsageSnapshotStats();
+  const aggregate = usageAggregateRetainedStats();
+  const oldest = [legacy.oldestAt, aggregate.oldestAt]
+    .filter((value): value is number => value !== null)
+    .sort((a, b) => a - b)[0] ?? null;
   return {
-    count: stats.count,
-    bytes: stats.bytes,
-    evictableBytes: stats.bytes,
-    pinnedBytes: 0,
-    oldestAt: stats.oldestAt,
+    count: legacy.count + aggregate.count,
+    bytes: legacy.bytes + aggregate.bytes,
+    evictableBytes: legacy.bytes + aggregate.evictableBytes,
+    pinnedBytes: aggregate.pinnedBytes,
+    oldestAt: oldest,
   };
 }
 
+function evictOldestUsageSnapshot(): number {
+  const legacy = retainedUsageSnapshotStats();
+  const aggregate = usageAggregateRetainedStats();
+  if (legacy.bytes > 0
+    && (aggregate.evictableBytes === 0
+      || (legacy.oldestAt ?? Number.POSITIVE_INFINITY) <= (aggregate.oldestAt ?? Number.POSITIVE_INFINITY))) {
+    return discardRetainedUsageSnapshot();
+  }
+  return discardRetainedUsageAggregate();
+}
+
 function providerDebugSnapshot(): RetainedStoreSnapshot {
   return ringSnapshot(debugBufferMetrics());
 }
@@ -154,7 +173,7 @@ export const APP_OWNED_RETAINED_STORE_REGISTRATIONS = [
     id: "usage_snapshot",
     category: "caches",
     snapshot: usageSnapshotRetainedStoreSnapshot,
-    evictOldest: discardRetainedUsageSnapshot,
+    evictOldest: evictOldestUsageSnapshot,
   },
   {
     id: "cursor_blobs",

src/server/management/api-key-usage.ts

@@ -1,9 +1,9 @@
 import {
   currentUsageLogRevision,
-  readUsageSnapshotForManagement,
   usageLogIdentityKey,
   type PersistedUsageEntry,
 } from "../../usage/log";
+import { scanUsageLedgerCooperatively } from "../../usage/ledger-scanner";
 
 /**
  * Per-key usage as the API tab renders it.
@@ -29,6 +29,11 @@ export interface ApiKeyUsageSnapshot {
   attributionSince?: string;
 }
 
+export interface ApiKeyUsageAccumulator {
+  add(entry: PersistedUsageEntry): void;
+  snapshot(): ApiKeyUsageSnapshot;
+}
+
 const SEVEN_DAYS_MS = 7 * 24 * 60 * 60 * 1000;
 
 /**
@@ -58,6 +63,21 @@ export function rollupApiKeyUsage(
   configuredIds: string[],
   now: number = Date.now(),
 ): ApiKeyUsageSnapshot {
+  const accumulator = createApiKeyUsageAccumulator(configuredIds, now);
+  for (const entry of entries) accumulator.add(entry);
+  return accumulator.snapshot();
+}
+
+/**
+ * Constant-memory fold for API-key attribution while the usage ledger streams.
+ *
+ * Only configured IDs are retained, so a hand-edited ledger containing an
+ * unbounded set of arbitrary `apiKeyId` values cannot grow this accumulator.
+ */
+export function createApiKeyUsageAccumulator(
+  configuredIds: string[],
+  now: number = Date.now(),
+): ApiKeyUsageAccumulator {
   const duplicated = new Set<string>();
   const seen = new Set<string>();
   for (const id of configuredIds) {
@@ -69,37 +89,40 @@ export function rollupApiKeyUsage(
   let attributionSince: number | undefined;
   const cutoff = now - SEVEN_DAYS_MS;
 
-  for (const entry of entries) {
-    if (!entry.admissionKind) continue;
-    const timestamp = usableTimestamp(entry.timestamp);
-    if (timestamp !== null && (attributionSince === undefined || timestamp < attributionSince)) {
-      attributionSince = timestamp;
-    }
-    if (entry.admissionKind !== "configured" || !entry.apiKeyId) continue;
-
-    const bucket = totals.get(entry.apiKeyId) ?? { requests7d: 0, totalRequests: 0 };
-    // The request happened even if its clock reading is unusable, so it still
-    // counts toward the total; only the time-based fields are skipped.
-    bucket.totalRequests += 1;
-    if (timestamp !== null) {
-      if (timestamp >= cutoff) bucket.requests7d += 1;
-      const iso = new Date(timestamp).toISOString();
-      if (!bucket.lastUsedAt || iso > bucket.lastUsedAt) bucket.lastUsedAt = iso;
-    }
-    totals.set(entry.apiKeyId, bucket);
-  }
-
-  const rollup = new Map<string, ApiKeyUsage>();
-  for (const id of configuredIds) {
-    if (duplicated.has(id)) {
-      rollup.set(id, { ambiguous: true });
-      continue;
-    }
-    rollup.set(id, totals.get(id) ?? { requests7d: 0, totalRequests: 0 });
-  }
   return {
-    rollup,
-    ...(attributionSince !== undefined ? { attributionSince: new Date(attributionSince).toISOString() } : {}),
+    add(entry) {
+      if (!entry.admissionKind) return;
+      const timestamp = usableTimestamp(entry.timestamp);
+      if (timestamp !== null && (attributionSince === undefined || timestamp < attributionSince)) {
+        attributionSince = timestamp;
+      }
+      if (entry.admissionKind !== "configured" || !entry.apiKeyId || !seen.has(entry.apiKeyId)) return;
+
+      const bucket = totals.get(entry.apiKeyId) ?? { requests7d: 0, totalRequests: 0 };
+      // The request happened even if its clock reading is unusable, so it still
+      // counts toward the total; only the time-based fields are skipped.
+      bucket.totalRequests += 1;
+      if (timestamp !== null) {
+        if (timestamp >= cutoff) bucket.requests7d += 1;
+        const iso = new Date(timestamp).toISOString();
+        if (!bucket.lastUsedAt || iso > bucket.lastUsedAt) bucket.lastUsedAt = iso;
+      }
+      totals.set(entry.apiKeyId, bucket);
+    },
+    snapshot() {
+      const rollup = new Map<string, ApiKeyUsage>();
+      for (const id of configuredIds) {
+        if (duplicated.has(id)) {
+          rollup.set(id, { ambiguous: true });
+          continue;
+        }
+        rollup.set(id, totals.get(id) ?? { requests7d: 0, totalRequests: 0 });
+      }
+      return {
+        rollup,
+        ...(attributionSince !== undefined ? { attributionSince: new Date(attributionSince).toISOString() } : {}),
+      };
+    },
   };
 }
 
@@ -111,6 +134,7 @@ export function rollupApiKeyUsage(
  * caching it costs nothing; a new row changes the revision and invalidates it.
  */
 let rollupCache: { revisionKey: string; expiresAt: number; lastSeenSize?: number; snapshot: ApiKeyUsageSnapshot } | null = null;
+const rollupFlights = new Map<string, Promise<ApiKeyUsageSnapshot>>();
 
 /**
  * The rollup is a function of the log AND of the clock: a request ages out of
@@ -127,6 +151,7 @@ const ROLLUP_CACHE_TTL_MS = 60_000;
 /** Test seam: the cache is module state and would otherwise leak between cases. */
 export function clearApiKeyUsageCacheForTests(): void {
   rollupCache = null;
+  rollupFlights.clear();
 }
 
 /**
@@ -159,6 +184,25 @@ export function cacheApiKeyUsageFromSnapshot(
   return rolled;
 }
 
+/** Seed the API-key cache from the accumulator already fed by `/api/usage`. */
+export function cacheApiKeyUsageFromRollup(
+  snapshot: ApiKeyUsageSnapshot,
+  configuredIds: string[],
+  identityKey: string,
+  lastSeenSize: number,
+  maxReadBytes: number | undefined,
+  now: number = Date.now(),
+): ApiKeyUsageSnapshot {
+  const idsKey = JSON.stringify([configuredIds, maxReadBytes]);
+  rollupCache = {
+    revisionKey: `${identityKey}|${idsKey}`,
+    expiresAt: now + ROLLUP_CACHE_TTL_MS,
+    lastSeenSize,
+    snapshot,
+  };
+  return snapshot;
+}
+
 export async function readApiKeyUsageRollup(configuredIds: string[], maxReadBytes?: number): Promise<ApiKeyUsageSnapshot> {
   // JSON rather than a joined string: ids are only validated as non-empty
   // strings, so `["a\0b","c"]` and `["a","b\0c"]` join to the same value and one
@@ -173,18 +217,28 @@ export async function readApiKeyUsageRollup(configuredIds: string[], maxReadByte
       return rollupCache.snapshot;
     }
 
-    const snapshot = await readUsageSnapshotForManagement(maxReadBytes);
-    const rolled = {
-      ...rollupApiKeyUsage(snapshot.entries, configuredIds, now),
-      ...(snapshot.truncatedPrefixBytes > 0 || snapshot.entriesTruncated ? { historyTruncated: true as const } : {}),
-    };
-    rollupCache = {
-      revisionKey: `${usageLogIdentityKey(snapshot.revision)}|${idsKey}`,
-      expiresAt: now + ROLLUP_CACHE_TTL_MS,
-      lastSeenSize: snapshot.revision?.size ?? 0,
-      snapshot: rolled,
-    };
-    return rolled;
+    const existing = rollupFlights.get(idsKey);
+    if (existing) return await existing;
+
+    const flight = (async (): Promise<ApiKeyUsageSnapshot> => {
+      const accumulator = createApiKeyUsageAccumulator(configuredIds, now);
+      const scan = await scanUsageLedgerCooperatively({ onEntry: entry => accumulator.add(entry) });
+      if (scan.oversizedRows > 0) throw new Error("usage ledger contains an oversized row");
+      return cacheApiKeyUsageFromRollup(
+        accumulator.snapshot(),
+        configuredIds,
+        usageLogIdentityKey(scan.revision),
+        scan.revision?.size ?? 0,
+        maxReadBytes,
+        now,
+      );
+    })();
+    rollupFlights.set(idsKey, flight);
+    try {
+      return await flight;
+    } finally {
+      if (rollupFlights.get(idsKey) === flight) rollupFlights.delete(idsKey);
+    }
   } catch {
     const rollup = new Map<string, ApiKeyUsage>();
     for (const id of configuredIds) rollup.set(id, { requests7d: 0, totalRequests: 0 });