diff --git a/devlog/_plan/260804_usage_rollup_preservation/000_research.md b/devlog/_plan/260804_usage_rollup_preservation/000_research.md new file mode 100644 index 000000000..b0aeed1d9 --- /dev/null +++ b/devlog/_plan/260804_usage_rollup_preservation/000_research.md @@ -0,0 +1,121 @@ +# 000 — Research: preserving usage history beyond the 64 MiB management read window + +## Problem + +`~/.opencodex/usage.jsonl` is append-only and unbounded (observed: 157 MB / 380k rows +after ~5 weeks). `readUsageSnapshotForManagement` reads only the newest +`managementUsageMaxReadBytes` (default 64 MiB) and caps parsed rows at +`MANAGEMENT_USAGE_MAX_ENTRIES` (200k). Once the file outgrows the window, every +`/api/usage` consumer silently loses the oldest days — the summary reports +`historyTruncated: true` but the data is simply absent from `days`/`models`/ +`providers` and from `summary` totals. Observed in production on 2026-08-04: +the 30d range was missing 11 of 30 days. + +Raising the byte window (256 MiB) or the entry cap (500k) only defers the loss and +makes every read slower. The fix: fold rows that leave the read window into a +compact daily aggregate sidecar, and make the reader merge "rollup (old) + raw +tail (recent)". + +## Prior-art survey (lunasearch, 5 lanes, Tier-2 source-proven 2026-08-04) + +### Lane 1 — TSDB downsampling (James) + +- Prometheus: block-merge compaction, aggregates live in *separate* recording-rule + series; safety boundary is the immutable block. [Storage docs](https://prometheus.io/docs/prometheus/latest/storage/) (primary). +- InfluxDB: raw bucket and aggregate bucket are separate retention domains; a + scheduled task windows recent data with an explicit lateness `offset`. + [Downsample and retain](https://docs.influxdata.com/enterprise_influxdb/v1/guides/downsample_and_retain/) (primary). +- Thanos Compactor: rewrites blocks into aggregate chunks *after* a watermark + (raw→5m only after 40h); single-compactor rule; halt on overlap. + [Compactor](https://thanos.io/tip/components/compact.md/) (primary). +- Takeaway: every mature system separates the raw store from the aggregate store + and only folds data past a stability boundary. + +### Lane 2 — append-only log compaction (Beauvoir) + +- Kafka never compacts the active segment; compaction reads older immutable + segments and atomically swaps results in. [Design](https://kafka.apache.org/41/design/design/) (primary). +- Readers merge "stable compacted snapshot + live tail by validated offset + boundary". Ordering and offsets never change. +- logrotate `copytruncate` documents a copy→truncate race that loses appends — + writer-uncoordinated truncation is inherently lossy. + [logrotate.conf(5)](https://man7.org/linux/man-pages/man5/logrotate.conf.5.html) (primary). +- JSONL: compact only complete newline-terminated records; quarantine a partial + final line. [jsonlines.org](https://jsonlines.org/) (lead). + +### Lane 3 — crash-safe ordering (Sartre) + +- SQLite WAL: persist WAL first, then the DB, then reset the WAL — the aggregate + never advances ahead of its source. [WAL](https://www.sqlite.org/wal.html) (primary). +- Atomic replace: temp file → fsync(temp) → rename → fsync(parent dir). + [fsync(2)] (primary). +- Two files cannot be flushed atomically; make the raw log authoritative, keep a + durable monotonic watermark in the sidecar, replay the raw suffix after the + watermark on recovery. Advancing the watermark *after* the aggregate is durable + converts crashes into idempotent re-work, never loss. + [USENIX ATC15 crash-consistency](https://www.usenix.org/system/files/conference/atc15/atc15-paper-min.pdf) (primary research). + +### Lane 4 — LLM CLI usage stores (Pascal) + +- Claude Code: append-only JSONL transcripts + a *separate* `stats-cache.json` + aggregate for `/usage` — the same raw/aggregate split at much coarser grain. + [Claude directory docs](https://code.claude.com/docs/en/claude-directory) (primary). +- Codex: documented unbounded-JSONL pathologies (issue #34061, ~755 GiB of + session JSONL) with no shipped rotation fix — nobody upstream has solved this + for us; also a warning that agents must not self-ingest their own logs + (issue #27131). +- ccusage consumes schema-complete JSONL/SQLite token records; our + `PersistedUsageEntry` is already schema-complete, so folding to aggregates + loses only per-request identity, not meterability. + +### Lane 5 — exactly-once boundaries (Averroes) + +- High-water-mark pattern: persist `(aggregate, mark)` together; read strictly + after the mark; advance the mark only with the aggregate. + [GOV.UK HWM guidance](https://pg-bulk-ingest.docs.trade.gov.uk/high-watermark/) (primary). +- A *source offset* answers "which records are consumed" (exactly-once); an + *event-time watermark* answers "when is a bucket final". For a local file the + offset is the correctness boundary — wall-clock time is unsafe. + [Flink streaming analytics](https://nightlies.apache.org/flink/flink-docs-stable/docs/learn-flink/streaming_analytics/) (primary). +- Emit deltas and add, or emit snapshots and replace — never add repeated + cumulative snapshots (Beam accumulating-pane trap). + +## Design conclusions adopted + +1. **Offset cutline, not date cutline.** The rollup covers exactly the byte range + `[0, cutlineOffset)` of a specific file lineage; the reader takes raw rows + from `[cutlineOffset, EOF]`. Dedupe is structural (disjoint byte ranges), so + out-of-order timestamps cannot double-count and boundary days merge additively + in the day grid. +2. **Rollup rows are deltas.** Each fold run appends aggregate rows covering only + the newly folded byte range. The reader sums all rollup rows plus the tail. + Repeated groups across fold runs are additive by construction. +3. **Ordering: fold → fsync rollup → advance meta (temp+fsync+rename).** A crash + between rollup append and meta advance re-folds the same range next run; to + make that idempotent each fold writes a `segment` record keyed by + `(lineageId, fromOffset)` and the folder skips ranges already recorded. +4. **Never truncate raw by default.** The window mechanism already ignores the + prefix; truncation is a policy decision deferred (documented for a follow-up). + No writer-coordination or copytruncate-style race exists as a result. +5. **Token aggregates only; cost stays display-time.** Cost is linear in tokens + for a fixed `(provider, model, tier, longContext)` price row, so grouping by + those keys preserves exact display-time recomputation and keeps price-table + fixes retroactive. Long-context is per-request non-linear, hence it is part of + the group key, evaluated at fold time with the same predicate used today. +6. **Two aggregation grains.** `summary`/`days` count *requests* (entry-level); + `models`/`providers` count *attributions* (attempt-level, combo-aware). The + rollup stores both kinds of rows explicitly rather than deriving one from the + other lossily. + +## Claim ledger + +| # | Claim | Source | Tier | +|---|-------|--------|------| +| 1 | Kafka excludes the active segment from compaction | kafka.apache.org/41/design | verified | +| 2 | copytruncate documents a lossy race window | man7 logrotate.conf(5) | verified | +| 3 | SQLite orders WAL-persist before DB-persist before WAL-reset | sqlite.org/wal.html | verified | +| 4 | Atomic replace requires temp+fsync+rename+dir-fsync | fsync(2) man page | verified | +| 5 | HWM advances only with its aggregate, source-derived not wall-clock | GOV.UK pg-bulk-ingest docs | verified | +| 6 | Prometheus/Thanos fold only past a stability watermark | prometheus.io, thanos.io | verified | +| 7 | Claude Code keeps raw JSONL + separate aggregate cache | code.claude.com | verified | +| 8 | Upstream Codex has no shipped rotation fix for JSONL growth | github.com/openai/codex#34061 | verified | diff --git a/devlog/_plan/260804_usage_rollup_preservation/001_roadmap.md b/devlog/_plan/260804_usage_rollup_preservation/001_roadmap.md new file mode 100644 index 000000000..b4abe9104 --- /dev/null +++ b/devlog/_plan/260804_usage_rollup_preservation/001_roadmap.md @@ -0,0 +1,120 @@ +# 001 — Roadmap: rollup sidecar as a rebuildable derived cache — rev2 (post-audit 002) + +## The one architectural decision everything else follows from + +**The raw log is never truncated, so the rollup is a derived cache, not a second +source of truth.** Any invalidation — lineage change (inode/birthtime), schema +version bump, price-table fingerprint mismatch — is answered by rebuilding the +rollup from the full raw file in a background cooperative parse. This dissolves +the two hardest problems found in research: + +- *Cost freezing*: rollup rows may store fold-time cost numbers (including exact + per-entry combo fail-closed decisions) because a price-table change flips the + fingerprint and triggers a rebuild. Display-time retroactivity is preserved by + rebuild, not by schema contortions. +- *Exactness*: the fold processes whole entries with the same functions the + display path uses (`bumpStatus`, `addTokens`, `addEstimatedCost`, + `usageAttributions`, `foldAttributionStatuses`), so + `summarize(fold(prefix) ⊕ tail)` ≡ `summarize(prefix ⊕ tail)` within the + exactness domain defined in 010 (unique requestIds across entries — an + assumption backed by the current generator — and no multi-model combo + overflow into "other"), + enforced with a property test; out-of-domain behavior is additive and + carries dedicated documenting tests (002 B2). + +## Correctness boundary + +- Cutline is a **byte offset at a local-day boundary** of the raw file: + rollup covers `[0, cutlineOffset)`, the reader's tail starts at + `cutlineOffset`. Disjoint byte ranges make additive merge structurally + dedupe-free; a late-timestamped row after the cutline simply counts in the + tail (still additive, never double-counted). +- **Single commit authority (002 B1):** the cutline is derived from validated + commit rows inside `usage-rollup.jsonl` itself — group rows first, a + `commit` row last, one fsync. Meta only carries lineage/fingerprint/throttle + via the existing `renameAtomicFile()`. Reader visibility and tail start + always agree because both derive from the same committed segments; a partial + append (no commit row / rowCount mismatch) is invisible garbage and the + retry re-folds the same range. +- Only *complete days* older than `ROLLUP_MIN_AGE_DAYS = 9` are folded + (stability watermark; keeps the entire 7d window tail-exact, 002 B3). + Fold-eligibility check and the fold run are throttled and cooperative (same + yield discipline as `parseUsageTextCooperatively`). +- **Same-lineage mutation policy (002 B5):** rebuild on lineage-tuple + mismatch, live size < committed cutline, or committed boundary-digest + mismatch (last 4 KiB of each committed segment re-verified at the next fold). + Deep hand-edits inside the folded prefix are answered by the documented + delete-the-rollup-files escape hatch (full rebuild), preserving the + hand-editable raw contract. + +## Range semantics + +- `all`: rollup days + tail — exact within the exactness domain (010; unique + requestIds assumed, no combo overflow into "other"). +- `7d`: tail-exact when `truncatedPrefixBytes === 0` — min-age 9 keeps the + fold away from it; an extreme-growth residual gap degrades to today's + reported truncation, never silently. +- `30d`: tail rows keep ms-precision filtering; rollup rows are day-grain, so + the one boundary day partially inside the window is included whole (≤ ~24h + overcount). This is steady-state behavior for days 9–30, documented in + docs-site (002 wording fix: not an exceptional state). + +## Row grains (one JSONL file, `kind`-discriminated) + +1. `meta` (separate file): `version`, `lineageKey`, `priceFingerprint`, + throttle stamp — correctness-light. +2. `commit`: `{ lineageKey, seg, attemptId, toOffset, rowCount, payloadDigest, + boundaryDigest, attributionSinceMs, oldestTimestampMs, foldedAt }` — + appended last; the segment's visibility gate and the cutline authority. + `attemptId` binds the commit to one append attempt so abandoned crash + leftovers can never collide with a retry (R2-1). +3. `day` rows: key `(date, surfaceKey)` → entry-grain status counts, + attemptCount, token sums, cost sums, priced/unpriced/unmetered counts. +4. `model` rows: key `(date, surfaceKey, providerKey, model)` → + distinct-request count, attemptCount, folded per-request status counts, + token sums, cost sums (attribution grain, combo-aware — exactly what + `buildModels`/day-model breakdown consume); `resolvedModel` rides along as + display-only first-seen. Separate `provider` rows store the within-entry + combo-deduped provider grain (not derivable from model rows). +5. `key` rows: key `(date, admissionKind, apiKeyId?)` → request count, + lastUsedAtMs max (for `api-key-usage.ts`). + +Cardinality: O(days × surfaces × models) — a few KB per day. + +## Cost (002 B4) + +Long-context tier selection depends on per-request `usage.inputTokens` +(`isLongContext`), so day-aggregated tokens cannot re-derive cost. The rollup +stores fold-time cost sums; retroactivity is preserved by rebuild: a wide +price fingerprint (full jawcode cost table + `EXPECTED_PRICE_OVERLAYS` + +`PRIORITY_MULTIPLIERS` + `CONTEXT_TIERS` + hand-bumped +`ROLLUP_COST_SEMANTICS_VERSION`) forces a full refold on any pricing or +estimator-semantics change. + +## Reader merge + +`readUsageSnapshotForManagement` gains a `fromOffset` mode: tail read starts at +`max(cutlineOffset, size - maxReadBytes)`. If `cutlineOffset < size - maxReadBytes` +the un-folded gap is reported as residual truncation (`truncatedPrefixBytes > 0`) +until the background fold catches up. `summarizeUsage` accepts optional rollup +contributions and merges them into totals/days/models/providers before the +display-time sort/cap steps. `historyTruncated` becomes false when rollup+tail +jointly cover the file. + +## Implementation phases (1 decade doc = 1 PABCD cycle) + +- **010** `src/usage/rollup.ts` core: types, fold, file IO, meta, fingerprint, + rebuild, idempotent segments + focused tests. +- **020** reader merge: `log.ts` offset tail, `summary.ts` merge, + `logs-usage-routes.ts` wiring + fold trigger, `api-key-usage.ts` merge, + config flag `usageRollupEnabled` (default true) + tests incl. the + fold⊕tail ≡ full property test. +- **030** real-data validation against the 157 MB production copy, docs-site, + push + dev PR. + +## Out of scope (documented for follow-ups) + +- Raw-file truncation/archival policy (requires writer coordination; research + lane 2 shows uncoordinated truncation is lossy). +- GUI changes beyond what the existing truncation metadata already drives. +- Go runtime (retired). diff --git a/devlog/_plan/260804_usage_rollup_preservation/002_audit_synthesis.md b/devlog/_plan/260804_usage_rollup_preservation/002_audit_synthesis.md new file mode 100644 index 000000000..a92ebba9e --- /dev/null +++ b/devlog/_plan/260804_usage_rollup_preservation/002_audit_synthesis.md @@ -0,0 +1,135 @@ +# 002 — Audit synthesis (round 1, reviewer: Carver/sol, verdict FAIL) + +Per-blocker RCA and accept/rebut decisions. All five blockers accepted; fixes +amended into 001/010/020 (marked "rev2"). + +## B1 crash safety — ACCEPT, redesign: commit-row authority + +RCA: two independently-visible files (rollup, meta) cannot share one commit +point; the window between rollup-fsync and meta-rename double-counts, and a +partial segment append could be accepted by the "skip existing fromOffset" +rule. + +Fix (rev2): **the cutline authority moves into the rollup file itself.** Every +group row carries `seg: fromOffset`; a fold appends group rows first and a +`{kind:"commit", seg, toOffset, rowCount}` row LAST, then fsyncs. A segment is +visible only when its commit row exists AND its row count matches; the +effective cutline is the max committed contiguous `toOffset`. Partial appends +(no commit row, or count mismatch) are permanently ignored garbage. Meta +becomes correctness-light (lineage, fingerprint, throttle stamp) and is +written AFTER the rollup fsync via the existing `renameAtomicFile()` +(config.ts:63, Windows-aware). Reader derives cutline and tail start from the +same single source, so no overlap window exists. Crash-injection tests at each +boundary (after rows, before commit; after commit, before fsync-visible meta). + +## B2 distinct-request exactness — ACCEPT with precisely-weakened claim + +RCA: buildModels/buildProviders dedupe requestIds range-wide and the overflow +"other" bucket unions requestIds across models; count-only rows cannot +reproduce unions. Production evidence (2026-08-04, 380,841 rows): exactly 3 +duplicate requestIds, all hand-written test fixtures ("ok-a","ok-b","fail"); +36 distinct models vs the 256-row cap. + +Fix (rev2): fold processes WHOLE entries, so within-entry combo dedupe is +exact by construction (one entry → one day → one segment; per entry, each +distinct model key contributes requests+=1 with folded status). The exactness +claim is scoped: equality holds when (a) requestIds are unique across entries +— the writer's contract (request-log.ts ocx--) — and (b) the +model cap does not overflow with multi-model combo requests inside "other". +Outside that domain the rollup is additive (documented divergence, count may +exceed by duplicates). Property test pins exact equality in-domain; two +dedicated tests document the out-of-domain additive behavior. + +## B3 requests7d — ACCEPT, fix by watermark: ROLLUP_MIN_AGE_DAYS = 9 + +RCA: min-age 2 meant the tail owns only ~2 days; requests7d and ms-precision +7d filtering would silently go day-grain. + +Fix (rev2): fold only days ≥ 9 local days old. The tail always owns the whole +7d window (+2-day margin): `requests7d`, the 7d range, and lastUsedAt stay +tail-exact with zero merge changes for them. Day-grain boundary approximation +now applies only to 30d (days 9–30). At production rate (~4.5 MB/day) 9 days ≈ +40 MB < 64 MiB; if traffic outgrows the window the residual gap is reported +via the existing truncation metadata (bounded regression, not silence). + +## B4 cost fingerprint — ACCEPT, widen inputs + manual version + +Fix (rev2): fingerprint = sha256 over stable-serialized {full generated +jawcode cost table, EXPECTED_PRICE_OVERLAYS, PRIORITY_MULTIPLIERS, +CONTEXT_TIERS} plus a hand-bumped `ROLLUP_COST_SEMANTICS_VERSION` constant +that estimator/canonicalization changes must increment (documented next to +the estimator). Any mismatch → rebuild from offset 0. + +## B5 same-lineage mutation — ACCEPT, bounded detection + escape hatch + +Fix (rev2): rebuild triggers: lineage tuple mismatch; live size < committed +cutline; boundary digest mismatch — each commit row stores sha256 of the last +4 KiB of its byte range, and the next fold run re-verifies the previous +committed segment's digest before appending. Hand-edits deep inside an +already-folded prefix are NOT auto-detected (fold-time-only verification); +documented: the rollup is a derived cache — delete both files to force a full +rebuild. This preserves the raw log's hand-editable contract with an explicit, +cheap policy instead of a silent one. + +## Advisories — all accepted + +0600 + recordOwnedConfigPath for BOTH files; uninstall-test coverage; reuse +`renameAtomicFile`; dir-fsync best-effort (no-op where unsupported, matching +platform reality); resolvedModel is display-only first-seen (identity stays +provider/model, matching buildModels); tz/DST/midnight/out-of-order tests; +research citations reworded as design analogies. + +--- + +# Round 2 (same reviewer, verdict FAIL — B3/B5 closed, 4 remaining) + +## R2-1 retry-attempt identity — ACCEPT + +RCA: abandoned uncommitted rows and retry rows share `seg = fromOffset`, so a +commit's `rowCount` can never match after a crash-retry (20 rows, commit says +10) — the retry deadlocks into permanent rejection. + +Fix (rev3): every fold attempt gets a unique `attemptId` (`-`) +carried by all its group rows and its commit row; the commit binds +`(seg, attemptId, rowCount, payloadDigest)` where `payloadDigest` is sha256 +over the serialized group rows of THAT attempt. Validation counts only rows +with the commit's attemptId and verifies the digest. Abandoned attempts are +permanently invisible garbage; a partial trailing line in the rollup file +itself is dropped by JSONL framing (complete-line parse), which also covers +the crash-mid-append case. + +## R2-2 synchronous validity gate — ACCEPT + +RCA: `ensureRollupCurrent()` is async fire-and-forget, so a request racing a +price change could merge stale costs if validity were only checked in the +folder. + +Fix (rev3): `readRollupSnapshot()` itself synchronously validates version, +lineage, and priceFingerprint against meta and the live raw file BEFORE +returning a snapshot; on any mismatch (or meta absent/corrupt) it returns +null — the route then serves raw-tail-only (legacy behavior, cutline 0) while +the background rebuild proceeds. Stale rollup data is structurally unreachable. + +## R2-3 exactness wording — ACCEPT + +`all` is "exact within the exactness domain" everywhere (001 range semantics, +020, docs-site). requestId uniqueness is stated as an assumption backed by the +current generator (`ocx--`), not a guarantee — a same-ms +restart collision is theoretically possible; out-of-domain behavior stays +additive-documented. + +## R2-4 write scope for fingerprint — ACCEPT + +`ROLLUP_COST_SEMANTICS_VERSION` lives in `src/usage/cost.ts` (beside the +estimator it versions). The generated module gains +`export const JAWCODE_TABLE_FINGERPRINT` computed by the generator at +generation time (packaged-runtime safe — no source-file reads at runtime). +Write scope for 010/020 adds: `src/usage/cost.ts` (constant only), +`scripts/generate-jawcode-metadata.ts` + regenerated +`src/generated/jawcode-model-metadata.ts`, `src/usage/expected-prices.ts` +(export `PRIORITY_MULTIPLIERS`/`CONTEXT_TIERS` if not already exported). + +## R2 advisory — ACCEPT + +7d wording: "tail-exact when `truncatedPrefixBytes === 0`"; the residual-gap +case degrades to today's truncated behavior, reported not silent. diff --git a/devlog/_plan/260804_usage_rollup_preservation/010_rollup_core.md b/devlog/_plan/260804_usage_rollup_preservation/010_rollup_core.md new file mode 100644 index 000000000..e974ea530 --- /dev/null +++ b/devlog/_plan/260804_usage_rollup_preservation/010_rollup_core.md @@ -0,0 +1,212 @@ +# 010 — Rollup core module (`src/usage/rollup.ts`) — rev3 (post-audit 002 R2) + +One PABCD cycle. Write scope: `src/usage/rollup.ts` (new), +`tests/usage-rollup.test.ts` (new), plus the fingerprint enablers (R2-4): +`src/usage/cost.ts` (add `ROLLUP_COST_SEMANTICS_VERSION` constant only), +`src/usage/expected-prices.ts` (export `PRIORITY_MULTIPLIERS`/`CONTEXT_TIERS` +if not already exported), `scripts/generate-jawcode-metadata.ts` + +regenerated `src/generated/jawcode-model-metadata.ts` +(`JAWCODE_TABLE_FINGERPRINT` computed at generation time). + +## Files + +- `usage-rollup.jsonl` — append-only, next to `usage.jsonl` (`getConfigDir()`), + 0600 like the raw log, registered via `recordOwnedConfigPath`. **This file is + the single commit authority** (002 B1): the effective cutline is derived from + its committed segments, never from meta. +- `usage-rollup-meta.json` — correctness-light (lineage, fingerprint, throttle + stamp). 0600, `recordOwnedConfigPath`, replaced via the existing + `renameAtomicFile()` (config.ts) after the rollup fsync. + +## Types (all exported for tests; field names final) + +```ts +export interface RollupMeta { + version: 1; // schema version; mismatch → rebuild + lineageKey: string; // dev\0ino\0birthtimeMs of usage.jsonl + priceFingerprint: string; // see below; mismatch → rebuild + lastFoldAttemptAt: number; // throttle stamp only + updatedAt: number; +} +// Commit row: appended LAST in a fold, after all group rows, then fsync. +// Every fold ATTEMPT has a unique attemptId carried by its group rows and its +// commit row (R2-1). A segment is visible ONLY when a commit row exists whose +// attemptId-matched group rows satisfy rowCount AND payloadDigest. Abandoned +// attempts (crash before commit) are permanently invisible garbage; retries use +// a fresh attemptId so they can never collide. Cutline = max contiguous +// committed toOffset. +export interface RollupCommitRow { + kind: "commit"; lineageKey: string; seg: number /* fromOffset */; toOffset: number; + attemptId: string; // `${foldedAt}-${rand}` — binds commit to one append attempt + rowCount: number; // group rows of THIS attempt + payloadDigest: string; // sha256 over this attempt's serialized group rows + boundaryDigest: string; // sha256 of the LAST 4 KiB of [seg, toOffset) raw bytes (002 B5) + attributionSinceMs: number | null; // per-segment min; reader takes min over committed segments + oldestTimestampMs: number | null; // per-segment min + foldedAt: number; +} +export interface RollupStatusCounts { reported: number; unreported: number; unsupported: number; estimated: number; } +export interface RollupTokenSums { inputTokens: number; outputTokens: number; cacheReadInputTokens: number; cacheCreationInputTokens: number; reasoningOutputTokens: number; totalTokens: number; } +export type RollupSurfaceKey = "codex" | "claude" | "claude-desktop" | "grok"; +export interface RollupDayRow { + kind: "day"; seg: number; attemptId: string; date: string; surface: RollupSurfaceKey; + statusCounts: RollupStatusCounts; attemptCount: number; + tokens: RollupTokenSums; + estimatedCostUsd: number; pricedRequests: number; unpricedRequests: number; unmeteredRequests: number; +} +export interface RollupModelRow { + kind: "model"; seg: number; attemptId: string; date: string; surface: RollupSurfaceKey; + provider: string; // baseProviderLabel() already applied + model: string; // usageModelIdentity() already applied + resolvedModel?: string; // display-only first-seen; identity stays provider/model (002 adv) + requests: number; // distinct requestIds this day×surface×model + attemptCount: number; + foldedStatusCounts: RollupStatusCounts; // foldAttributionStatuses() per request, then counted + tokens: Pick; + estimatedCostUsd: number; // attempt-attributed, matching buildModels cost pass +} +// providers stored explicitly: within-entry combo dedupe (one requestId across +// two models of one provider counts once) is applied at fold time per entry and +// cannot be recovered from model rows. +export interface RollupProviderRow { + kind: "provider"; seg: number; attemptId: string; date: string; surface: RollupSurfaceKey; provider: string; + requests: number; attemptCount: number; foldedStatusCounts: RollupStatusCounts; + totalTokens: number; estimatedCostUsd: number; +} +export interface RollupKeyRow { + kind: "key"; seg: number; attemptId: string; date: string; admissionKind: "configured" | "environment" | "loopback"; + apiKeyId?: string; requests: number; requestsWithTimestamp: number; lastUsedAtMs: number | null; +} +``` + +### Surfaces + +`surface` filtering has four disjoint predicates (`summarizeUsage`): `claude` = +`claude|claude-desktop`, `grok`, `codex` = `undefined`, `all` = everything. +`RollupSurfaceKey = "codex" | "claude" | "claude-desktop" | "grok"` — store the +raw stored value (with `codex` for undefined) and let the reader apply the same +predicates. Folding `claude-desktop` into `claude` at write time would bake in +today's display rule; storing the raw key keeps the fold lossless w.r.t. the +filter axes. + +### Price fingerprint (002 B4) + +`priceFingerprint = sha256(stableStringify({ semantics: ROLLUP_COST_SEMANTICS_VERSION, +jawcode: JAWCODE_TABLE_FINGERPRINT, overlays: EXPECTED_PRICE_OVERLAYS, +priority: PRIORITY_MULTIPLIERS, contextTiers: CONTEXT_TIERS }))`. +`JAWCODE_TABLE_FINGERPRINT` is emitted by the generator at generation time +(packaged-runtime safe — no source reads at runtime). +`ROLLUP_COST_SEMANTICS_VERSION` lives in `src/usage/cost.ts` next to the +estimator: any change to estimator/canonicalization behavior not visible in +the hashed tables must increment it (documented in a comment there). Any +mismatch → `rollupNeedsRebuild()` → discard both files and refold from 0. + +## Fold algorithm (`foldUsagePrefix`) — rev2 + +``` +recovery preflight: + read rollup file (complete JSONL lines only; a partial trailing line is dropped + by framing); validate each commit row by (attemptId, rowCount, payloadDigest) + against its attempt's group rows; committed cutline = max contiguous committed + toOffset; rows of uncommitted/failed attempts are permanently invisible garbage + verify previous committed segment's boundaryDigest against live raw bytes; + mismatch OR live size < cutline OR lineage/fingerprint mismatch → full rebuild from 0 +eligibleCutline(raw): + scan forward from committed cutline in 1 MiB chunks (readExactly, same as log.ts) + find the byte offset of the first row whose localDateKey(timestamp) >= (today - ROLLUP_MIN_AGE_DAYS) + clamp so the fold only covers COMPLETE newline-terminated rows + → { fromOffset: committed cutline, toOffset } +if toOffset <= fromOffset: NOOP +parse rows in [fromOffset, toOffset) cooperatively (yield every 1k lines) +accumulate day/model/provider/key groups using THE SAME helpers the display path + uses: normalizeUsageEntry, usageAttributions, foldAttributionStatuses, + usageDisplayTotalTokens, serviceTierContext + estimateComboCost/estimateRequestCost, + baseProviderLabel, usageModelIdentity, localDateKey (exported from summary.ts) +append-boundary repair: if usage-rollup.jsonl does not end in "\n" (crash left a + partial trailing line), truncate to the last complete newline before appending + (ftruncate at lastNewline+1) — never append after a fragment +mint attemptId; append group rows (tagged seg=fromOffset, attemptId), THEN the +commit row binding (seg, attemptId, rowCount, payloadDigest); fsync(fd) +write meta (throttle stamp) via renameAtomicFile; dir-fsync best-effort +``` + +`ROLLUP_MIN_AGE_DAYS = 9` (002 B3): the tail always owns the full 7d window +plus a 2-day margin, so `requests7d`, the 7d range, and lastUsedAt stay +tail-exact. Day-grain boundary approximation applies only to 30d. + +Crash behavior: a crash before the commit row leaves uncommitted garbage rows +that are never visible and never block a retry — the retry re-folds the same +range under a NEW attemptId, so abandoned rows can never satisfy (or poison) +a later commit's rowCount/payloadDigest (R2-1). A crash after fsync but before +the meta stamp costs only an early next throttle check. There is no state in +which reader-visible rollup data and the derived tail start disagree, because +both come from the same committed cutline. + +Hand-edits deep inside an already-folded prefix are NOT auto-detected (only the +committed boundary digest is re-verified at fold time). Documented policy: the +rollup is a derived cache — delete both files to force a full rebuild. + +## Read API + +```ts +export interface RollupSnapshot { + cutlineOffset: number; // derived from committed segments + attributionSinceMs: number | null; oldestTimestampMs: number | null; + days: RollupDayRow[]; models: RollupModelRow[]; providers: RollupProviderRow[]; keys: RollupKeyRow[]; +} +export function readRollupSnapshot(): RollupSnapshot | null; // null = absent/invalid/lineage-mismatch +export async function ensureRollupCurrent(): Promise; // fold-if-eligible, throttled (min 10 min between attempts), single-flight +export function resetRollupForTests(): void; +``` + +**Synchronous validity gate (R2-2):** `readRollupSnapshot()` itself validates +version, lineage (against the live raw file), and priceFingerprint BEFORE +returning; any mismatch or absent/corrupt meta returns null, and the caller +serves raw-tail-only (cutline 0, legacy behavior) while the background rebuild +proceeds. Stale-cost data is structurally unreachable regardless of the async +fold's timing. + +Reading includes only rows whose `seg` has a validated commit row; groups with +the same `(kind, date, surface, …key)` across committed segments merge +additively (delta-not-snapshot rule). Uncommitted/duplicate-seg rows are +dropped (first committed wins). + +### Exactness domain (002 B2) + +`summarize(fold(prefix) ⊕ tail) ≡ summarize(prefix ⊕ tail)` holds exactly when +(a) requestIds are unique across entries — an ASSUMPTION backed by the current +generator (`ocx--`, request-log.ts); a same-millisecond restart +collision is theoretically possible — and (b) multi-model combo requests +do not overflow into the 256-row "other" bucket. Within one entry, combo +dedupe is exact by construction (fold processes whole entries). Outside the +domain the merge is additive: duplicate hand-written requestIds may count more +than once, and "other" unions degrade to sums. Both divergences carry +dedicated documenting tests. Production check (2026-08-04, 380,841 rows): 3 +duplicate ids, all hand-written fixtures; 36 models vs the 256 cap. + +## Tests (tests/usage-rollup.test.ts) + +1. Fold of a 3-day fixture produces day/model/provider/key rows equal to + hand-computed aggregates (incl. combo attempts, claude-desktop surface, + unpriced model, estimated status). +2. Crash injection: (a) truncate the rollup file to cut the commit row (crash + mid-append) → rows invisible, retry folds the same range under a new + attemptId, totals exact and abandoned rows ignored; (b) abandoned attempt + followed by successful retry → validation counts only the committed + attempt's rows (the R2-1 collision case); (c) commit row with + rowCount/payloadDigest mismatch → segment rejected; (d) meta missing → + snapshot null, raw-tail fallback, rebuild path; (e) partial trailing group + row (no newline) followed by retry → append-boundary truncation repairs the + file, the retry's first row survives intact, and its commit validates. +3. Lineage mismatch → snapshot null, refold from 0 rebuilds. +4. Price-fingerprint mismatch → rebuild resets cutline to 0; boundary-digest + mismatch (prefix edited in place) → rebuild; live size < cutline → rebuild. +5. Cutline never lands mid-line; partial trailing line never folds. +6. Min-age watermark: rows younger than 9 local days never fold (tz-aware). +7. Property test (in-domain by construction — the generator emits unique + requestIds; out-of-domain cases live in the two documenting tests): for a + randomized fixture, summarize(foldPrefix ⊕ tail) equals + summarize(allRaw) for range "all" — this lands in 020 when the merge exists, + but the fixture generator is written here. Generator must include combo + attempts, duplicate requestIds across days (out-of-domain documenting case), + out-of-order timestamps, and fold-boundary-day entries. diff --git a/devlog/_plan/260804_usage_rollup_preservation/020_reader_merge.md b/devlog/_plan/260804_usage_rollup_preservation/020_reader_merge.md new file mode 100644 index 000000000..bd2295892 --- /dev/null +++ b/devlog/_plan/260804_usage_rollup_preservation/020_reader_merge.md @@ -0,0 +1,99 @@ +# 020 — Reader merge: rollup(old) + raw tail(recent) — rev3 (post-audit 002 R2) + +One PABCD cycle. Write scope: `src/usage/log.ts`, `src/usage/summary.ts`, +`src/server/management/logs-usage-routes.ts`, +`src/server/management/api-key-usage.ts`, `src/config.ts`, `src/types.ts`, +`tests/usage-rollup-merge.test.ts` (new), existing usage tests updated only +where signatures change. + +## log.ts — offset-aware tail + +`readUsageSnapshotForManagement(maxReadBytes, fromOffset = 0)`: + +- effective start = `max(fromOffset, size - maxReadBytes)`; existing + newline-realignment logic reused verbatim. +- `truncatedPrefixBytes` becomes the *residual* gap: `start - fromOffset` + clamp ≥ 0. With a caught-up rollup the residual is 0 and `historyTruncated` + turns false even at 157 MB. +- entry cap unchanged (200k applies to the tail only; rollup keeps history). +- in-flight sharing key gains the `fromOffset` component. + +## summary.ts — merge contributions + +`summarizeUsage(entries, range, now, surface, rollup?: RollupContribution)`: + +```ts +export interface RollupContribution { + days: RollupDayRow[]; models: RollupModelRow[]; + providers: RollupProviderRow[]; oldestTimestampMs: number | null; +} +``` + +- Surface filter: apply the same four disjoint predicates to row.surface. +- Range filter: include a rollup day when `localDate(date)` overlaps the + window; `all` includes everything. Day-grain inclusion affects only 30d in + the caught-up steady state (min-age 9 keeps 7d fully in the tail); the + boundary day may overcount by up to ~24h vs ms-filtering — documented in + docs-site and asserted in tests (002 B3/adv4). +- `summary` totals: add day-row statusCounts/attemptCount/tokens/cost fields + additively before `finalizeCoverage`. +- `days`: seed the grid from rollup day rows (additive with tail rows landing + on the same date — disjoint byte ranges make this pure addition). `all`-range + day count uses `min(oldestTimestampMs, oldest tail ts)`. +- `models`/`providers`: convert rollup rows into pre-aggregated + `UsageModel`/`UsageProvider` seeds (requests/attemptCount/statuses/tokens/ + cost), merge tail-built rows by the same key, then share-ratio/sort/cap as + today. Merged "other" buckets sum pre-aggregated request counts (additive; + exact within the 010 exactness domain). Per-day model breakdown + (`day.models`) merges rollup model rows for that date. +- No behavior change when `rollup` is undefined (all call sites outside the + usage route pass nothing). + +## logs-usage-routes.ts — wiring + +In `GET /api/usage`: + +1. `if (config.usageRollupEnabled !== false) void ensureRollupCurrent()` — + fire-and-forget, throttled internally; the fold never blocks the request. +2. `const rollup = readRollupSnapshot()` — synchronously validated (R2-2: + version/lineage/fingerprint checked inside; null on any mismatch → raw-tail + legacy path with cutline 0, so stale costs are unreachable); pass + `fromOffset = rollup?.cutlineOffset ?? 0` into the snapshot read and + the contribution into `summarizeUsage`. +3. Cache key/revision: append the rollup `cutlineOffset` (single commit + authority — a fold advance changes it) to the + revision key so a fold invalidates cached summaries. +4. Truncation metadata: `historyTruncated = residualPrefix > 0 || entriesTruncated`. + +## api-key-usage.ts — key rollup merge (002 B3) + +`readApiKeyUsageRollup` reads tail from the cutline and seeds +`totalRequests`/`attributionSince`/`lastUsedAt` from `RollupKeyRow`s. +`requests7d` is tail-exact when `truncatedPrefixBytes === 0` (the min-age-9 +watermark keeps the last 9 local days raw). If the byte window is smaller than +the un-folded suffix (extreme growth), the residual gap surfaces as the +existing `historyTruncated` metadata — a bounded, reported regression +identical to today's behavior, never silent. + +## config + +`usageRollupEnabled: z.boolean().default(true)` in config schema + types + +defaults. No new tuning knobs — cadence/min-age are module constants. + +## Tests + +1. Property test (fixture generator from 010): random entries across 40 days → + fold prefix, merge tail → `summarizeUsage` equality with full-raw parse for + `all` (totals, days, models, providers, costs to 1e-9), within the 010 + exactness domain; out-of-domain cases (duplicate ids, combo-into-other) + covered by dedicated documenting tests. +2. Boundary day: same date present in both rollup and tail → additive, not + doubled. +3. Surface filters against rollup rows (claude vs claude-desktop vs codex). +4. 7d/30d windows: 7d never touches rollup (min-age 9); 30d includes rollup + days inside the window, excludes outside, boundary-day whole-day inclusion + asserted; midnight/tz/out-of-order-timestamp cases. +5. Route-level: cache invalidates on fold advance; `historyTruncated` false + with caught-up rollup on an oversized file; flag off → legacy behavior. +6. api-key totals include folded history; `requests7d` exact vs full-raw + reference on a fixture whose history spans the fold boundary. diff --git a/devlog/_plan/260804_usage_rollup_preservation/030_validation_docs_pr.md b/devlog/_plan/260804_usage_rollup_preservation/030_validation_docs_pr.md new file mode 100644 index 000000000..ebad990c9 --- /dev/null +++ b/devlog/_plan/260804_usage_rollup_preservation/030_validation_docs_pr.md @@ -0,0 +1,38 @@ +# 030 — Real-data validation, docs, PR + +One PABCD cycle. Write scope: `docs-site/` usage/observability page(s), +`.tmp/` scratch for the production-copy validation, PR creation. + +## Real-data validation (evidence for goalplan c4) + +1. Copy `~/.opencodex/usage.jsonl` (157 MB, 380k rows) into a `mktemp -d` + sandbox with `OPENCODEX_HOME` pointed at it. +2. Reference: full-raw parse via `readUsageEntries()` + `summarizeUsage(all)` + in a bun script (no byte cap). +3. Candidate: run `foldUsagePrefix` to catch-up, then route-equivalent + `readUsageSnapshotForManagement(64 MiB, cutline)` + merge. +4. Assert equality: summary totals, day grid, models, providers (cost to 1e-6). +5. Record timings: fold duration, post-fold read duration vs 157 MB full parse. + +## docs-site + +Update the usage/monitoring page: what `usage-rollup.jsonl` / +`usage-rollup-meta.json` are, that history is preserved past the 64 MiB read +window, day-grain nuance for 7d/30d, `usageRollupEnabled` flag, and that +deleting the rollup files is safe (they rebuild). Check translated locales for +contradictions per repo policy (update English; note locale sync if present). + +## Gates before PR + +- `bun run typecheck` +- `bun run test` +- `bun run privacy:scan` +- `bun run lint:gui` only if gui/ touched (it is not). + +## PR + +- Push `codex/260804-usage-rollup`; PR targets `dev`. +- Description: problem (production truncation incident 2026-08-04), design + (offset cutline, delta segments, fold ordering, rebuildable cache), research + provenance summary, test evidence, real-data validation numbers, flag, + follow-ups (raw truncation policy deliberately out of scope). diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index bc72f17b3..fd85e6a39 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -21,6 +21,7 @@ runs helper features around provider requests. | `apiKeys?` | `OcxApiKey[]` | `[]` | Generated `ocx_…` credentials accepted by management and data-plane auth on non-loopback binds. Dashboard-managed. | | `storageCleanupPolicy?` | `StorageCleanupPolicy` | disabled | Opt-in archived-session cleanup policy. Never enabled implicitly. | | `appOwnedMemoryBudgetMb?` | `number` | `256` | Cap in MiB for evictable app-owned logs, caches, blobs, and continuation payloads. Range 64–4096; not an RSS cap. | +| `usageRollupEnabled?` | `boolean` | `true` | Preserve usage history beyond the bounded management read window by folding old `usage.jsonl` rows into a daily-aggregate sidecar (`usage-rollup.jsonl` + `usage-rollup-meta.json`). False serves summaries from the raw tail only, restoring the pre-rollup truncation behavior. | | `codexAutoStart?` | `boolean` | `true` | Let the Codex shim run `ocx ensure` before launching Codex. False makes ensure a no-op. | | `codexShimAutoRestore?` | `boolean` | `true` | Restore an installed shim after a completed external Codex update replaces it. Environment opt-out: `OPENCODEX_CODEX_SHIM_AUTO_RESTORE=0`. | | `syncResumeHistory?` | `boolean` | `true` | Reversible Codex App history compatibility. Original metadata is backed up and restored by `ocx stop` / `ocx restore`. | diff --git a/docs-site/src/content/docs/reference/management-api.md b/docs-site/src/content/docs/reference/management-api.md index 2635b719a..7e6b0e1b2 100644 --- a/docs-site/src/content/docs/reference/management-api.md +++ b/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 | Returns an `error: "read_failed"` summary if storage cannot be read | +| `GET /api/usage` | Summarize usage by range and client surface. History older than the bounded read window is served from the daily rollup sidecar (`usage-rollup.jsonl`) when the rollup is enabled and valid, so `all`-range summaries keep the full history; `7d` stays raw-exact while the raw tail covers it, and `30d` includes rolled-up days at day granularity. When the rollup is disabled, invalid, or still rebuilding, responses fall back to the bounded raw tail only, so older history may be temporarily absent and totals are lower bounds rather than exact. Deleting the rollup files is safe — they rebuild in the background. | 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 | diff --git a/scripts/generate-jawcode-metadata.ts b/scripts/generate-jawcode-metadata.ts index 8f3e2da80..a929a077a 100644 --- a/scripts/generate-jawcode-metadata.ts +++ b/scripts/generate-jawcode-metadata.ts @@ -1,4 +1,5 @@ import { existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"; +import { createHash } from "node:crypto"; import { dirname, resolve } from "node:path"; import { deriveJawcodeAliases } from "../src/providers/derive"; @@ -55,6 +56,15 @@ const EXCLUDED_MODELS = new Set([ ]); const lines: string[] = []; +function stableStringify(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]`; + if (value && typeof value === "object") { + const record = value as Record; + return `{${Object.keys(record).sort().map(key => `${JSON.stringify(key)}:${stableStringify(record[key])}`).join(",")}}`; + } + return JSON.stringify(value); +} + function compactRow(values: unknown[]): unknown[] { while (values.length > 0 && values[values.length - 1] === undefined) values.pop(); return values; @@ -77,7 +87,7 @@ lines.push(""); lines.push("const PROVIDER_ALIASES: Record = " + JSON.stringify(PROVIDER_ALIASES, null, 2) + " as const;"); lines.push(""); lines.push("type Row = readonly [id: string, contextWindow?: number | null, maxTokens?: number | null, input?: string | null, reasoning?: 0 | 1 | null, wireModelId?: string | null, costInput?: number | null, costOutput?: number | null, costCacheRead?: number | null, costCacheWrite?: number | null];"); -lines.push("const DATA: Record = {"); +const data: Record = {}; for (const provider of allowedProviders) { const models = registry[provider] ?? {}; @@ -96,11 +106,20 @@ for (const provider of allowedProviders) { model.cost?.cacheRead, model.cost?.cacheWrite, ])); - lines.push(` ${JSON.stringify(provider)}: ${JSON.stringify(rows)},`); + data[provider] = rows; } +lines.push("const DATA: Record = {"); +for (const provider of allowedProviders) { + lines.push(` ${JSON.stringify(provider)}: ${JSON.stringify(data[provider])},`); +} lines.push("};"); lines.push(""); +const tableFingerprint = createHash("sha256") + .update(stableStringify({ DATA: data, PROVIDER_ALIASES })) + .digest("hex"); +lines.push(`export const JAWCODE_TABLE_FINGERPRINT = ${JSON.stringify(tableFingerprint)};`); +lines.push(""); lines.push("export function resolveJawcodeProvider(provider: string): string | undefined {"); lines.push(" return PROVIDER_ALIASES[provider];"); lines.push("}"); diff --git a/src/config.ts b/src/config.ts index 8f9513deb..48ecd91d6 100644 --- a/src/config.ts +++ b/src/config.ts @@ -991,6 +991,7 @@ const clientIntegrationsSchema = z.object({ const configSchema = z.object({ port: z.number().int().min(0).max(65535).default(10100), managementUsageMaxReadBytes: z.number().int().positive().default(64 * 1024 * 1024), + usageRollupEnabled: z.boolean().default(true), appOwnedMemoryBudgetMb: z.number().int() .min(MIN_APP_OWNED_MEMORY_BUDGET_MB) .max(MAX_APP_OWNED_MEMORY_BUDGET_MB) @@ -2538,6 +2539,7 @@ export function getDefaultConfig(): OcxConfig { return { port: 10100, managementUsageMaxReadBytes: 64 * 1024 * 1024, + usageRollupEnabled: true, appOwnedMemoryBudgetMb: DEFAULT_APP_OWNED_MEMORY_BUDGET_BYTES / (1024 * 1024), // Fresh/re-initialized configs are already written in the current three-tier // OpenAI shape. Mark them as such so startup does not mistake them for a diff --git a/src/generated/jawcode-model-metadata.ts b/src/generated/jawcode-model-metadata.ts index 6615a12cd..2edfad66c 100644 --- a/src/generated/jawcode-model-metadata.ts +++ b/src/generated/jawcode-model-metadata.ts @@ -52,6 +52,8 @@ const DATA: Record = { "zai": [["glm-4.5",131072,98304,"text",1,null,0,0,0,0],["glm-4.5-air",131072,98304,"text",1,null,0,0,0,0],["glm-4.5-flash",131072,98304,"text",1,null,0,0,0,0],["glm-4.5v",64000,16384,"text,image",1,null,0,0,0,0],["glm-4.6",204800,131072,"text",1,null,0,0,0,0],["glm-4.6v",128000,32768,"text,image",1,null,0,0,0,0],["glm-4.7",204800,131072,"text",1,null,0,0,0,0],["glm-4.7-flash",200000,131072,"text",1,null,0,0,0,0],["glm-4.7-flashx",200000,131072,"text",1,null,0.07,0.4,0.01,0],["glm-5",204800,131072,"text",1,null,0,0,0,0],["glm-5-turbo",200000,131072,"text",1,null,0,0,0,0],["glm-5.1",200000,131072,"text",1,null,0,0,0,0],["glm-5.2",1000000,131072,"text",1,null,0,0,0,0],["glm-5v-turbo",200000,131072,"text,image",1,null,0,0,0,0]], }; +export const JAWCODE_TABLE_FINGERPRINT = "da3da34ecaa44481f8fa98178c687b880c28d6fb0305fc0093f753a9bd4149d9"; + export function resolveJawcodeProvider(provider: string): string | undefined { return PROVIDER_ALIASES[provider]; } diff --git a/src/server/management/api-key-usage.ts b/src/server/management/api-key-usage.ts index 25aacd65c..b384d32ac 100644 --- a/src/server/management/api-key-usage.ts +++ b/src/server/management/api-key-usage.ts @@ -4,6 +4,7 @@ import { usageLogRevisionKey, type PersistedUsageEntry, } from "../../usage/log"; +import { readRollupSnapshot, type RollupKeyRow } from "../../usage/rollup"; /** * Per-key usage as the API tab renders it. @@ -57,6 +58,8 @@ export function rollupApiKeyUsage( entries: PersistedUsageEntry[], configuredIds: string[], now: number = Date.now(), + foldedKeys: readonly RollupKeyRow[] = [], + foldedAttributionSinceMs: number | null = null, ): ApiKeyUsageSnapshot { const duplicated = new Set(); const seen = new Set(); @@ -66,9 +69,20 @@ export function rollupApiKeyUsage( } const totals = new Map(); - let attributionSince: number | undefined; + let attributionSince: number | undefined = foldedAttributionSinceMs ?? undefined; const cutoff = now - SEVEN_DAYS_MS; + for (const row of foldedKeys) { + if (row.admissionKind !== "configured" || !row.apiKeyId) continue; + const bucket = totals.get(row.apiKeyId) ?? { requests7d: 0, totalRequests: 0 }; + bucket.totalRequests += row.requests; + if (row.lastUsedAtMs !== null) { + const iso = new Date(row.lastUsedAtMs).toISOString(); + if (!bucket.lastUsedAt || iso > bucket.lastUsedAt) bucket.lastUsedAt = iso; + } + totals.set(row.apiKeyId, bucket); + } + for (const entry of entries) { if (!entry.admissionKind) continue; const timestamp = usableTimestamp(entry.timestamp); @@ -136,25 +150,29 @@ export function clearApiKeyUsageCacheForTests(): void { * `attributionSince`. Key management working matters more than usage numbers * being present, and the GUI already treats an absent field as "no data". */ -export async function readApiKeyUsageRollup(configuredIds: string[], maxReadBytes?: number): Promise { +export async function readApiKeyUsageRollup(configuredIds: string[], maxReadBytes?: number, rollupEnabled = true): Promise { // 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 // config could be served the other's cached rollup. - const idsKey = JSON.stringify([configuredIds, maxReadBytes]); + const idsKey = JSON.stringify([configuredIds, maxReadBytes, rollupEnabled]); const now = Date.now(); try { - const observedKey = `${usageLogRevisionKey(currentUsageLogRevision())}|${idsKey}`; + // Honor usageRollupEnabled the same way /api/usage does: a disabled rollup + // must not keep serving folded history from an orphaned sidecar. + const folded = rollupEnabled ? readRollupSnapshot() : null; + const cutlineOffset = folded?.cutlineOffset ?? 0; + const observedKey = `${usageLogRevisionKey(currentUsageLogRevision())}|${cutlineOffset}|${idsKey}`; if (rollupCache?.revisionKey === observedKey && now < rollupCache.expiresAt) { return rollupCache.snapshot; } - const snapshot = await readUsageSnapshotForManagement(maxReadBytes); + const snapshot = await readUsageSnapshotForManagement(maxReadBytes, cutlineOffset); const rolled = { - ...rollupApiKeyUsage(snapshot.entries, configuredIds, now), + ...rollupApiKeyUsage(snapshot.entries, configuredIds, now, folded?.keys, folded?.attributionSinceMs ?? null), ...(snapshot.truncatedPrefixBytes > 0 || snapshot.entriesTruncated ? { historyTruncated: true as const } : {}), }; rollupCache = { - revisionKey: `${usageLogRevisionKey(snapshot.revision)}|${idsKey}`, + revisionKey: `${usageLogRevisionKey(snapshot.revision)}|${cutlineOffset}|${idsKey}`, expiresAt: now + ROLLUP_CACHE_TTL_MS, snapshot: rolled, }; diff --git a/src/server/management/logs-usage-routes.ts b/src/server/management/logs-usage-routes.ts index 6501e5636..7b6b3a12e 100644 --- a/src/server/management/logs-usage-routes.ts +++ b/src/server/management/logs-usage-routes.ts @@ -55,6 +55,7 @@ import { } from "../../usage/log"; import { getUsageDebugLogEntries } from "../../usage/debug"; import { parseRange, parseUsageSurface, summarizeUsage, type UsageRange, type UsageSummary, type UsageSurface } from "../../usage/summary"; +import { ensureRollupCurrent, readRollupSnapshot } from "../../usage/rollup"; import { stripCodexRuntimeProviderFields } from "../../codex/auth-context"; import { getProviderRegistryEntry } from "../../providers/registry"; import { getDebugLogEntries } from "../../lib/debug-log-buffer"; @@ -191,23 +192,31 @@ export async function handleLogsUsageRoutes(ctx: ManagementContext): Promise 0 || snapshot.entriesTruncated, truncatedPrefixBytes: snapshot.truncatedPrefixBytes, entriesTruncated: snapshot.entriesTruncated, entriesDropped: snapshot.entriesDropped, }; setUsageSummaryCacheEntry(cacheKey, { - revisionKey: `${usageLogRevisionKey(snapshot.revision)}\0${effectiveReadLimit}`, + revisionKey: `${usageLogRevisionKey(snapshot.revision)}\0${effectiveReadLimit}\0${cutlineOffset}`, expiresAt: usageSummaryExpiresAt(snapshot.entries, range, surface, now), revisionReadAt, summary, diff --git a/src/server/management/oauth-account-routes.ts b/src/server/management/oauth-account-routes.ts index f3db21c6a..f3517dbb2 100644 --- a/src/server/management/oauth-account-routes.ts +++ b/src/server/management/oauth-account-routes.ts @@ -491,7 +491,7 @@ export async function handleOauthAccountRoutes(ctx: ManagementContext): Promise< requestOrigin: req.headers.get("origin"), }); const { readApiKeyUsageRollup } = await import("./api-key-usage"); - const { rollup, attributionSince, historyTruncated } = await readApiKeyUsageRollup(keys.map(k => k.id), config.managementUsageMaxReadBytes); + const { rollup, attributionSince, historyTruncated } = await readApiKeyUsageRollup(keys.map(k => k.id), config.managementUsageMaxReadBytes, config.usageRollupEnabled !== false); return jsonResponse({ // 8 random hex past the fixed `ocx_data_` literal: enough to tell two keys // apart in a list, with 128 bits of the tail still unrevealed. Masking only diff --git a/src/types.ts b/src/types.ts index 171628caf..0c60495da 100644 --- a/src/types.ts +++ b/src/types.ts @@ -559,6 +559,8 @@ export interface OcxConfig { port: number; /** Maximum usage-log bytes read for one management snapshot. */ managementUsageMaxReadBytes?: number; + /** Rebuild and merge the derived usage-history rollup cache. Defaults to true. */ + usageRollupEnabled?: boolean; providers: Record; defaultProvider: string; /** OpenAI provider-contract migration marker (v2 = single `openai` provider with account mode). */ diff --git a/src/usage/cost.ts b/src/usage/cost.ts index 45f78c626..fa563e079 100644 Binary files a/src/usage/cost.ts and b/src/usage/cost.ts differ diff --git a/src/usage/log.ts b/src/usage/log.ts index b86aa05d3..131d41167 100644 --- a/src/usage/log.ts +++ b/src/usage/log.ts @@ -320,7 +320,7 @@ export function normalizeUsageEntryForTest(entry: PersistedUsageEntry): Persiste return normalizeUsageEntry(entry); } -function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry { +export function normalizeUsageEntry(entry: PersistedUsageEntry): PersistedUsageEntry { const attempts = normalizedAttempts(entry.attempts); const routeDecision = entry.routeDecision ? normalizeRouteDecisionTrace(entry.routeDecision) @@ -523,13 +523,15 @@ async function readUsageEntriesFullCooperatively( path: string, signal: AbortSignal, maxReadBytes: number, + fromOffset: number, ): Promise { let fd: number | undefined; try { fd = openSync(path, "r"); const stat = fstatSync(fd); const size = Number(stat.size); - const start = Math.max(0, size - maxReadBytes); + if (fromOffset > size) throw new Error("usage log changed while it was being read"); + const start = Math.max(fromOffset, size - maxReadBytes); const chunks: Buffer[] = []; for (let position = start; position < size;) { if (signal.aborted) throw signal.reason; @@ -540,7 +542,7 @@ async function readUsageEntriesFullCooperatively( position += length; } let bytes = Buffer.concat(chunks); - let truncatedPrefixBytes = start; + let truncatedPrefixBytes = Math.max(0, start - fromOffset); if (start > 0) { const preceding = readExactly(fd, 1, start - 1); if (preceding === null) throw new Error("usage log changed while it was being read"); @@ -574,7 +576,10 @@ async function readUsageEntriesFullCooperatively( * callers share work only when they observed the same exact file revision. Parsed rows * are returned to the request and never retained in module state. */ -export async function readUsageSnapshotForManagement(maxReadBytes = MANAGEMENT_USAGE_MAX_READ_BYTES): Promise<{ +export async function readUsageSnapshotForManagement( + maxReadBytes = MANAGEMENT_USAGE_MAX_READ_BYTES, + fromOffset = 0, +): Promise<{ entries: PersistedUsageEntry[]; revision: UsageLogRevision | null; truncatedPrefixBytes: number; @@ -582,10 +587,11 @@ export async function readUsageSnapshotForManagement(maxReadBytes = MANAGEMENT_U entriesDropped: number; }> { if (!Number.isSafeInteger(maxReadBytes) || maxReadBytes <= 0) throw new RangeError("management usage max read bytes must be positive"); + if (!Number.isSafeInteger(fromOffset) || fromOffset < 0) throw new RangeError("management usage from offset must be non-negative"); const path = usageLogPath(); if (!existsSync(path)) return { entries: [], revision: null, truncatedPrefixBytes: 0, entriesTruncated: false, entriesDropped: 0 }; const observed = currentUsageLogRevision(); - const key = `${usageLogRevisionKey(observed)}\0${maxReadBytes}`; + const key = `${usageLogRevisionKey(observed)}\0${maxReadBytes}\0${fromOffset}`; const existing = managementUsageReadInflight; if (existing?.key === key && Date.now() - existing.startedAt <= MANAGEMENT_USAGE_FLIGHT_STALE_MS) { const shared = await existing.promise; @@ -593,7 +599,7 @@ export async function readUsageSnapshotForManagement(maxReadBytes = MANAGEMENT_U } existing?.abort.abort(new Error("management usage read superseded")); const abort = new AbortController(); - const promise = readUsageEntriesFullCooperatively(path, abort.signal, maxReadBytes); + const promise = readUsageEntriesFullCooperatively(path, abort.signal, maxReadBytes, fromOffset); managementUsageReadInflight = { key, promise, startedAt: Date.now(), abort }; try { const snapshot = await promise; diff --git a/src/usage/rollup.ts b/src/usage/rollup.ts new file mode 100644 index 000000000..9d79b61ca --- /dev/null +++ b/src/usage/rollup.ts @@ -0,0 +1,890 @@ +import { + chmodSync, + closeSync, + constants, + existsSync, + fstatSync, + fsyncSync, + ftruncateSync, + mkdirSync, + openSync, + readFileSync, + readSync, + unlinkSync, + writeFileSync, + writeSync, +} from "node:fs"; +import { createHash, randomBytes } from "node:crypto"; +import { join } from "node:path"; +import { getConfigDir, renameAtomicFile } from "../config"; +import { recordOwnedConfigPath } from "../lib/config-ownership"; +import { baseProviderLabel } from "../providers/label"; +import { JAWCODE_TABLE_FINGERPRINT } from "../generated/jawcode-model-metadata"; +import { + currentUsageLogRevision, + normalizeUsageEntry, + usageLogPath, + usageLogRevisionKey, + type PersistedUsageEntry, + type UsageStatus, +} from "./log"; +import { + estimateComboCost, + estimateRequestCost, + ROLLUP_COST_SEMANTICS_VERSION, + serviceTierContext, +} from "./cost"; +import { + CONTEXT_TIERS, + EXPECTED_PRICE_OVERLAYS, + PRIORITY_MULTIPLIERS, +} from "./expected-prices"; +import { + foldAttributionStatuses, + localDateKey, + usageAttributions, + usageModelIdentity, + type UsageAttribution, +} from "./summary"; +import { usageDisplayTotalTokens } from "./totals"; + +export const ROLLUP_MIN_AGE_DAYS = 9; +const ROLLUP_ATTEMPT_THROTTLE_MS = 10 * 60 * 1_000; +const READ_CHUNK_BYTES = 1024 * 1024; +const BOUNDARY_DIGEST_BYTES = 4 * 1024; +/** + * Upper bound on the raw-byte span folded into ONE segment per iteration. The + * fold loop still catches up a larger backlog in a single call, but each + * iteration materializes at most this many bytes of parsed entries, so first + * fold over a multi-GB usage.jsonl is bounded-memory instead of loading the + * whole eligible prefix (review thread: unbounded prefix materialization). + */ +let foldSegmentCapBytes = 64 * 1024 * 1024; + +/** Test seam: shrink the per-segment fold cap so bounded-fold behavior is testable with small fixtures. */ +export function setFoldSegmentCapForTests(bytes: number | null): void { + foldSegmentCapBytes = bytes ?? 64 * 1024 * 1024; +} + +export interface RollupMeta { + version: 1; + lineageKey: string; + priceFingerprint: string; + lastFoldAttemptAt: number; + updatedAt: number; +} + +export interface RollupCommitRow { + kind: "commit"; lineageKey: string; seg: number; toOffset: number; + attemptId: string; rowCount: number; payloadDigest: string; boundaryDigest: string; + attributionSinceMs: number | null; oldestTimestampMs: number | null; foldedAt: number; +} + +export interface RollupStatusCounts { + reported: number; unreported: number; unsupported: number; estimated: number; +} + +export interface RollupTokenSums { + inputTokens: number; outputTokens: number; cacheReadInputTokens: number; + cacheCreationInputTokens: number; reasoningOutputTokens: number; totalTokens: number; +} + +export type RollupSurfaceKey = "codex" | "claude" | "claude-desktop" | "grok"; + +export interface RollupDayRow { + kind: "day"; seg: number; attemptId: string; date: string; surface: RollupSurfaceKey; + statusCounts: RollupStatusCounts; attemptCount: number; tokens: RollupTokenSums; + estimatedCostUsd: number; pricedRequests: number; unpricedRequests: number; unmeteredRequests: number; +} + +export interface RollupModelRow { + kind: "model"; seg: number; attemptId: string; date: string; surface: RollupSurfaceKey; + provider: string; model: string; resolvedModel?: string; requests: number; attemptCount: number; + foldedStatusCounts: RollupStatusCounts; + tokens: Pick; + estimatedCostUsd: number; +} + +export interface RollupProviderRow { + kind: "provider"; seg: number; attemptId: string; date: string; surface: RollupSurfaceKey; provider: string; + requests: number; attemptCount: number; foldedStatusCounts: RollupStatusCounts; + totalTokens: number; estimatedCostUsd: number; +} + +export interface RollupKeyRow { + kind: "key"; seg: number; attemptId: string; date: string; + admissionKind: "configured" | "environment" | "loopback"; + apiKeyId?: string; requests: number; requestsWithTimestamp: number; lastUsedAtMs: number | null; +} + +export interface RollupSnapshot { + cutlineOffset: number; + attributionSinceMs: number | null; + oldestTimestampMs: number | null; + days: RollupDayRow[]; + models: RollupModelRow[]; + providers: RollupProviderRow[]; + keys: RollupKeyRow[]; +} + +type RollupGroupRow = RollupDayRow | RollupModelRow | RollupProviderRow | RollupKeyRow; +type ParsedSegment = { commit: RollupCommitRow; rows: RollupGroupRow[] }; + +interface ParsedRollup { + snapshot: RollupSnapshot; + segments: ParsedSegment[]; +} + +interface DayAccumulator { + statusCounts: RollupStatusCounts; + attemptCount: number; + tokens: RollupTokenSums; + estimatedCostUsd: number; + pricedRequests: number; + unpricedRequests: number; + unmeteredRequests: number; +} + +interface AttributionAccumulator { + firstResolvedModel?: string; + attemptCount: number; + statusesByRequest: Map; + inputTokens: number; + outputTokens: number; + totalTokens: number; + estimatedCostUsd: number; +} + +interface ProviderAccumulator { + attemptCount: number; + statusesByRequest: Map; + totalTokens: number; + estimatedCostUsd: number; +} + +interface KeyAccumulator { + requests: number; + requestsWithTimestamp: number; + lastUsedAtMs: number | null; +} + +let rollupFlight: Promise | null = null; +/** Memoized read-time boundary validation, keyed on the raw log's revision. */ +let boundaryValidationCache: { key: string; valid: boolean } | null = null; + +function rollupPath(): string { + return join(getConfigDir(), "usage-rollup.jsonl"); +} + +function rollupMetaPath(): string { + return join(getConfigDir(), "usage-rollup-meta.json"); +} + +function sha256(value: string | Uint8Array): string { + return createHash("sha256").update(value).digest("hex"); +} + +function stableStringify(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]`; + if (value && typeof value === "object") { + const record = value as Record; + return `{${Object.keys(record).sort().map(key => `${JSON.stringify(key)}:${stableStringify(record[key])}`).join(",")}}`; + } + return JSON.stringify(value); +} + +export function currentRollupPriceFingerprint(): string { + return sha256(stableStringify({ + semantics: ROLLUP_COST_SEMANTICS_VERSION, + jawcode: JAWCODE_TABLE_FINGERPRINT, + overlays: EXPECTED_PRICE_OVERLAYS, + priority: PRIORITY_MULTIPLIERS, + contextTiers: CONTEXT_TIERS, + })); +} + +function lineageKey(): string | null { + const revision = currentUsageLogRevision(); + return revision ? `${revision.dev}\0${revision.ino}\0${revision.birthtimeMs}` : null; +} + +function readExactly(fd: number, length: number, position: number): Buffer | null { + const output = Buffer.allocUnsafe(length); + let offset = 0; + while (offset < length) { + const count = readSync(fd, output, offset, length - offset, position + offset); + if (count === 0) return null; + offset += count; + } + return output; +} + +function blankStatuses(): RollupStatusCounts { + return { reported: 0, unreported: 0, unsupported: 0, estimated: 0 }; +} + +function blankTokens(): RollupTokenSums { + return { + inputTokens: 0, + outputTokens: 0, + cacheReadInputTokens: 0, + cacheCreationInputTokens: 0, + reasoningOutputTokens: 0, + totalTokens: 0, + }; +} + +function bumpStatus(counts: RollupStatusCounts, status: UsageStatus): void { + counts[status] += 1; +} + +function addStatusCounts(target: RollupStatusCounts, source: RollupStatusCounts): void { + target.reported += source.reported; + target.unreported += source.unreported; + target.unsupported += source.unsupported; + target.estimated += source.estimated; +} + +function addTokenSums(target: RollupTokenSums, source: RollupTokenSums): void { + target.inputTokens += source.inputTokens; + target.outputTokens += source.outputTokens; + target.cacheReadInputTokens += source.cacheReadInputTokens; + target.cacheCreationInputTokens += source.cacheCreationInputTokens; + target.reasoningOutputTokens += source.reasoningOutputTokens; + target.totalTokens += source.totalTokens; +} + +function entryTokenSums(entry: Pick): RollupTokenSums { + const sums = blankTokens(); + if (!entry.usage) return sums; + sums.inputTokens = entry.usage.inputTokens; + sums.outputTokens = entry.usage.outputTokens; + const creation = entry.usage.cacheCreationInputTokens ?? 0; + const read = typeof entry.usage.cacheReadInputTokens === "number" + ? entry.usage.cacheReadInputTokens + : typeof entry.usage.cachedInputTokens === "number" && typeof entry.usage.cacheCreationInputTokens === "number" + ? Math.max(0, entry.usage.cachedInputTokens - entry.usage.cacheCreationInputTokens) + : entry.usage.cachedInputTokens ?? 0; + sums.cacheReadInputTokens = read; + sums.cacheCreationInputTokens = creation; + sums.reasoningOutputTokens = entry.usage.reasoningOutputTokens ?? 0; + sums.totalTokens = usageDisplayTotalTokens(entry.usage, entry.totalTokens) ?? 0; + return sums; +} + +function surfaceKey(entry: PersistedUsageEntry): RollupSurfaceKey { + return entry.surface ?? "codex"; +} + +function usableTimestamp(value: unknown): value is number { + return typeof value === "number" && Number.isFinite(value) && !Number.isNaN(new Date(value).getTime()); +} + +function isObject(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +function isGroupRow(value: unknown): value is RollupGroupRow { + if (!isObject(value) || typeof value.seg !== "number" || typeof value.attemptId !== "string") return false; + return value.kind === "day" || value.kind === "model" || value.kind === "provider" || value.kind === "key"; +} + +function isCommitRow(value: unknown): value is RollupCommitRow { + if (!isObject(value) || value.kind !== "commit") return false; + return typeof value.lineageKey === "string" + && typeof value.seg === "number" && Number.isSafeInteger(value.seg) && value.seg >= 0 + && typeof value.toOffset === "number" && Number.isSafeInteger(value.toOffset) && value.toOffset > value.seg + && typeof value.attemptId === "string" && value.attemptId.length > 0 + && typeof value.rowCount === "number" && Number.isSafeInteger(value.rowCount) && value.rowCount >= 0 + && typeof value.payloadDigest === "string" && typeof value.boundaryDigest === "string" + && (value.attributionSinceMs === null || typeof value.attributionSinceMs === "number") + && (value.oldestTimestampMs === null || typeof value.oldestTimestampMs === "number") + && typeof value.foldedAt === "number"; +} + +function readMeta(): RollupMeta | null { + try { + const parsed = JSON.parse(readFileSync(rollupMetaPath(), "utf8")) as unknown; + if (!isObject(parsed)) return null; + if (parsed.version !== 1 || typeof parsed.lineageKey !== "string" + || typeof parsed.priceFingerprint !== "string" + || typeof parsed.lastFoldAttemptAt !== "number" || typeof parsed.updatedAt !== "number") return null; + return parsed as unknown as RollupMeta; + } catch { + return null; + } +} + +function parseRollup(expectedLineage: string): ParsedRollup { + const empty: RollupSnapshot = { + cutlineOffset: 0, + attributionSinceMs: null, + oldestTimestampMs: null, + days: [], models: [], providers: [], keys: [], + }; + if (!existsSync(rollupPath())) return { snapshot: empty, segments: [] }; + let text: string; + try { + text = readFileSync(rollupPath(), "utf8"); + } catch { + return { snapshot: empty, segments: [] }; + } + const lastNewline = text.lastIndexOf("\n"); + if (lastNewline < 0) return { snapshot: empty, segments: [] }; + const attempts = new Map(); + const validBySeg = new Map(); + for (const line of text.slice(0, lastNewline + 1).split("\n")) { + if (!line.trim()) continue; + let parsed: unknown; + try { parsed = JSON.parse(line); } catch { continue; } + if (isGroupRow(parsed)) { + const key = `${parsed.seg}\0${parsed.attemptId}`; + const attempt = attempts.get(key) ?? { rows: [], payload: "" }; + attempt.rows.push(parsed); + attempt.payload += `${line}\n`; + attempts.set(key, attempt); + continue; + } + if (!isCommitRow(parsed) || parsed.lineageKey !== expectedLineage || validBySeg.has(parsed.seg)) continue; + const attempt = attempts.get(`${parsed.seg}\0${parsed.attemptId}`); + if (!attempt || attempt.rows.length !== parsed.rowCount || sha256(attempt.payload) !== parsed.payloadDigest) continue; + validBySeg.set(parsed.seg, { commit: parsed, rows: [...attempt.rows] }); + } + const segments: ParsedSegment[] = []; + for (let offset = 0; validBySeg.has(offset);) { + const segment = validBySeg.get(offset)!; + segments.push(segment); + offset = segment.commit.toOffset; + } + return { snapshot: mergeSegments(segments), segments }; +} + +function mergeSegments(segments: ParsedSegment[]): RollupSnapshot { + const days = new Map(); + const models = new Map(); + const providers = new Map(); + const keys = new Map(); + let attributionSinceMs: number | null = null; + let oldestTimestampMs: number | null = null; + for (const { commit, rows } of segments) { + if (commit.attributionSinceMs !== null) attributionSinceMs = attributionSinceMs === null + ? commit.attributionSinceMs : Math.min(attributionSinceMs, commit.attributionSinceMs); + if (commit.oldestTimestampMs !== null) oldestTimestampMs = oldestTimestampMs === null + ? commit.oldestTimestampMs : Math.min(oldestTimestampMs, commit.oldestTimestampMs); + for (const row of rows) mergeGroupRow(row, days, models, providers, keys); + } + return { + cutlineOffset: segments.at(-1)?.commit.toOffset ?? 0, + attributionSinceMs, + oldestTimestampMs, + days: [...days.values()], + models: [...models.values()], + providers: [...providers.values()], + keys: [...keys.values()], + }; +} + +function mergeGroupRow( + row: RollupGroupRow, + days: Map, + models: Map, + providers: Map, + keys: Map, +): void { + if (row.kind === "day") { + const key = `${row.date}\0${row.surface}`; + const current = days.get(key); + if (!current) { days.set(key, structuredClone(row)); return; } + addStatusCounts(current.statusCounts, row.statusCounts); + addTokenSums(current.tokens, row.tokens); + current.attemptCount += row.attemptCount; + current.estimatedCostUsd += row.estimatedCostUsd; + current.pricedRequests += row.pricedRequests; + current.unpricedRequests += row.unpricedRequests; + current.unmeteredRequests += row.unmeteredRequests; + return; + } + if (row.kind === "model") { + const key = `${row.date}\0${row.surface}\0${row.provider}\0${row.model}`; + const current = models.get(key); + if (!current) { models.set(key, structuredClone(row)); return; } + current.requests += row.requests; + current.attemptCount += row.attemptCount; + addStatusCounts(current.foldedStatusCounts, row.foldedStatusCounts); + current.tokens.inputTokens += row.tokens.inputTokens; + current.tokens.outputTokens += row.tokens.outputTokens; + current.tokens.totalTokens += row.tokens.totalTokens; + current.estimatedCostUsd += row.estimatedCostUsd; + return; + } + if (row.kind === "provider") { + const key = `${row.date}\0${row.surface}\0${row.provider}`; + const current = providers.get(key); + if (!current) { providers.set(key, structuredClone(row)); return; } + current.requests += row.requests; + current.attemptCount += row.attemptCount; + addStatusCounts(current.foldedStatusCounts, row.foldedStatusCounts); + current.totalTokens += row.totalTokens; + current.estimatedCostUsd += row.estimatedCostUsd; + return; + } + const key = `${row.date}\0${row.admissionKind}\0${row.apiKeyId ?? ""}`; + const current = keys.get(key); + if (!current) { keys.set(key, structuredClone(row)); return; } + current.requests += row.requests; + current.requestsWithTimestamp += row.requestsWithTimestamp; + if (row.lastUsedAtMs !== null) current.lastUsedAtMs = current.lastUsedAtMs === null + ? row.lastUsedAtMs : Math.max(current.lastUsedAtMs, row.lastUsedAtMs); +} + +export function readRollupSnapshot(): RollupSnapshot | null { + try { + if (!existsSync(rollupPath())) return null; + const meta = readMeta(); + const currentLineage = lineageKey(); + if (!meta || !currentLineage || meta.version !== 1 + || meta.lineageKey !== currentLineage + || meta.priceFingerprint !== currentRollupPriceFingerprint()) return null; + const parsed = parseRollup(currentLineage); + // Read-time boundary validity (review threads: truncated/rewritten raw log + // within a throttle window must not serve a stale sidecar). The fold path + // performs the same check before appending; readers need it too because a + // truncate-then-regrow can happen entirely between folds. EVERY segment's + // boundary window is validated, not just the last — a same-size rewrite of + // an earlier folded span would leave the final segment's tail intact. + // + // Scope, stated honestly: each boundary digest covers the trailing + // BOUNDARY_DIGEST_BYTES of its segment, so a surgical same-size rewrite + // strictly inside a segment's untailed span is not detected here. That is + // the fold-time contract too (the sidecar trusts an append-only log); the + // read-time check exists to catch the realistic failure — truncation, + // rotation, regrowth — not an adversarial editor with byte-level care. + // + // Cost: the validation is memoized per raw-log revision (dev/ino/size/ + // mtime/ctime), so cached hot paths pay one fstat, not a re-hash of every + // segment on every call. + if (parsed.segments.length > 0) { + const revisionKey = `${usageLogRevisionKey(currentUsageLogRevision())}|${parsed.snapshot.cutlineOffset}|${parsed.segments.length}`; + if (boundaryValidationCache?.key === revisionKey) { + if (!boundaryValidationCache.valid) return null; + } else { + let valid = false; + let fd: number | undefined; + try { + fd = openSync(usageLogPath(), "r"); + const rawSize = Number(fstatSync(fd).size); + valid = rawSize >= parsed.snapshot.cutlineOffset + && parsed.segments.every(segment => boundaryMatches(fd!, segment)); + } catch { + valid = false; + } finally { + if (fd !== undefined) closeSync(fd); + } + boundaryValidationCache = { key: revisionKey, valid }; + if (!valid) return null; + } + } + return parsed.snapshot; + } catch { + return null; + } +} + +function cutoffDateKey(now: number): string { + const cutoff = new Date(now); + cutoff.setHours(0, 0, 0, 0); + cutoff.setDate(cutoff.getDate() - ROLLUP_MIN_AGE_DAYS); + return localDateKey(cutoff.getTime()); +} + +async function eligibleCutline(fd: number, size: number, fromOffset: number, now: number): Promise { + const cutoff = cutoffDateKey(now); + let pending: Buffer = Buffer.alloc(0); + let pendingStart = fromOffset; + let eligible = fromOffset; + for (let position = fromOffset; position < size;) { + const length = Math.min(READ_CHUNK_BYTES, size - position); + const chunk = readExactly(fd, length, position); + if (!chunk) break; + pending = pending.length === 0 ? chunk : Buffer.concat([pending, chunk]); + let cursor = 0; + for (;;) { + const newline = pending.indexOf(0x0a, cursor); + if (newline < 0) break; + const line = pending.subarray(cursor, newline).toString("utf8").replace(/\r$/, ""); + if (line.trim()) { + // A COMPLETE (newline-terminated) row that cannot carry usage data — + // unparseable JSON, a non-object, no string requestId, or an unusable + // timestamp — must not stall the cutline forever: the fold parse skips + // it identically (parseUsageRange requires a string requestId), so + // advancing past it loses nothing. Only a REAL usage row whose usable + // timestamp is at or past the cutoff stops eligibility (still too recent). + let parsed: unknown; + try { parsed = JSON.parse(line); } catch { parsed = null; } + if (isObject(parsed) && typeof parsed.requestId === "string" + && usableTimestamp(parsed.timestamp) + && localDateKey(parsed.timestamp) >= cutoff) return pendingStart + cursor; + } + eligible = pendingStart + newline + 1; + cursor = newline + 1; + // Segment cap: one fold segment materializes at most this raw span. + // The caller loops, so a large backlog still catches up — in bounded slices. + if (eligible - fromOffset >= foldSegmentCapBytes) return eligible; + } + if (cursor > 0) { + pending = pending.subarray(cursor); + pendingStart += cursor; + } + position += length; + // Yield between chunks so a first scan over a large existing log cannot + // monopolize the event loop (review thread: yield while locating the cutline). + await new Promise(resolve => setTimeout(resolve, 0)); + } + return eligible; +} + +async function parseUsageRange(fd: number, fromOffset: number, toOffset: number): Promise { + const entries: PersistedUsageEntry[] = []; + let pending: Buffer = Buffer.alloc(0); + let parsedLines = 0; + for (let position = fromOffset; position < toOffset;) { + const length = Math.min(READ_CHUNK_BYTES, toOffset - position); + const chunk = readExactly(fd, length, position); + if (!chunk) break; + pending = pending.length === 0 ? chunk : Buffer.concat([pending, chunk]); + let cursor = 0; + for (;;) { + const newline = pending.indexOf(0x0a, cursor); + if (newline < 0) break; + const line = pending.subarray(cursor, newline).toString("utf8").replace(/\r$/, ""); + cursor = newline + 1; + if (!line.trim()) continue; + try { + const parsed = JSON.parse(line) as PersistedUsageEntry; + if (parsed && typeof parsed === "object" && typeof parsed.requestId === "string") { + entries.push(normalizeUsageEntry(parsed)); + } + } catch { /* complete malformed rows are skipped like usage/log.ts */ } + parsedLines += 1; + if (parsedLines % 1_000 === 0) await new Promise(resolve => setTimeout(resolve, 0)); + } + if (cursor > 0) pending = pending.subarray(cursor); + position += length; + } + return entries; +} + +function dayAccumulator(): DayAccumulator { + return { + statusCounts: blankStatuses(), attemptCount: 0, tokens: blankTokens(), estimatedCostUsd: 0, + pricedRequests: 0, unpricedRequests: 0, unmeteredRequests: 0, + }; +} + +function attributionAccumulator(resolvedModel?: string): AttributionAccumulator { + return { + ...(resolvedModel ? { firstResolvedModel: resolvedModel } : {}), + attemptCount: 0, statusesByRequest: new Map(), inputTokens: 0, outputTokens: 0, + totalTokens: 0, estimatedCostUsd: 0, + }; +} + +function providerAccumulator(): ProviderAccumulator { + return { attemptCount: 0, statusesByRequest: new Map(), totalTokens: 0, estimatedCostUsd: 0 }; +} + +function appendStatus(map: Map, requestId: string, status: UsageStatus): void { + const statuses = map.get(requestId) ?? []; + statuses.push(status); + map.set(requestId, statuses); +} + +function modelIdentity(attribution: UsageAttribution): { provider: string; model: string; resolvedModel?: string } { + return { + provider: baseProviderLabel(attribution.provider), + model: attribution.model, + ...(attribution.resolvedModel ? { resolvedModel: attribution.resolvedModel } : {}), + }; +} + +function accumulateEntries(entries: PersistedUsageEntry[], seg: number, attemptId: string): { + rows: RollupGroupRow[]; attributionSinceMs: number | null; oldestTimestampMs: number | null; +} { + const days = new Map(); + const models = new Map(); + const providers = new Map(); + const keys = new Map(); + let attributionSinceMs: number | null = null; + let oldestTimestampMs: number | null = null; + for (const entry of entries) { + if (!usableTimestamp(entry.timestamp)) continue; + oldestTimestampMs = oldestTimestampMs === null ? entry.timestamp : Math.min(oldestTimestampMs, entry.timestamp); + const date = localDateKey(entry.timestamp); + const surface = surfaceKey(entry); + const dayKey = `${date}\0${surface}`; + const day = days.get(dayKey) ?? dayAccumulator(); + bumpStatus(day.statusCounts, entry.usageStatus); + day.attemptCount += entry.attempts?.length ?? 1; + addTokenSums(day.tokens, entryTokenSums(entry)); + const tier = serviceTierContext(entry); + const estimate = entry.attempts?.length + ? estimateComboCost(entry.attempts, undefined, tier) + : estimateRequestCost({ provider: entry.provider, model: entry.model, usage: entry.usage, usageStatus: entry.usageStatus, serviceTier: tier }); + if (entry.usageStatus === "unreported" || entry.usageStatus === "unsupported" + || (!entry.usage && !entry.attempts?.length)) day.unmeteredRequests += 1; + else if (!estimate) day.unpricedRequests += 1; + else { day.pricedRequests += 1; day.estimatedCostUsd += estimate.cost.total; } + days.set(dayKey, day); + + const attributions = usageAttributions(entry); + for (const attribution of attributions) { + const identity = modelIdentity(attribution); + const modelKey = `${date}\0${surface}\0${identity.provider}\0${identity.model}`; + const model = models.get(modelKey) ?? attributionAccumulator(identity.resolvedModel); + model.attemptCount += 1; + appendStatus(model.statusesByRequest, attribution.requestId, attribution.usageStatus); + if (attribution.usage) { + model.inputTokens += attribution.usage.inputTokens; + model.outputTokens += attribution.usage.outputTokens; + model.totalTokens += usageDisplayTotalTokens(attribution.usage, attribution.totalTokens) ?? 0; + } + models.set(modelKey, model); + + const providerKey = `${date}\0${surface}\0${identity.provider}`; + const provider = providers.get(providerKey) ?? providerAccumulator(); + provider.attemptCount += 1; + appendStatus(provider.statusesByRequest, attribution.requestId, attribution.usageStatus); + provider.totalTokens += usageDisplayTotalTokens(attribution.usage, attribution.totalTokens) ?? 0; + providers.set(providerKey, provider); + } + if (estimate?.attempts) { + for (const attemptEstimate of estimate.attempts) { + const identity = usageModelIdentity(attemptEstimate.provider, attemptEstimate.model); + const provider = baseProviderLabel(attemptEstimate.provider); + const model = models.get(`${date}\0${surface}\0${provider}\0${identity.model}`); + if (model) model.estimatedCostUsd += attemptEstimate.cost.total; + const providerRow = providers.get(`${date}\0${surface}\0${provider}`); + if (providerRow) providerRow.estimatedCostUsd += attemptEstimate.cost.total; + } + } else if (estimate) { + const identity = usageModelIdentity(entry.provider, entry.model, entry.resolvedModel); + const provider = baseProviderLabel(entry.provider); + const model = models.get(`${date}\0${surface}\0${provider}\0${identity.model}`); + if (model) model.estimatedCostUsd += estimate.cost.total; + const providerRow = providers.get(`${date}\0${surface}\0${provider}`); + if (providerRow) providerRow.estimatedCostUsd += estimate.cost.total; + } + + if (entry.admissionKind) { + attributionSinceMs = attributionSinceMs === null ? entry.timestamp : Math.min(attributionSinceMs, entry.timestamp); + const key = `${date}\0${entry.admissionKind}\0${entry.apiKeyId ?? ""}`; + const bucket = keys.get(key) ?? { requests: 0, requestsWithTimestamp: 0, lastUsedAtMs: null }; + bucket.requests += 1; + bucket.requestsWithTimestamp += 1; + bucket.lastUsedAtMs = bucket.lastUsedAtMs === null ? entry.timestamp : Math.max(bucket.lastUsedAtMs, entry.timestamp); + keys.set(key, bucket); + } + } + + const rows: RollupGroupRow[] = []; + for (const [key, value] of [...days].sort(([a], [b]) => a.localeCompare(b))) { + const [date, surface] = key.split("\0") as [string, RollupSurfaceKey]; + rows.push({ kind: "day", seg, attemptId, date, surface, ...value }); + } + for (const [key, value] of [...models].sort(([a], [b]) => a.localeCompare(b))) { + const [date, surface, provider, model] = key.split("\0") as [string, RollupSurfaceKey, string, string]; + const foldedStatusCounts = blankStatuses(); + for (const statuses of value.statusesByRequest.values()) bumpStatus(foldedStatusCounts, foldAttributionStatuses(statuses)); + rows.push({ + kind: "model", seg, attemptId, date, surface, provider, model, + ...(value.firstResolvedModel ? { resolvedModel: value.firstResolvedModel } : {}), + requests: value.statusesByRequest.size, attemptCount: value.attemptCount, foldedStatusCounts, + tokens: { inputTokens: value.inputTokens, outputTokens: value.outputTokens, totalTokens: value.totalTokens }, + estimatedCostUsd: value.estimatedCostUsd, + }); + } + for (const [key, value] of [...providers].sort(([a], [b]) => a.localeCompare(b))) { + const [date, surface, provider] = key.split("\0") as [string, RollupSurfaceKey, string]; + const foldedStatusCounts = blankStatuses(); + for (const statuses of value.statusesByRequest.values()) bumpStatus(foldedStatusCounts, foldAttributionStatuses(statuses)); + rows.push({ + kind: "provider", seg, attemptId, date, surface, provider, + requests: value.statusesByRequest.size, attemptCount: value.attemptCount, foldedStatusCounts, + totalTokens: value.totalTokens, estimatedCostUsd: value.estimatedCostUsd, + }); + } + for (const [key, value] of [...keys].sort(([a], [b]) => a.localeCompare(b))) { + const [date, admissionKind, apiKeyId] = key.split("\0") as [string, RollupKeyRow["admissionKind"], string]; + rows.push({ kind: "key", seg, attemptId, date, admissionKind, ...(apiKeyId ? { apiKeyId } : {}), ...value }); + } + return { rows, attributionSinceMs, oldestTimestampMs }; +} + +function ensureOwnedFiles(): void { + const dir = getConfigDir(); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + recordOwnedConfigPath(dir, rollupPath()); + recordOwnedConfigPath(dir, rollupMetaPath()); + if (!existsSync(rollupPath())) writeFileSync(rollupPath(), "", { mode: 0o600 }); + try { chmodSync(rollupPath(), 0o600); } catch { /* best-effort */ } +} + +function repairAppendBoundary(fd: number): void { + const size = Number(fstatSync(fd).size); + if (size === 0) return; + const lastByte = readExactly(fd, 1, size - 1); + if (lastByte?.[0] === 0x0a) return; + for (let end = size; end > 0;) { + const start = Math.max(0, end - READ_CHUNK_BYTES); + const chunk = readExactly(fd, end - start, start); + const lastNewline = chunk?.lastIndexOf(0x0a) ?? -1; + if (lastNewline >= 0) { + ftruncateSync(fd, start + lastNewline + 1); + return; + } + end = start; + } + ftruncateSync(fd, 0); +} + +function boundaryDigest(fd: number, fromOffset: number, toOffset: number): string { + const length = Math.min(BOUNDARY_DIGEST_BYTES, toOffset - fromOffset); + const bytes = readExactly(fd, length, toOffset - length); + if (!bytes) throw new Error("usage boundary became unreadable"); + return sha256(bytes); +} + +function boundaryMatches(fd: number, segment: ParsedSegment): boolean { + try { + return boundaryDigest(fd, segment.commit.seg, segment.commit.toOffset) === segment.commit.boundaryDigest; + } catch { + return false; + } +} + +function removeDerivedFiles(): void { + for (const path of [rollupPath(), rollupMetaPath()]) { + try { unlinkSync(path); } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } + } +} + +function writeMeta(meta: RollupMeta): void { + const path = rollupMetaPath(); + const temporary = `${path}.tmp-${process.pid}-${randomBytes(6).toString("hex")}`; + let fd: number | undefined; + try { + fd = openSync(temporary, "wx", 0o600); + const bytes = Buffer.from(`${JSON.stringify(meta, null, 2)}\n`); + writeSync(fd, bytes, 0, bytes.length, null); + fsyncSync(fd); + closeSync(fd); + fd = undefined; + renameAtomicFile(temporary, path); + try { chmodSync(path, 0o600); } catch { /* best-effort */ } + try { + const dirFd = openSync(getConfigDir(), constants.O_RDONLY); + try { fsyncSync(dirFd); } finally { closeSync(dirFd); } + } catch { /* directory fsync is unavailable on some platforms */ } + } finally { + if (fd !== undefined) closeSync(fd); + try { unlinkSync(temporary); } catch { /* renamed or never created */ } + } +} + +export async function foldUsagePrefix(): Promise { + const now = Date.now(); + const currentLineage = lineageKey(); + if (!currentLineage) return; + ensureOwnedFiles(); + const fingerprint = currentRollupPriceFingerprint(); + let meta = readMeta(); + if (!meta || meta.version !== 1 || meta.lineageKey !== currentLineage || meta.priceFingerprint !== fingerprint) { + removeDerivedFiles(); + ensureOwnedFiles(); + meta = null; + } + let parsed = parseRollup(currentLineage); + let rawFd: number | undefined; + try { + rawFd = openSync(usageLogPath(), "r"); + const rawSize = Number(fstatSync(rawFd).size); + const lastSegment = parsed.segments.at(-1); + if (rawSize < parsed.snapshot.cutlineOffset || (lastSegment && !boundaryMatches(rawFd, lastSegment))) { + closeSync(rawFd); + rawFd = undefined; + removeDerivedFiles(); + ensureOwnedFiles(); + parsed = parseRollup(currentLineage); + rawFd = openSync(usageLogPath(), "r"); + } + const size = Number(fstatSync(rawFd).size); + let fromOffset = parsed.snapshot.cutlineOffset; + // Segment loop: each iteration folds at most `foldSegmentCapBytes` of raw + // history (the cutline stops at the cap), so a large backlog catches up in + // bounded-memory slices instead of one whole-prefix materialization. + for (;;) { + const toOffset = await eligibleCutline(rawFd, size, fromOffset, now); + if (toOffset <= fromOffset) break; + const entries = await parseUsageRange(rawFd, fromOffset, toOffset); + const foldedAt = Date.now(); + const attemptId = `${foldedAt}-${randomBytes(8).toString("hex")}`; + const aggregate = accumulateEntries(entries, fromOffset, attemptId); + const payload = aggregate.rows.map(row => `${JSON.stringify(row)}\n`).join(""); + const commit: RollupCommitRow = { + kind: "commit", lineageKey: currentLineage, seg: fromOffset, toOffset, attemptId, + rowCount: aggregate.rows.length, payloadDigest: sha256(payload), + boundaryDigest: boundaryDigest(rawFd, fromOffset, toOffset), + attributionSinceMs: aggregate.attributionSinceMs, + oldestTimestampMs: aggregate.oldestTimestampMs, + foldedAt, + }; + const rollupFd = openSync(rollupPath(), constants.O_RDWR | constants.O_CREAT, 0o600); + try { + repairAppendBoundary(rollupFd); + const appendAt = Number(fstatSync(rollupFd).size); + const bytes = Buffer.from(`${payload}${JSON.stringify(commit)}\n`); + let written = 0; + while (written < bytes.length) written += writeSync(rollupFd, bytes, written, bytes.length - written, appendAt + written); + fsyncSync(rollupFd); + } finally { + closeSync(rollupFd); + } + fromOffset = toOffset; + } + writeMeta({ + version: 1, + lineageKey: currentLineage, + priceFingerprint: fingerprint, + lastFoldAttemptAt: now, + updatedAt: Date.now(), + }); + } finally { + if (rawFd !== undefined) closeSync(rawFd); + } +} + +export async function ensureRollupCurrent(): Promise { + if (rollupFlight) return rollupFlight; + const task = (async () => { + try { + const meta = readMeta(); + const currentLineage = lineageKey(); + const fingerprint = currentRollupPriceFingerprint(); + if (meta && currentLineage && meta.version === 1 && meta.lineageKey === currentLineage + && meta.priceFingerprint === fingerprint + && Date.now() - meta.lastFoldAttemptAt < ROLLUP_ATTEMPT_THROTTLE_MS) return; + await foldUsagePrefix(); + } catch { + // The rollup is a derived cache; raw-tail-only service is always safe. + } + })(); + rollupFlight = task; + try { await task; } finally { if (rollupFlight === task) rollupFlight = null; } +} + +export function resetRollupForTests(): void { + rollupFlight = null; + boundaryValidationCache = null; +} diff --git a/src/usage/summary.ts b/src/usage/summary.ts index 8337505ed..532d64e66 100644 --- a/src/usage/summary.ts +++ b/src/usage/summary.ts @@ -3,6 +3,7 @@ import { canonicalAntigravityUsageModel } from "../providers/antigravity-models" import { usageDisplayTotalTokens } from "./totals"; import type { PersistedUsageEntry, UsageStatus } from "./log"; import { estimateComboCost, estimateRequestCost, serviceTierContext } from "./cost"; +import type { RollupDayRow, RollupModelRow, RollupProviderRow, RollupSurfaceKey } from "./rollup"; export type UsageRange = "7d" | "30d" | "all"; export type UsageSurface = "all" | "codex" | "claude" | "grok"; @@ -90,6 +91,13 @@ export interface UsageSummary { providers: UsageProvider[]; } +export interface RollupContribution { + days: RollupDayRow[]; + models: RollupModelRow[]; + providers: RollupProviderRow[]; + oldestTimestampMs: number | null; +} + const DAY_MS = 86_400_000; export const MAX_USAGE_MODEL_BREAKDOWN_ROWS = 256; @@ -119,7 +127,7 @@ function rangeWindow(range: UsageRange, now: number): { since: number | null; da return { since: null, days: 0 }; } -function localDateKey(ts: number): string { +export function localDateKey(ts: number): string { const d = new Date(ts); const y = d.getFullYear(); const m = String(d.getMonth() + 1).padStart(2, "0"); @@ -127,13 +135,41 @@ function localDateKey(ts: number): string { return `${y}-${m}-${day}`; } -function dayCountForAllRange(entries: PersistedUsageEntry[], now: number): number { - if (entries.length === 0) return 1; - const oldest = entries.reduce((min, e) => Math.min(min, e.timestamp), entries[0].timestamp); +function dayCountForAllRange(entries: PersistedUsageEntry[], now: number, rollupOldest: number | null = null): number { + const tailOldest = entries.length === 0 + ? null + : entries.reduce((min, entry) => Math.min(min, entry.timestamp), entries[0].timestamp); + const oldest = tailOldest === null ? rollupOldest + : rollupOldest === null ? tailOldest + : Math.min(tailOldest, rollupOldest); + if (oldest === null) return 1; const days = Math.ceil((now - oldest) / DAY_MS) + 1; return Math.max(1, days); } +function surfaceKeyMatches(surfaceKey: RollupSurfaceKey, surface: UsageSurface): boolean { + if (surface === "claude") return surfaceKey === "claude" || surfaceKey === "claude-desktop"; + if (surface === "grok") return surfaceKey === "grok"; + if (surface === "codex") return surfaceKey === "codex"; + return true; +} + +function localDayStart(date: string): number | null { + const match = /^(\d{4})-(\d{2})-(\d{2})$/.exec(date); + if (!match) return null; + const value = new Date(Number(match[1]), Number(match[2]) - 1, Number(match[3])).getTime(); + return Number.isNaN(value) ? null : value; +} + +function rollupDateOverlapsRange(date: string, since: number | null, now: number): boolean { + if (since === null) return true; + const start = localDayStart(date); + if (start === null) return false; + const end = new Date(start); + end.setDate(end.getDate() + 1); + return start <= now && end.getTime() > since; +} + function blankTotals(): UsageSummaryTotals { return { requests: 0, @@ -162,7 +198,7 @@ function isMeasuredStatus(status: UsageStatus): boolean { return status === "reported" || status === "estimated"; } -interface UsageAttribution { +export interface UsageAttribution { requestId: string; provider: string; model: string; @@ -178,7 +214,7 @@ interface UsageAttribution { * Google Antigravity collapses wire/compat/suffix ids to picker/call base models so * historical effort-variant logs merge with current base-model invocations. */ -function usageModelIdentity( +export function usageModelIdentity( provider: string, model: string, resolvedModel?: string, @@ -207,7 +243,7 @@ function antigravityUsageModel(provider: string, model: string): string { return canonicalAntigravityUsageModel(model); } -function usageAttributions(entry: PersistedUsageEntry): UsageAttribution[] { +export function usageAttributions(entry: PersistedUsageEntry): UsageAttribution[] { if (!entry.attempts?.length) { return [{ requestId: entry.requestId, @@ -228,7 +264,7 @@ function usageAttributions(entry: PersistedUsageEntry): UsageAttribution[] { })); } -function foldAttributionStatuses(statuses: readonly UsageStatus[]): UsageStatus { +export function foldAttributionStatuses(statuses: readonly UsageStatus[]): UsageStatus { if (statuses.length > 0 && statuses.every(status => status === "unsupported")) { return "unsupported"; } @@ -298,14 +334,23 @@ function addEstimatedCost( totals.estimatedCostUsd += estimate.cost.total; } -function buildDayGrid(range: UsageRange, since: number | null, now: number, entries: PersistedUsageEntry[]): UsageDay[] { +function buildDayGrid( + range: UsageRange, + since: number | null, + now: number, + entries: PersistedUsageEntry[], + rollupDays: readonly RollupDayRow[] = [], + rollupModels: readonly RollupModelRow[] = [], + rollupOldest: number | null = null, +): UsageDay[] { const window = rangeWindow(range, now); - const days = range === "all" ? dayCountForAllRange(entries, now) : window.days; + const days = range === "all" ? dayCountForAllRange(entries, now, rollupOldest) : window.days; const grid = new Map(); // Per-day model breakdown accumulator, keyed by day then provider/model, so the 7d bar chart can // render a per-model stacked bar with a hover tooltip without a second pass over the entries. const dayModels = new Map>(); const dayModelRequests = new Map>(); + const dayModelSeedRequests = new Map(); const bumpDayModel = (dayKey: string, attribution: UsageAttribution): void => { let models = dayModels.get(dayKey); if (!models) { models = new Map(); dayModels.set(dayKey, models); } @@ -320,7 +365,7 @@ function buildDayGrid(range: UsageRange, since: number | null, now: number, entr let requests = dayModelRequests.get(requestKey); if (!requests) { requests = new Set(); dayModelRequests.set(requestKey, requests); } requests.add(attribution.requestId); - m.requests = requests.size; + m.requests = (dayModelSeedRequests.get(requestKey) ?? 0) + requests.size; m.attemptCount += 1; m.totalTokens += usageDisplayTotalTokens(attribution.usage, attribution.totalTokens) ?? 0; }; @@ -328,6 +373,28 @@ function buildDayGrid(range: UsageRange, since: number | null, now: number, entr const key = localDateKey(now - i * DAY_MS); grid.set(key, { date: key, requests: 0, measuredRequests: 0, reportedRequests: 0, totalTokens: 0, models: [] }); } + for (const row of rollupDays) { + const day = grid.get(row.date) ?? { date: row.date, requests: 0, measuredRequests: 0, reportedRequests: 0, totalTokens: 0, models: [] }; + day.requests += row.statusCounts.reported + row.statusCounts.unreported + + row.statusCounts.unsupported + row.statusCounts.estimated; + day.measuredRequests += row.statusCounts.reported + row.statusCounts.estimated; + day.reportedRequests += row.statusCounts.reported; + day.totalTokens += row.tokens.totalTokens; + grid.set(row.date, day); + } + for (const row of rollupModels) { + let models = dayModels.get(row.date); + if (!models) { models = new Map(); dayModels.set(row.date, models); } + const key = usageModelKey(row.provider, row.model); + const current = models.get(key) ?? { + model: row.model, provider: row.provider, requests: 0, attemptCount: 0, totalTokens: 0, + }; + current.requests += row.requests; + current.attemptCount += row.attemptCount; + current.totalTokens += row.tokens.totalTokens; + models.set(key, current); + dayModelSeedRequests.set(`${row.date}\0${key}`, current.requests); + } for (const entry of entries) { const key = localDateKey(entry.timestamp); let day = grid.get(key); @@ -349,24 +416,57 @@ function buildDayGrid(range: UsageRange, since: number | null, now: number, entr const sorted = [...models.values()].sort((a, b) => b.requests - a.requests); day.models = retainedBreakdownRows(sorted, overflow => { const requests = new Set(); + let seededRequests = 0; let attemptCount = 0; let totalTokens = 0; for (const model of overflow) { attemptCount += model.attemptCount; totalTokens += model.totalTokens; const requestKey = `${day.date}\0${usageModelKey(model.provider, model.model)}`; + seededRequests += dayModelSeedRequests.get(requestKey) ?? 0; for (const requestId of dayModelRequests.get(requestKey) ?? []) requests.add(requestId); } - return { model: "other", provider: "other", requests: requests.size, attemptCount, totalTokens }; + return { model: "other", provider: "other", requests: seededRequests + requests.size, attemptCount, totalTokens }; }); } } return out; } -function buildModels(entries: PersistedUsageEntry[], totalTokens: number): UsageModel[] { +function buildModels( + entries: PersistedUsageEntry[], + totalTokens: number, + rollupRows: readonly RollupModelRow[] = [], +): UsageModel[] { const byKey = new Map(); const statusesByKey = new Map>(); + const rollupByKey = new Map(); + for (const row of rollupRows) { + const key = usageModelKey(row.provider, row.model); + const model = byKey.get(key) ?? { + provider: row.provider, + model: row.model, + ...(row.resolvedModel ? { resolvedModel: row.resolvedModel } : {}), + requests: 0, attemptCount: 0, measuredRequests: 0, reportedRequests: 0, + estimatedRequests: 0, totalTokens: 0, inputTokens: 0, outputTokens: 0, shareRatio: 0, + }; + model.requests += row.requests; + model.attemptCount += row.attemptCount; + model.measuredRequests += row.foldedStatusCounts.reported + row.foldedStatusCounts.estimated; + model.reportedRequests += row.foldedStatusCounts.reported; + model.estimatedRequests += row.foldedStatusCounts.estimated; + model.totalTokens += row.tokens.totalTokens; + model.inputTokens += row.tokens.inputTokens; + model.outputTokens += row.tokens.outputTokens; + if (row.estimatedCostUsd !== 0) model.estimatedCostUsd = (model.estimatedCostUsd ?? 0) + row.estimatedCostUsd; + byKey.set(key, model); + const seed = rollupByKey.get(key) ?? { requests: 0, measured: 0, reported: 0, estimated: 0 }; + seed.requests += row.requests; + seed.measured += row.foldedStatusCounts.reported + row.foldedStatusCounts.estimated; + seed.reported += row.foldedStatusCounts.reported; + seed.estimated += row.foldedStatusCounts.estimated; + rollupByKey.set(key, seed); + } for (const entry of entries) { for (const attribution of usageAttributions(entry)) { const providerKey = baseProviderLabel(attribution.provider); @@ -405,7 +505,7 @@ function buildModels(entries: PersistedUsageEntry[], totalTokens: number): Usage } for (const [key, model] of byKey) { const groups = statusesByKey.get(key) ?? new Map(); - model.requests = groups.size; + model.requests += groups.size; for (const statuses of groups.values()) { const status = foldAttributionStatuses(statuses); if (isMeasuredStatus(status)) model.measuredRequests += 1; @@ -442,6 +542,7 @@ function buildModels(entries: PersistedUsageEntry[], totalTokens: number): Usage const sorted = models.sort((a, b) => b.requests - a.requests); return retainedBreakdownRows(sorted, overflow => { const statusesByRequest = new Map(); + let seededRequests = 0; const other: UsageModel = { provider: "other", model: "other", @@ -464,13 +565,20 @@ function buildModels(entries: PersistedUsageEntry[], totalTokens: number): Usage other.estimatedCostUsd = (other.estimatedCostUsd ?? 0) + model.estimatedCostUsd; } const key = usageModelKey(model.provider, model.model); + const seed = rollupByKey.get(key); + if (seed) { + seededRequests += seed.requests; + other.measuredRequests += seed.measured; + other.reportedRequests += seed.reported; + other.estimatedRequests += seed.estimated; + } for (const [requestId, statuses] of statusesByKey.get(key) ?? []) { const combined = statusesByRequest.get(requestId) ?? []; combined.push(...statuses); statusesByRequest.set(requestId, combined); } } - other.requests = statusesByRequest.size; + other.requests = seededRequests + statusesByRequest.size; for (const statuses of statusesByRequest.values()) { const status = foldAttributionStatuses(statuses); if (isMeasuredStatus(status)) other.measuredRequests += 1; @@ -482,9 +590,27 @@ function buildModels(entries: PersistedUsageEntry[], totalTokens: number): Usage }); } -function buildProviders(entries: PersistedUsageEntry[], totalTokens: number): UsageProvider[] { +function buildProviders( + entries: PersistedUsageEntry[], + totalTokens: number, + rollupRows: readonly RollupProviderRow[] = [], +): UsageProvider[] { const byKey = new Map(); const statusesByKey = new Map>(); + for (const row of rollupRows) { + const provider = byKey.get(row.provider) ?? { + provider: row.provider, requests: 0, attemptCount: 0, measuredRequests: 0, + reportedRequests: 0, estimatedRequests: 0, totalTokens: 0, shareRatio: 0, + }; + provider.requests += row.requests; + provider.attemptCount += row.attemptCount; + provider.measuredRequests += row.foldedStatusCounts.reported + row.foldedStatusCounts.estimated; + provider.reportedRequests += row.foldedStatusCounts.reported; + provider.estimatedRequests += row.foldedStatusCounts.estimated; + provider.totalTokens += row.totalTokens; + if (row.estimatedCostUsd !== 0) provider.estimatedCostUsd = (provider.estimatedCostUsd ?? 0) + row.estimatedCostUsd; + byKey.set(row.provider, provider); + } for (const entry of entries) { for (const attribution of usageAttributions(entry)) { const providerKey = baseProviderLabel(attribution.provider); @@ -515,7 +641,7 @@ function buildProviders(entries: PersistedUsageEntry[], totalTokens: number): Us } for (const [key, provider] of byKey) { const groups = statusesByKey.get(key) ?? new Map(); - provider.requests = groups.size; + provider.requests += groups.size; for (const statuses of groups.values()) { const status = foldAttributionStatuses(statuses); if (isMeasuredStatus(status)) provider.measuredRequests += 1; @@ -552,6 +678,7 @@ export function summarizeUsage( range: UsageRange, now: number, surface: UsageSurface = "all", + rollup?: RollupContribution, ): UsageSummary { const { since } = rangeWindow(range, now); const filteredEntries = entries.filter(entry => { @@ -564,7 +691,34 @@ export function summarizeUsage( if (surface === "codex") return entry.surface === undefined; return true; }); + const rollupDays = rollup?.days.filter(row => + surfaceKeyMatches(row.surface, surface) && rollupDateOverlapsRange(row.date, since, now)) ?? []; + const rollupModels = rollup?.models.filter(row => + surfaceKeyMatches(row.surface, surface) && rollupDateOverlapsRange(row.date, since, now)) ?? []; + const rollupProviders = rollup?.providers.filter(row => + surfaceKeyMatches(row.surface, surface) && rollupDateOverlapsRange(row.date, since, now)) ?? []; const totals = blankTotals(); + for (const row of rollupDays) { + totals.requests += row.statusCounts.reported + row.statusCounts.unreported + + row.statusCounts.unsupported + row.statusCounts.estimated; + totals.attemptCount += row.attemptCount; + totals.measuredRequests += row.statusCounts.reported + row.statusCounts.estimated; + totals.reportedRequests += row.statusCounts.reported; + totals.unreportedRequests += row.statusCounts.unreported; + totals.unsupportedRequests += row.statusCounts.unsupported; + totals.estimatedRequests += row.statusCounts.estimated; + totals.inputTokens += row.tokens.inputTokens; + totals.outputTokens += row.tokens.outputTokens; + totals.cachedInputTokens += row.tokens.cacheReadInputTokens; + totals.cacheReadInputTokens += row.tokens.cacheReadInputTokens; + totals.cacheCreationInputTokens += row.tokens.cacheCreationInputTokens; + totals.reasoningOutputTokens += row.tokens.reasoningOutputTokens; + totals.totalTokens += row.tokens.totalTokens; + totals.estimatedCostUsd += row.estimatedCostUsd; + totals.pricedRequests += row.pricedRequests; + totals.unpricedRequests += row.unpricedRequests; + totals.unmeteredRequests += row.unmeteredRequests; + } for (const entry of filteredEntries) { bumpStatus(totals, entry.usageStatus); totals.attemptCount += entry.attempts?.length ?? 1; @@ -578,8 +732,11 @@ export function summarizeUsage( since, generatedAt: now, summary: totals, - days: buildDayGrid(range, since, now, filteredEntries), - models: buildModels(filteredEntries, totals.totalTokens), - providers: buildProviders(filteredEntries, totals.totalTokens), + days: buildDayGrid( + range, since, now, filteredEntries, rollupDays, rollupModels, + range === "all" ? rollup?.oldestTimestampMs ?? null : null, + ), + models: buildModels(filteredEntries, totals.totalTokens, rollupModels), + providers: buildProviders(filteredEntries, totals.totalTokens, rollupProviders), }; } diff --git a/tests/api-usage.test.ts b/tests/api-usage.test.ts index ae7e32cd7..cdd3f1ed4 100644 --- a/tests/api-usage.test.ts +++ b/tests/api-usage.test.ts @@ -108,7 +108,7 @@ describe("GET /api/usage", () => { test("usage route cache preserves truncation metadata and invalidates when configured byte limit changes", async () => { writeFixture(Date.now()); - saveConfig({ ...baseConfig(), managementUsageMaxReadBytes: 256 }); + saveConfig({ ...baseConfig(), managementUsageMaxReadBytes: 256, usageRollupEnabled: false }); const server = startServer(0); try { const first = await fetch(new URL("/api/usage?range=all", server.url)).then(response => response.json()); @@ -128,6 +128,7 @@ describe("GET /api/usage", () => { test("reuses only a compact summary for an unchanged revision", async () => { writeFixture(Date.now()); + saveConfig({ ...baseConfig(), usageRollupEnabled: false }); const server = startServer(0); try { const first = await fetch(new URL("/api/usage?range=30d", server.url)).then(res => res.json()); diff --git a/tests/helpers/usage-rollup-fixtures.ts b/tests/helpers/usage-rollup-fixtures.ts new file mode 100644 index 000000000..1bc115bb9 --- /dev/null +++ b/tests/helpers/usage-rollup-fixtures.ts @@ -0,0 +1,88 @@ +import type { PersistedUsageEntry, PersistedUsageAttempt, UsageStatus } from "../../src/usage/log"; + +export interface RandomizedUsageRollupFixture { + entries: PersistedUsageEntry[]; + foldBoundaryTimestamp: number; + duplicateRequestId: string; +} + +function mulberry32(seed: number): () => number { + let state = seed >>> 0; + return () => { + state += 0x6d2b79f5; + let value = state; + value = Math.imul(value ^ value >>> 15, value | 1); + value ^= value + Math.imul(value ^ value >>> 7, value | 61); + return ((value ^ value >>> 14) >>> 0) / 4_294_967_296; + }; +} + +function attempt( + ordinal: number, + provider: string, + model: string, + usageStatus: UsageStatus, + inputTokens: number, + outputTokens: number, +): PersistedUsageAttempt { + return { + ordinal, provider, model, adapter: "openai-chat", status: 200, durationMs: ordinal, + sendCount: 1, recoveryKinds: [], usageStatus, + usage: { inputTokens, outputTokens }, totalTokens: inputTokens + outputTokens, + }; +} + +/** + * Deterministic randomized corpus for the phase-020 fold-plus-tail property test. + * It deliberately includes combo attempts, out-of-order timestamps, entries on + * the fold-boundary day, and a cross-day duplicate-id pair for the documented + * out-of-domain assertion. Phase 020 filters that pair for its in-domain run. + */ +export function generateRandomizedUsageRollupFixture( + seed = 0x260804, + now = Date.UTC(2026, 7, 4, 12), + count = 80, +): RandomizedUsageRollupFixture { + const random = mulberry32(seed); + const dayMs = 86_400_000; + const foldBoundaryTimestamp = now - 9 * dayMs; + const providers = ["openai", "anthropic", "kiro"] as const; + const models = ["gpt-5.5", "claude-fable-5", "claude-sonnet-4-6"] as const; + const statuses: UsageStatus[] = ["reported", "estimated", "unreported", "unsupported"]; + const entries: PersistedUsageEntry[] = []; + for (let index = 0; index < count; index += 1) { + const dayOffset = Math.floor(random() * 18); + const timestamp = index % 11 === 0 + ? foldBoundaryTimestamp + (index % 2) * 60_000 + : now - dayOffset * dayMs - Math.floor(random() * dayMs); + const provider = providers[Math.floor(random() * providers.length)]!; + const model = models[Math.floor(random() * models.length)]!; + const status = statuses[Math.floor(random() * statuses.length)]!; + const inputTokens = 1 + Math.floor(random() * 500); + const outputTokens = 1 + Math.floor(random() * 100); + const combo = index % 7 === 0; + entries.push({ + requestId: `property-${seed}-${index}`, + timestamp, + provider, + model, + ...(index % 4 === 0 ? { surface: "claude-desktop" as const } : {}), + ...(index % 3 === 0 ? { admissionKind: "configured" as const, apiKeyId: `key-${index % 5}` } : {}), + status: 200, + durationMs: 10, + usageStatus: status, + ...(status === "reported" || status === "estimated" + ? { usage: { inputTokens, outputTokens, ...(status === "estimated" ? { estimated: true } : {}) }, totalTokens: inputTokens + outputTokens } + : {}), + ...(combo ? { attempts: [ + attempt(1, "openai", "gpt-5.5", "reported", inputTokens, outputTokens), + attempt(2, "anthropic", "claude-fable-5", "estimated", inputTokens + 1, outputTokens + 1), + ] } : {}), + }); + } + const duplicateRequestId = `property-${seed}-duplicate`; + entries.push({ ...entries[0]!, requestId: duplicateRequestId, timestamp: now - 15 * dayMs }); + entries.push({ ...entries[1]!, requestId: duplicateRequestId, timestamp: now - 12 * dayMs }); + entries.sort(() => random() - 0.5); + return { entries, foldBoundaryTimestamp, duplicateRequestId }; +} diff --git a/tests/usage-rollup-merge.test.ts b/tests/usage-rollup-merge.test.ts new file mode 100644 index 000000000..3e47c2f07 --- /dev/null +++ b/tests/usage-rollup-merge.test.ts @@ -0,0 +1,350 @@ +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { appendFileSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getConfigDir, saveConfig } from "../src/config"; +import { startServer } from "../src/server"; +import { clearApiKeyUsageCacheForTests, readApiKeyUsageRollup, rollupApiKeyUsage } from "../src/server/management/api-key-usage"; +import { resetUsageSummaryCacheForTests } from "../src/server/management/usage-summary-cache"; +import { readUsageSnapshotForManagement, resetUsageReadCacheForTests, usageLogPath, usageReadCacheStatsForTests, type PersistedUsageEntry } from "../src/usage/log"; +import { + foldUsagePrefix, + readRollupSnapshot, + resetRollupForTests, + type RollupDayRow, + type RollupModelRow, + type RollupProviderRow, + type RollupSurfaceKey, +} from "../src/usage/rollup"; +import { localDateKey, summarizeUsage, type RollupContribution, type UsageSummary } from "../src/usage/summary"; +import type { OcxConfig } from "../src/types"; +import { generateRandomizedUsageRollupFixture } from "./helpers/usage-rollup-fixtures"; +import { installIsolatedCodexHome, type IsolatedCodexHome } from "./helpers/isolated-codex-home"; +import { managementFetch } from "./helpers/management-auth"; + +const DAY_MS = 86_400_000; +const FIXED_NOW = Date.UTC(2026, 7, 4, 12); + +let testDir = ""; +let previousHome: string | undefined; +let previousTimezone: string | undefined; +let isolatedCodexHome: IsolatedCodexHome | null = null; +let clock: ReturnType; + +interface UsageRouteBody { + summary: { requests: number }; + historyTruncated: boolean; + truncatedPrefixBytes: number; +} + +function baseConfig(overrides: Partial = {}): OcxConfig { + return { + port: 0, + hostname: "127.0.0.1", + defaultProvider: "openai", + providers: { + openai: { adapter: "openai-responses", baseUrl: "https://api.openai.com/v1", authMode: "forward" }, + }, + ...overrides, + } as OcxConfig; +} + +function entry(overrides: Partial & { requestId: string; timestamp: number }): PersistedUsageEntry { + return { + provider: "openai", model: "gpt-5.5", status: 200, durationMs: 1, + usageStatus: "reported", usage: { inputTokens: 10, outputTokens: 2 }, totalTokens: 12, + ...overrides, + }; +} + +function rawLine(value: PersistedUsageEntry): string { + return `${JSON.stringify(value)}\n`; +} + +function writeRaw(entries: PersistedUsageEntry[]): void { + writeFileSync(usageLogPath(), entries.map(rawLine).join(""), { mode: 0o600 }); +} + +function dayRow(date: string, surface: RollupSurfaceKey, requests = 1, tokens = 10): RollupDayRow { + return { + kind: "day", seg: 0, attemptId: "test", date, surface, + statusCounts: { reported: requests, unreported: 0, unsupported: 0, estimated: 0 }, + attemptCount: requests, + tokens: { + inputTokens: tokens - 2 * requests, outputTokens: 2 * requests, + cacheReadInputTokens: 0, cacheCreationInputTokens: 0, reasoningOutputTokens: 0, + totalTokens: tokens, + }, + estimatedCostUsd: tokens / 1_000_000, + pricedRequests: requests, unpricedRequests: 0, unmeteredRequests: 0, + }; +} + +function modelRow(date: string, surface: RollupSurfaceKey, provider = "openai", model = "gpt-5.5", requests = 1, tokens = 10): RollupModelRow { + return { + kind: "model", seg: 0, attemptId: "test", date, surface, provider, model, + requests, attemptCount: requests, + foldedStatusCounts: { reported: requests, unreported: 0, unsupported: 0, estimated: 0 }, + tokens: { inputTokens: tokens - 2 * requests, outputTokens: 2 * requests, totalTokens: tokens }, + estimatedCostUsd: tokens / 1_000_000, + }; +} + +function providerRow(date: string, surface: RollupSurfaceKey, provider = "openai", requests = 1, tokens = 10): RollupProviderRow { + return { + kind: "provider", seg: 0, attemptId: "test", date, surface, provider, + requests, attemptCount: requests, + foldedStatusCounts: { reported: requests, unreported: 0, unsupported: 0, estimated: 0 }, + totalTokens: tokens, estimatedCostUsd: tokens / 1_000_000, + }; +} + +function contribution( + days: RollupDayRow[], + models: RollupModelRow[], + providers: RollupProviderRow[], + oldestTimestampMs: number | null, +): RollupContribution { + return { days, models, providers, oldestTimestampMs }; +} + +function canonicalSummary(summary: UsageSummary): unknown { + const canonicalModels = summary.models.map(row => ({ ...row, estimatedCostUsd: row.estimatedCostUsd ?? 0 })) + .sort((a, b) => `${a.provider}/${a.model}`.localeCompare(`${b.provider}/${b.model}`)); + const canonicalProviders = summary.providers.map(row => ({ ...row, estimatedCostUsd: row.estimatedCostUsd ?? 0 })) + .sort((a, b) => a.provider.localeCompare(b.provider)); + const days = summary.days.map(day => ({ + ...day, + models: [...day.models].sort((a, b) => `${a.provider}/${a.model}`.localeCompare(`${b.provider}/${b.model}`)), + })); + return { summary: summary.summary, days, models: canonicalModels, providers: canonicalProviders }; +} + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + previousTimezone = process.env.TZ; + isolatedCodexHome = installIsolatedCodexHome("ocx-rollup-merge-codex-"); + testDir = mkdtempSync(join(tmpdir(), "ocx-rollup-merge-")); + process.env.OPENCODEX_HOME = testDir; + clock = spyOn(Date, "now").mockReturnValue(FIXED_NOW); + resetRollupForTests(); + resetUsageReadCacheForTests(); + resetUsageSummaryCacheForTests(); + clearApiKeyUsageCacheForTests(); + saveConfig(baseConfig()); +}); + +afterEach(() => { + clock.mockRestore(); + resetRollupForTests(); + resetUsageReadCacheForTests(); + resetUsageSummaryCacheForTests(); + clearApiKeyUsageCacheForTests(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousTimezone === undefined) delete process.env.TZ; + else process.env.TZ = previousTimezone; + isolatedCodexHome?.restore(); + isolatedCodexHome = null; + if (testDir) rmSync(testDir, { recursive: true, force: true }); +}); + +describe("usage rollup reader merge", () => { + test("1. randomized in-domain fold plus tail equals the full raw summary for all", async () => { + const fixture = generateRandomizedUsageRollupFixture(0x020, FIXED_NOW, 120); + const generated = fixture.entries.filter(row => row.requestId !== fixture.duplicateRequestId); + const entries = generated.map((row, index) => { + const ageDays = index < 80 ? 10 + (index * 17) % 31 : (index * 5) % 9; + const timeOfDay = (index * 37_001) % DAY_MS; + return { ...row, timestamp: FIXED_NOW - ageDays * DAY_MS - timeOfDay }; + }); + writeRaw(entries); + await foldUsagePrefix(); + const folded = readRollupSnapshot(); + expect(folded).not.toBeNull(); + expect(folded!.cutlineOffset).toBeGreaterThan(0); + expect(folded!.cutlineOffset).toBeLessThan(readFileSync(usageLogPath()).byteLength); + const tail = await readUsageSnapshotForManagement(Number.MAX_SAFE_INTEGER, folded!.cutlineOffset); + expect(tail.truncatedPrefixBytes).toBe(0); + + const merged = summarizeUsage(tail.entries, "all", FIXED_NOW, "all", { + days: folded!.days, + models: folded!.models, + providers: folded!.providers, + oldestTimestampMs: folded!.oldestTimestampMs, + }); + const full = summarizeUsage(entries, "all", FIXED_NOW); + const mergedCanonical = canonicalSummary(merged) as { summary: UsageSummary["summary"] }; + const fullCanonical = canonicalSummary(full) as { summary: UsageSummary["summary"] }; + expect(mergedCanonical.summary.estimatedCostUsd).toBeCloseTo(fullCanonical.summary.estimatedCostUsd, 9); + const mergedWithoutCost = structuredClone(mergedCanonical); + const fullWithoutCost = structuredClone(fullCanonical); + mergedWithoutCost.summary.estimatedCostUsd = 0; + fullWithoutCost.summary.estimatedCostUsd = 0; + for (const row of (mergedWithoutCost as { models: Array<{ estimatedCostUsd: number }> }).models) row.estimatedCostUsd = 0; + for (const row of (fullWithoutCost as { models: Array<{ estimatedCostUsd: number }> }).models) row.estimatedCostUsd = 0; + for (const row of (mergedWithoutCost as { providers: Array<{ estimatedCostUsd: number }> }).providers) row.estimatedCostUsd = 0; + for (const row of (fullWithoutCost as { providers: Array<{ estimatedCostUsd: number }> }).providers) row.estimatedCostUsd = 0; + expect(mergedWithoutCost).toEqual(fullWithoutCost); + for (const row of merged.models) { + const expected = full.models.find(candidate => candidate.provider === row.provider && candidate.model === row.model); + expect(row.estimatedCostUsd ?? 0).toBeCloseTo(expected?.estimatedCostUsd ?? 0, 9); + } + for (const row of merged.providers) { + const expected = full.providers.find(candidate => candidate.provider === row.provider); + expect(row.estimatedCostUsd ?? 0).toBeCloseTo(expected?.estimatedCostUsd ?? 0, 9); + } + }); + + test("2. rollup and tail rows on the same boundary date merge additively", () => { + const date = localDateKey(FIXED_NOW - 10 * DAY_MS); + const tail = entry({ + requestId: "tail-boundary", timestamp: new Date(`${date}T18:00:00`).getTime(), + usage: { inputTokens: 3, outputTokens: 2 }, totalTokens: 5, + }); + const merged = summarizeUsage([tail], "all", FIXED_NOW, "all", contribution( + [dayRow(date, "codex", 1, 10)], + [modelRow(date, "codex", "openai", "gpt-5.5", 1, 10)], + [providerRow(date, "codex", "openai", 1, 10)], + FIXED_NOW - 10 * DAY_MS, + )); + expect(merged.summary).toMatchObject({ requests: 2, attemptCount: 2, totalTokens: 15 }); + expect(merged.days.find(row => row.date === date)).toMatchObject({ requests: 2, totalTokens: 15 }); + expect(merged.days.find(row => row.date === date)?.models).toEqual([ + expect.objectContaining({ provider: "openai", model: "gpt-5.5", requests: 2, attemptCount: 2, totalTokens: 15 }), + ]); + expect(merged.models).toEqual([expect.objectContaining({ requests: 2, attemptCount: 2, totalTokens: 15 })]); + expect(merged.providers).toEqual([expect.objectContaining({ requests: 2, attemptCount: 2, totalTokens: 15 })]); + }); + + test("3. surface predicates keep codex disjoint and combine claude with claude-desktop", () => { + const date = localDateKey(FIXED_NOW - 12 * DAY_MS); + const surfaces: RollupSurfaceKey[] = ["codex", "claude", "claude-desktop", "grok"]; + const rolled = contribution( + surfaces.map(surface => dayRow(date, surface)), + surfaces.map(surface => modelRow(date, surface, surface === "codex" ? "openai" : surface, `${surface}-model`)), + surfaces.map(surface => providerRow(date, surface, surface === "codex" ? "openai" : surface)), + FIXED_NOW - 12 * DAY_MS, + ); + const claude = summarizeUsage([], "all", FIXED_NOW, "claude", rolled); + expect(claude.summary.requests).toBe(2); + expect(claude.models.map(row => row.model).sort()).toEqual(["claude-desktop-model", "claude-model"]); + const codex = summarizeUsage([], "all", FIXED_NOW, "codex", rolled); + expect(codex.summary.requests).toBe(1); + expect(codex.models.map(row => row.model)).toEqual(["codex-model"]); + const grok = summarizeUsage([], "all", FIXED_NOW, "grok", rolled); + expect(grok.summary.requests).toBe(1); + expect(grok.providers.map(row => row.provider)).toEqual(["grok"]); + }); + + test("4. seven-day summaries stay tail-only while thirty-day summaries include whole overlapping days", () => { + process.env.TZ = "America/New_York"; + const localNow = new Date(2026, 7, 4, 12).getTime(); + clock.mockReturnValue(localNow); + const since30 = localNow - 30 * DAY_MS; + const boundaryDate = localDateKey(since30); + const insideDate = localDateKey(localNow - 10 * DAY_MS); + const outsideDate = localDateKey(localNow - 31 * DAY_MS); + const rows = [dayRow(boundaryDate, "codex"), dayRow(insideDate, "codex"), dayRow(outsideDate, "codex")]; + const models = [modelRow(boundaryDate, "codex"), modelRow(insideDate, "codex"), modelRow(outsideDate, "codex")]; + const providers = [providerRow(boundaryDate, "codex"), providerRow(insideDate, "codex"), providerRow(outsideDate, "codex")]; + const recent = entry({ requestId: "recent", timestamp: localNow - DAY_MS }); + const outOfOrderTail = [recent, entry({ requestId: "tail-older", timestamp: localNow - 6 * DAY_MS })]; + const rolled = contribution(rows, models, providers, localNow - 31 * DAY_MS); + + const seven = summarizeUsage(outOfOrderTail, "7d", localNow, "all", rolled); + expect(seven.summary.requests).toBe(2); + expect(seven.days.some(day => day.date === insideDate && day.requests > 0)).toBe(false); + const thirty = summarizeUsage(outOfOrderTail, "30d", localNow, "all", rolled); + expect(thirty.summary.requests).toBe(4); + expect(thirty.days.find(day => day.date === boundaryDate)?.requests).toBe(1); + expect(thirty.days.find(day => day.date === insideDate)?.requests).toBe(1); + expect(thirty.days.some(day => day.date === outsideDate && day.requests > 0)).toBe(false); + }); + + test("5. route cache invalidates on cutline advance and flag-off keeps the legacy truncated path", async () => { + saveConfig(baseConfig({ managementUsageMaxReadBytes: 300, usageRollupEnabled: true })); + const firstOld = entry({ requestId: "old-1", timestamp: FIXED_NOW - 13 * DAY_MS }); + writeRaw([firstOld]); + await foldUsagePrefix(); + appendFileSync(usageLogPath(), rawLine(entry({ requestId: "old-2", timestamp: FIXED_NOW - 12 * DAY_MS }))); + const server = startServer(0); + try { + const beforeFold = await managementFetch(new URL("/api/usage?range=all", server.url)).then(response => response.json()) as UsageRouteBody; + expect(beforeFold.summary.requests).toBe(2); + const readsBefore = usageReadCacheStatsForTests().fullReads; + const metaPath = join(getConfigDir(), "usage-rollup-meta.json"); + const meta = JSON.parse(readFileSync(metaPath, "utf8")) as Record; + writeFileSync(metaPath, `${JSON.stringify({ ...meta, lastFoldAttemptAt: 0 })}\n`); + await foldUsagePrefix(); + const afterFold = await managementFetch(new URL("/api/usage?range=all", server.url)).then(response => response.json()) as UsageRouteBody; + expect(afterFold.summary.requests).toBe(2); + expect(afterFold.historyTruncated).toBe(false); + expect(afterFold.truncatedPrefixBytes).toBe(0); + expect(usageReadCacheStatsForTests().fullReads).toBe(readsBefore + 1); + expect(readFileSync(usageLogPath()).byteLength).toBeGreaterThan(300); + } finally { + await server.stop(true); + } + + resetUsageSummaryCacheForTests(); + resetUsageReadCacheForTests(); + const primaryDir = testDir; + const legacyDir = mkdtempSync(join(tmpdir(), "ocx-rollup-legacy-")); + process.env.OPENCODEX_HOME = legacyDir; + saveConfig(baseConfig({ managementUsageMaxReadBytes: 300, usageRollupEnabled: false })); + writeRaw([firstOld, entry({ requestId: "old-2", timestamp: FIXED_NOW - 12 * DAY_MS })]); + const legacyServer = startServer(0); + try { + const legacy = await managementFetch(new URL("/api/usage?range=all", legacyServer.url)).then(response => response.json()) as UsageRouteBody; + expect(legacy.historyTruncated).toBe(true); + expect(legacy.truncatedPrefixBytes).toBeGreaterThan(0); + } finally { + await legacyServer.stop(true); + process.env.OPENCODEX_HOME = primaryDir; + rmSync(legacyDir, { recursive: true, force: true }); + } + }); + + test("6. API-key totals and attribution merge across the fold boundary while requests7d stays tail-exact", async () => { + const entries = [ + entry({ requestId: "key-old", timestamp: FIXED_NOW - 15 * DAY_MS, admissionKind: "configured", apiKeyId: "key-a" }), + entry({ requestId: "key-recent", timestamp: FIXED_NOW - 2 * DAY_MS, admissionKind: "configured", apiKeyId: "key-a" }), + entry({ requestId: "loopback", timestamp: FIXED_NOW - DAY_MS, admissionKind: "loopback", apiKeyId: undefined }), + ]; + writeRaw(entries); + await foldUsagePrefix(); + const folded = readRollupSnapshot()!; + expect(folded.cutlineOffset).toBeGreaterThan(0); + const merged = await readApiKeyUsageRollup(["key-a"], 64 * 1024 * 1024); + const reference = rollupApiKeyUsage(entries, ["key-a"], FIXED_NOW); + expect(merged.rollup.get("key-a")).toEqual(reference.rollup.get("key-a")); + expect(merged.attributionSince).toBe(reference.attributionSince); + expect(merged.historyTruncated).toBeUndefined(); + expect(merged.rollup.get("key-a")).toMatchObject({ totalRequests: 2, requests7d: 1 }); + }); + + test("7. disabling the rollup makes API-key summaries fall back to the raw tail only", async () => { + const entries = [ + entry({ requestId: "old-folded", timestamp: FIXED_NOW - 15 * DAY_MS, admissionKind: "configured", apiKeyId: "key-a" }), + entry({ requestId: "recent-raw", timestamp: FIXED_NOW - 2 * DAY_MS, admissionKind: "configured", apiKeyId: "key-a" }), + ]; + writeRaw(entries); + await foldUsagePrefix(); + const cutline = readRollupSnapshot()!.cutlineOffset; + expect(cutline).toBeGreaterThan(0); + + clearApiKeyUsageCacheForTests(); + const withRollup = await readApiKeyUsageRollup(["key-a"], 64 * 1024 * 1024, true); + expect(withRollup.rollup.get("key-a")).toMatchObject({ totalRequests: 2 }); + + clearApiKeyUsageCacheForTests(); + // Constrain the raw read so the folded prefix cannot be re-read from raw: + // with the rollup DISABLED the sidecar must contribute nothing, so only the + // raw entry that fits in the read window is counted. + const tailBytes = statSync(usageLogPath()).size - cutline; + const withoutRollup = await readApiKeyUsageRollup(["key-a"], tailBytes, false); + expect(withoutRollup.rollup.get("key-a")).toMatchObject({ totalRequests: 1 }); + expect(withoutRollup.historyTruncated).toBe(true); + }); +}); diff --git a/tests/usage-rollup.test.ts b/tests/usage-rollup.test.ts new file mode 100644 index 000000000..65d114c8b --- /dev/null +++ b/tests/usage-rollup.test.ts @@ -0,0 +1,397 @@ +import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; +import { + appendFileSync, + mkdtempSync, + readFileSync, + renameSync, + rmSync, + statSync, + truncateSync, + unlinkSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { getConfigDir } from "../src/config"; +import { recordOwnedConfigPath } from "../src/lib/config-ownership"; +import { usageLogPath, type PersistedUsageEntry } from "../src/usage/log"; +import { + ensureRollupCurrent, + readRollupSnapshot, + resetRollupForTests, + setFoldSegmentCapForTests, + type RollupCommitRow, + type RollupMeta, +} from "../src/usage/rollup"; + +const DAY_MS = 86_400_000; +const FIXED_NOW = Date.UTC(2026, 7, 4, 12); + +let testDir = ""; +let previousHome: string | undefined; +let previousTimezone: string | undefined; +let clock: ReturnType; + +function rollupPath(): string { + return join(getConfigDir(), "usage-rollup.jsonl"); +} + +function metaPath(): string { + return join(getConfigDir(), "usage-rollup-meta.json"); +} + +function entry(overrides: Partial & { requestId: string; timestamp: number }): PersistedUsageEntry { + return { + provider: "openai", + model: "gpt-5.5", + status: 200, + durationMs: 10, + usageStatus: "reported", + usage: { inputTokens: 10, outputTokens: 2 }, + totalTokens: 12, + ...overrides, + }; +} + +function rawLine(value: PersistedUsageEntry): string { + return `${JSON.stringify(value)}\n`; +} + +function writeRaw(entries: PersistedUsageEntry[]): void { + writeFileSync(usageLogPath(), entries.map(rawLine).join(""), { mode: 0o600 }); +} + +function readMeta(): RollupMeta { + return JSON.parse(readFileSync(metaPath(), "utf8")) as RollupMeta; +} + +function writeMeta(meta: RollupMeta): void { + writeFileSync(metaPath(), `${JSON.stringify(meta, null, 2)}\n`, { mode: 0o600 }); +} + +function expireThrottle(): void { + writeMeta({ ...readMeta(), lastFoldAttemptAt: 0, updatedAt: 0 }); +} + +function parsedRollupLines(): Array> { + return readFileSync(rollupPath(), "utf8").split("\n").filter(Boolean).map(line => JSON.parse(line) as Record); +} + +function commits(): RollupCommitRow[] { + return parsedRollupLines().filter(row => row.kind === "commit") as unknown as RollupCommitRow[]; +} + +beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + previousTimezone = process.env.TZ; + testDir = mkdtempSync(join(tmpdir(), "ocx-rollup-")); + process.env.OPENCODEX_HOME = testDir; + recordOwnedConfigPath(testDir, usageLogPath()); + clock = spyOn(Date, "now").mockReturnValue(FIXED_NOW); + resetRollupForTests(); +}); + +afterEach(() => { + clock.mockRestore(); + resetRollupForTests(); + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (previousTimezone === undefined) delete process.env.TZ; + else process.env.TZ = previousTimezone; + if (testDir) rmSync(testDir, { recursive: true, force: true }); +}); + +describe("usage rollup core", () => { + test("1. folds day, model, provider, and key aggregates with combo and surface semantics", async () => { + writeRaw([ + entry({ + requestId: "priced", timestamp: FIXED_NOW - 12 * DAY_MS, + admissionKind: "configured", apiKeyId: "key-a", + usage: { inputTokens: 100, outputTokens: 10 }, totalTokens: 110, + }), + entry({ + requestId: "combo", timestamp: FIXED_NOW - 11 * DAY_MS, + surface: "claude-desktop", provider: "anthropic", model: "claude-fable-5", + usageStatus: "estimated", usage: { inputTokens: 20, outputTokens: 3, estimated: true }, totalTokens: 23, + attempts: [ + { ordinal: 1, provider: "anthropic", model: "claude-fable-5", adapter: "anthropic-messages", status: 200, durationMs: 5, sendCount: 1, recoveryKinds: [], usageStatus: "reported", usage: { inputTokens: 20, outputTokens: 2 }, totalTokens: 22 }, + { ordinal: 2, provider: "unknown", model: "unpriced-model", adapter: "openai-chat", status: 200, durationMs: 5, sendCount: 1, recoveryKinds: [], usageStatus: "estimated", usage: { inputTokens: 5, outputTokens: 1, estimated: true }, totalTokens: 6 }, + ], + }), + entry({ + requestId: "unsupported", timestamp: FIXED_NOW - 10 * DAY_MS, + surface: "grok", provider: "xai", model: "grok-4", usageStatus: "unsupported", + usage: undefined, totalTokens: undefined, admissionKind: "environment", + }), + ]); + + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot(); + expect(snapshot).not.toBeNull(); + expect(snapshot!.days).toHaveLength(3); + expect(snapshot!.models).toHaveLength(4); + expect(snapshot!.providers).toHaveLength(4); + expect(snapshot!.keys).toHaveLength(2); + expect(snapshot!.days.find(row => row.surface === "codex")).toMatchObject({ + statusCounts: { reported: 1, unreported: 0, unsupported: 0, estimated: 0 }, + attemptCount: 1, + tokens: { inputTokens: 100, outputTokens: 10, totalTokens: 110 }, + pricedRequests: 1, unpricedRequests: 0, unmeteredRequests: 0, + }); + expect(snapshot!.days.find(row => row.surface === "claude-desktop")).toMatchObject({ + statusCounts: { reported: 0, unreported: 0, unsupported: 0, estimated: 1 }, + attemptCount: 2, + tokens: { inputTokens: 20, outputTokens: 3, totalTokens: 23 }, + pricedRequests: 0, unpricedRequests: 1, unmeteredRequests: 0, + }); + expect(snapshot!.days.find(row => row.surface === "grok")).toMatchObject({ + statusCounts: { reported: 0, unreported: 0, unsupported: 1, estimated: 0 }, + unmeteredRequests: 1, + }); + expect(snapshot!.models.find(row => row.model === "unpriced-model")).toMatchObject({ + surface: "claude-desktop", requests: 1, attemptCount: 1, + foldedStatusCounts: { reported: 0, unreported: 0, unsupported: 0, estimated: 1 }, + tokens: { inputTokens: 5, outputTokens: 1, totalTokens: 6 }, + estimatedCostUsd: 0, + }); + expect(snapshot!.providers.find(row => row.provider === "anthropic")).toMatchObject({ requests: 1, attemptCount: 1, totalTokens: 22 }); + expect(snapshot!.keys.find(row => row.apiKeyId === "key-a")).toMatchObject({ requests: 1, requestsWithTimestamp: 1 }); + expect(snapshot!.attributionSinceMs).toBe(FIXED_NOW - 12 * DAY_MS); + expect(snapshot!.oldestTimestampMs).toBe(FIXED_NOW - 12 * DAY_MS); + }); + + test("2a-b. truncated commit and abandoned rows stay invisible while a fresh attempt succeeds", async () => { + const entries = [entry({ requestId: "a", timestamp: FIXED_NOW - 12 * DAY_MS })]; + writeRaw(entries); + await ensureRollupCurrent(); + const firstAttempt = commits()[0]!.attemptId; + const text = readFileSync(rollupPath(), "utf8"); + const commitStart = text.lastIndexOf('{"kind":"commit"'); + writeFileSync(rollupPath(), text.slice(0, commitStart) + text.slice(commitStart, commitStart + 24)); + expireThrottle(); + + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot(); + expect(snapshot?.days.reduce((sum, row) => sum + Object.values(row.statusCounts).reduce((a, b) => a + b, 0), 0)).toBe(1); + const attempts = new Set(parsedRollupLines().filter(row => row.kind !== "commit").map(row => row.attemptId)); + expect(attempts.has(firstAttempt)).toBe(true); + expect(attempts.size).toBe(2); + expect(commits()).toHaveLength(1); + expect(commits()[0]!.attemptId).not.toBe(firstAttempt); + }); + + test("2c. row-count or payload-digest mismatch rejects a segment", async () => { + writeRaw([entry({ requestId: "bad-commit", timestamp: FIXED_NOW - 12 * DAY_MS })]); + await ensureRollupCurrent(); + const rows = parsedRollupLines(); + const commit = rows.at(-1)!; + const originalDigest = String(commit.payloadDigest); + commit.payloadDigest = "0".repeat(64); + writeFileSync(rollupPath(), `${rows.map(row => JSON.stringify(row)).join("\n")}\n`); + expect(readRollupSnapshot()).toMatchObject({ cutlineOffset: 0, days: [] }); + // Restore the ORIGINAL digest (re-reading commits() here would read the + // tampered file back and leave the digest broken, so the rowCount check + // below would never be exercised in isolation). + commit.payloadDigest = originalDigest; + commit.rowCount = Number(commit.rowCount) + 1; + writeFileSync(rollupPath(), `${rows.map(row => JSON.stringify(row)).join("\n")}\n`); + expect(readRollupSnapshot()).toMatchObject({ cutlineOffset: 0, days: [] }); + }); + + test("2d. missing meta makes the snapshot null and rebuilds from raw", async () => { + writeRaw([entry({ requestId: "meta", timestamp: FIXED_NOW - 12 * DAY_MS })]); + await ensureRollupCurrent(); + unlinkSync(metaPath()); + expect(readRollupSnapshot()).toBeNull(); + await ensureRollupCurrent(); + expect(readRollupSnapshot()).toMatchObject({ days: [expect.objectContaining({ attemptCount: 1 })] }); + }); + + test("2e. append-boundary truncation repairs a partial trailing group row", async () => { + const first = entry({ requestId: "first", timestamp: FIXED_NOW - 12 * DAY_MS }); + writeRaw([first]); + await ensureRollupCurrent(); + const firstCutline = readRollupSnapshot()!.cutlineOffset; + appendFileSync(rollupPath(), '{"kind":"day","seg":'); + appendFileSync(usageLogPath(), rawLine(entry({ requestId: "second", timestamp: FIXED_NOW - 11 * DAY_MS }))); + expireThrottle(); + + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot()!; + expect(snapshot.cutlineOffset).toBeGreaterThan(firstCutline); + expect(snapshot.days.reduce((sum, row) => sum + row.attemptCount, 0)).toBe(2); + expect(readFileSync(rollupPath(), "utf8")).not.toContain('{"kind":"day","seg":{"kind"'); + expect(commits()).toHaveLength(2); + }); + + test("3. lineage mismatch returns null and refolds from offset zero", async () => { + const rows = [entry({ requestId: "lineage", timestamp: FIXED_NOW - 12 * DAY_MS })]; + writeRaw(rows); + await ensureRollupCurrent(); + const oldAttempt = commits()[0]!.attemptId; + renameSync(usageLogPath(), `${usageLogPath()}.old`); + writeRaw(rows); + expect(readRollupSnapshot()).toBeNull(); + await ensureRollupCurrent(); + expect(commits()).toHaveLength(1); + expect(commits()[0]!.seg).toBe(0); + expect(commits()[0]!.attemptId).not.toBe(oldAttempt); + }); + + test("4a. price-fingerprint mismatch rebuilds the cache from zero", async () => { + writeRaw([entry({ requestId: "price", timestamp: FIXED_NOW - 12 * DAY_MS })]); + await ensureRollupCurrent(); + const oldAttempt = commits()[0]!.attemptId; + writeMeta({ ...readMeta(), priceFingerprint: "stale", lastFoldAttemptAt: 0 }); + expect(readRollupSnapshot()).toBeNull(); + await ensureRollupCurrent(); + expect(commits()).toHaveLength(1); + expect(commits()[0]).toMatchObject({ seg: 0 }); + expect(commits()[0]!.attemptId).not.toBe(oldAttempt); + }); + + test("4b. boundary-digest mismatch rebuilds the cache", async () => { + writeRaw([entry({ requestId: "boundary-a", timestamp: FIXED_NOW - 12 * DAY_MS })]); + await ensureRollupCurrent(); + const oldAttempt = commits()[0]!.attemptId; + const raw = readFileSync(usageLogPath(), "utf8").replace("boundary-a", "boundary-b"); + writeFileSync(usageLogPath(), raw); + expireThrottle(); + await ensureRollupCurrent(); + expect(commits()).toHaveLength(1); + expect(commits()[0]!.attemptId).not.toBe(oldAttempt); + }); + + test("4c. live raw size below the committed cutline rebuilds", async () => { + const first = entry({ requestId: "first", timestamp: FIXED_NOW - 13 * DAY_MS }); + const second = entry({ requestId: "second", timestamp: FIXED_NOW - 12 * DAY_MS }); + writeRaw([first, second]); + await ensureRollupCurrent(); + const oldCutline = readRollupSnapshot()!.cutlineOffset; + truncateSync(usageLogPath(), Buffer.byteLength(rawLine(first))); + expireThrottle(); + await ensureRollupCurrent(); + expect(readRollupSnapshot()!.cutlineOffset).toBeLessThan(oldCutline); + expect(readRollupSnapshot()!.days.reduce((sum, row) => sum + row.attemptCount, 0)).toBe(1); + }); + + test("5. cutline ends on a newline and a partial raw row remains in the tail", async () => { + const complete = rawLine(entry({ requestId: "complete", timestamp: FIXED_NOW - 12 * DAY_MS })); + const partial = JSON.stringify(entry({ requestId: "partial", timestamp: FIXED_NOW - 13 * DAY_MS })).slice(0, 40); + writeFileSync(usageLogPath(), `${complete}${partial}`); + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot()!; + expect(snapshot.cutlineOffset).toBe(Buffer.byteLength(complete)); + expect(readFileSync(usageLogPath()).subarray(snapshot.cutlineOffset).toString()).toBe(partial); + }); + + test("6. the nine-local-day watermark is timezone-aware and stops at the first young row", async () => { + process.env.TZ = "America/New_York"; + const localNow = new Date(2026, 7, 4, 0, 30).getTime(); + clock.mockReturnValue(localNow); + const old = entry({ requestId: "old", timestamp: new Date(2026, 6, 25, 23, 30).getTime() }); + const boundaryDay = entry({ requestId: "boundary", timestamp: new Date(2026, 6, 26, 0, 1).getTime() }); + const outOfOrderOld = entry({ requestId: "old-after-boundary", timestamp: new Date(2026, 6, 24, 12).getTime() }); + writeRaw([old, boundaryDay, outOfOrderOld]); + + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot()!; + expect(snapshot.cutlineOffset).toBe(Buffer.byteLength(rawLine(old))); + expect(snapshot.days.reduce((sum, row) => sum + row.attemptCount, 0)).toBe(1); + expect(readFileSync(usageLogPath(), "utf8").slice(snapshot.cutlineOffset)).toContain("old-after-boundary"); + }); + + test("7. a complete malformed row does not stall the cutline; the fold advances past it", async () => { + const before = entry({ requestId: "before-junk", timestamp: FIXED_NOW - 12 * DAY_MS }); + const after = entry({ requestId: "after-junk", timestamp: FIXED_NOW - 11 * DAY_MS }); + writeFileSync(usageLogPath(), `${rawLine(before)}{not json}\n${rawLine(after)}`, { mode: 0o600 }); + + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot()!; + // The cutline covers all three complete rows (junk skipped like the parser does). + expect(snapshot.cutlineOffset).toBe(statSync(usageLogPath()).size); + expect(snapshot.days.reduce((sum, row) => sum + row.attemptCount, 0)).toBe(2); + }); + + test("7b. a recent object row without a requestId cannot stall the cutline either", async () => { + const old = entry({ requestId: "old-real", timestamp: FIXED_NOW - 12 * DAY_MS }); + const noId = `{"timestamp":${FIXED_NOW - DAY_MS},"provider":"openai"}\n`; + const oldAfter = entry({ requestId: "old-after", timestamp: FIXED_NOW - 11 * DAY_MS }); + writeFileSync(usageLogPath(), `${rawLine(old)}${noId}${rawLine(oldAfter)}`, { mode: 0o600 }); + + await ensureRollupCurrent(); + const snapshot = readRollupSnapshot()!; + // parseUsageRange requires a string requestId, so the id-less row can never + // contribute usage — the cutline advances past it instead of stalling. + expect(snapshot.cutlineOffset).toBe(statSync(usageLogPath()).size); + expect(snapshot.days.reduce((sum, row) => sum + row.attemptCount, 0)).toBe(2); + }); + + test("9b. a same-size rewrite of an EARLIER folded segment invalidates the snapshot at read time", async () => { + setFoldSegmentCapForTests(1); // one row per segment → multiple segments + try { + const first = entry({ requestId: "seg-one", timestamp: FIXED_NOW - 14 * DAY_MS }); + const second = entry({ requestId: "seg-two", timestamp: FIXED_NOW - 12 * DAY_MS }); + writeRaw([first, second]); + await ensureRollupCurrent(); + expect(commits().length).toBeGreaterThanOrEqual(2); + expect(readRollupSnapshot()).not.toBeNull(); + + // Rewrite ONLY the first row, same byte length, leaving the final + // segment's bytes (and its boundary tail) untouched. + const tampered = entry({ requestId: "seg-0ne", timestamp: FIXED_NOW - 14 * DAY_MS }); + expect(rawLine(tampered).length).toBe(rawLine(first).length); + writeFileSync(usageLogPath(), `${rawLine(tampered)}${rawLine(second)}`, { mode: 0o600 }); + expect(readRollupSnapshot()).toBeNull(); + } finally { + setFoldSegmentCapForTests(null); + } + }); + + test("8. the fold catches up a backlog in bounded segments, one commit per segment", async () => { + setFoldSegmentCapForTests(1); // every row closes a segment + try { + const entries = [ + entry({ requestId: "seg-a", timestamp: FIXED_NOW - 14 * DAY_MS }), + entry({ requestId: "seg-b", timestamp: FIXED_NOW - 13 * DAY_MS }), + entry({ requestId: "seg-c", timestamp: FIXED_NOW - 12 * DAY_MS }), + ]; + writeRaw(entries); + await ensureRollupCurrent(); + expect(commits()).toHaveLength(3); + const snapshot = readRollupSnapshot()!; + expect(snapshot.cutlineOffset).toBe(statSync(usageLogPath()).size); + expect(snapshot.days.reduce((sum, row) => sum + row.attemptCount, 0)).toBe(3); + // Segments chain: each commit's seg is the previous commit's toOffset. + const chain = commits(); + for (let i = 1; i < chain.length; i++) expect(chain[i]!.seg).toBe(chain[i - 1]!.toOffset); + } finally { + setFoldSegmentCapForTests(null); + } + }); + + test("9. a truncated-then-regrown raw log invalidates the snapshot at read time, before any fold", async () => { + const old = entry({ requestId: "will-vanish", timestamp: FIXED_NOW - 12 * DAY_MS }); + writeRaw([old]); + await ensureRollupCurrent(); + expect(readRollupSnapshot()).not.toBeNull(); + + // Rewrite the raw file to the SAME length with different bytes (regrow after + // truncation) without touching meta or the throttle: a reader must notice. + const replacement = entry({ requestId: "will-van1sh", timestamp: FIXED_NOW - 12 * DAY_MS }); + writeFileSync(usageLogPath(), rawLine(replacement), { mode: 0o600 }); + expect(readRollupSnapshot()).toBeNull(); + }); + + test("rollup files are owned config state and mode 0600 on POSIX", async () => { + writeRaw([entry({ requestId: "mode", timestamp: FIXED_NOW - 12 * DAY_MS })]); + await ensureRollupCurrent(); + const manifest = JSON.parse(readFileSync(join(getConfigDir(), ".opencodex-uninstall.json"), "utf8")) as { paths: string[] }; + expect(manifest.paths).toContain("usage-rollup.jsonl"); + expect(manifest.paths).toContain("usage-rollup-meta.json"); + if (process.platform !== "win32") { + expect(statSync(rollupPath()).mode & 0o777).toBe(0o600); + expect(statSync(metaPath()).mode & 0o777).toBe(0o600); + } + }); +});