From 846d0d02b0965e3691494d8c196f841040c3bc7d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Demir=20B=C3=BClb=C3=BClo=C4=9Flu?= Date: Tue, 15 Sep 2026 16:28:37 +0300 Subject: [PATCH] fix(server): include Codex archived_sessions in usage scan Codex moves rolled-out sessions to archived_sessions, but the usage scanner only read sessions/, so archived history never contributed cost. Scan both directories and give Codex events a stable dedupeKey so copies across the two are charged once. Bump the scan cache version since v3 Codex records have null keys. --- apps/server/src/usage/UsageService.test.ts | 20 +++++++++++++------ apps/server/src/usage/UsageService.ts | 14 +++++++++++++ apps/server/src/usage/usageScanCache.test.ts | 2 +- apps/server/src/usage/usageScanCache.ts | 4 +++- .../server/src/usage/usageTranscripts.test.ts | 18 +++++++++++++++++ apps/server/src/usage/usageTranscripts.ts | 18 ++++++++++++++--- 6 files changed, 65 insertions(+), 11 deletions(-) diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 27c3a3ff3efe..a337ceb9d9f6 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -126,8 +126,7 @@ describe("UsageService", () => { ); await NodeFSP.mkdir(NodePath.join(codexHome, "sessions"), { recursive: true }); await NodeFSP.symlink(codexHome, alias, "junction"); - await NodeFSP.writeFile( - NodePath.join(codexHome, "sessions", "rollout.jsonl"), + const codexRollout = [ { type: "session_meta", payload: { id: "codex-account-session" } }, { type: "turn_context", payload: { model: "gpt-5.6-sol" } }, @@ -141,7 +140,16 @@ describe("UsageService", () => { }, ] .map((line) => encodeUnknownJsonString(line)) - .join("\n") + "\n", + .join("\n") + "\n"; + await NodeFSP.writeFile( + NodePath.join(codexHome, "sessions", "rollout.jsonl"), + codexRollout, + ); + // An identical archived copy must be scanned but charged only once. + await NodeFSP.mkdir(NodePath.join(codexHome, "archived_sessions"), { recursive: true }); + await NodeFSP.writeFile( + NodePath.join(codexHome, "archived_sessions", "rollout.jsonl"), + codexRollout, ); await NodeFSP.mkdir(NodePath.join(grokHome, "sessions", "session"), { recursive: true }); await NodeFSP.writeFile( @@ -200,14 +208,14 @@ describe("UsageService", () => { const summary = yield* service.readSummary(WINDOW); assert.strictEqual(totalOutputTokens(summary), 36); const sources = summary.sources.filter((source) => source.status === "ok"); - assert.strictEqual(sources.length, 4); + assert.strictEqual(sources.length, 5); assert.strictEqual( sources.reduce((sum, source) => sum + source.scannedFiles, 0), - 4, + 5, ); assert.strictEqual( sources.filter((source) => source.fingerprint.provider === "codex").length, - 1, + 2, ); }).pipe(Effect.scoped), ); diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 7c942499996e..4fa14f1dde68 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -288,6 +288,20 @@ export const make = Effect.gen(function* () { if (seen.has(key)) continue; seen.add(key); dirs.push({ provider, dir, ...(provider === "grok" ? { fileName: "updates.jsonl" } : {}) }); + // Codex moves rolled-out sessions to archived_sessions; without this + // their cost never reaches usage. Copies across the two directories + // are charged once via the record dedupeKey in the aggregator. + if (provider === "codex") { + const archivedDirectory = path.resolve(home, "archived_sessions"); + const archivedDir = yield* fileSystem + .realPath(archivedDirectory) + .pipe(Effect.orElseSucceed(() => archivedDirectory)); + const archivedKey = `${provider}\0${archivedDir}`; + if (!seen.has(archivedKey)) { + seen.add(archivedKey); + dirs.push({ provider, dir: archivedDir }); + } + } } } return dirs; diff --git a/apps/server/src/usage/usageScanCache.test.ts b/apps/server/src/usage/usageScanCache.test.ts index fdb0aabafa40..036767522baa 100644 --- a/apps/server/src/usage/usageScanCache.test.ts +++ b/apps/server/src/usage/usageScanCache.test.ts @@ -125,7 +125,7 @@ describe("scan cache round trip", () => { it("rejects a document from the previous cache version", () => { const encoded = encodeScanCache(cacheWith([["/a.jsonl", 100, [record()]]])); - const previous = { ...encoded, version: 2 }; + const previous = { ...encoded, version: 3 }; expect(decodeScanCache(JSON.parse(JSON.stringify(previous))).size).toBe(0); }); diff --git a/apps/server/src/usage/usageScanCache.ts b/apps/server/src/usage/usageScanCache.ts index 224f109147e4..d1e28bc98ab8 100644 --- a/apps/server/src/usage/usageScanCache.ts +++ b/apps/server/src/usage/usageScanCache.ts @@ -26,7 +26,9 @@ import type { CodexScanState, UsageRecord } from "./usageTranscripts.ts"; // entries would keep serving double-counted records forever. // v3: entries carry the parse position and reducer state so a grown file // re-parses only its appended bytes instead of starting over. -const USAGE_SCAN_CACHE_VERSION = 3 as const; +// v4: Codex events carry a stable dedupeKey so archived copies dedupe; +// v3 entries with null Codex keys would double count them. +const USAGE_SCAN_CACHE_VERSION = 4 as const; export interface CachedFile { readonly size: number; diff --git a/apps/server/src/usage/usageTranscripts.test.ts b/apps/server/src/usage/usageTranscripts.test.ts index b09db613ed85..673c728225cf 100644 --- a/apps/server/src/usage/usageTranscripts.test.ts +++ b/apps/server/src/usage/usageTranscripts.test.ts @@ -113,6 +113,24 @@ describe("parseCodexLine", () => { expect(record?.totals.reasoningTokens).toBe(116); }); + it("gives identical events a stable identity across rollout copies", () => { + const firstState = initialCodexScanState(); + parseCodexLine(sessionMeta, firstState); + parseCodexLine(turnContext, firstState); + const first = parseCodexLine(tokenCount(19239, 11008, 299, 116), firstState); + + // A second parse of the same copied file (e.g. sessions/ vs + // archived_sessions) must produce the same key so the aggregator + // charges it once. + const secondState = initialCodexScanState(); + parseCodexLine(sessionMeta, secondState); + parseCodexLine(turnContext, secondState); + const second = parseCodexLine(tokenCount(19239, 11008, 299, 116), secondState); + + expect(first?.dedupeKey).not.toBeNull(); + expect(second?.dedupeKey).toBe(first?.dedupeKey); + }); + it("skips a repeated token_count so deltas are not double counted", () => { const state = initialCodexScanState(); parseCodexLine(turnContext, state); diff --git a/apps/server/src/usage/usageTranscripts.ts b/apps/server/src/usage/usageTranscripts.ts index 5d909379eb10..213d57ef8bd8 100644 --- a/apps/server/src/usage/usageTranscripts.ts +++ b/apps/server/src/usage/usageTranscripts.ts @@ -304,9 +304,21 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord totals, // Codex does not report cost in the rollout. reportedCostUsd: null, - // Events surviving the fork-copy suppression above are unique to this - // rollout, so they need no global dedup. - dedupeKey: null, + // Archiving or copying a rollout must not charge the same event twice. + dedupeKey: + state.sessionId.length === 0 + ? null + : JSON.stringify([ + "codex", + state.sessionId, + timestampMs, + state.model, + totals.uncachedInputTokens, + totals.cachedInputTokens, + totals.cacheCreationTokens, + totals.outputTokens, + totals.reasoningTokens, + ]), }; }