| author | |
| committer | |
| log | 0d38866dcf63d133b2ed732bbb303dc533b5934f |
| tree | d875c8887211c7f284c0b93a41b522f3579bbeb7 |
| parent | f993fa1c5fe51dfae78ec8869c9694b7cb97115f |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>3 files changed, 159 insertions(+), 3 deletions(-)
apps/server/src/usage/usageScanCache.ts+3-1| ... | @@ -18,7 +18,9 @@ import type { UsageProviderKind } from "@t3tools/contracts"; | ... | @@ -18,7 +18,9 @@ import type { UsageProviderKind } from "@t3tools/contracts"; |
| 18 | 18 | ||
| 19 | import type { UsageRecord } from "./usageTranscripts.ts"; | 19 | import type { UsageRecord } from "./usageTranscripts.ts"; |
| 20 | 20 | ||
| 21 | export const USAGE_SCAN_CACHE_VERSION = 1 as const; | 21 | // v2: Codex fork-copy suppression changed what a file parses to, so v1 |
| 22 | // entries would keep serving double-counted records forever. | ||
| 23 | export const USAGE_SCAN_CACHE_VERSION = 2 as const; | ||
| 22 | 24 | ||
| 23 | export interface CachedFile { | 25 | export interface CachedFile { |
| 24 | readonly size: number; | 26 | readonly size: number; |
apps/server/src/usage/usageTranscripts.test.ts+100| ... | @@ -134,6 +134,106 @@ describe("parseCodexLine", () => { | ... | @@ -134,6 +134,106 @@ describe("parseCodexLine", () => { |
| 134 | parseCodexLine(turnContext, state); | 134 | parseCodexLine(turnContext, state); |
| 135 | expect(parseCodexLine(tokenCount(100, 0, 10, 0), state)).not.toBeNull(); | 135 | expect(parseCodexLine(tokenCount(100, 0, 10, 0), state)).not.toBeNull(); |
| 136 | }); | 136 | }); |
| 137 | |||
| 138 | // A forked/subagent rollout opens with the parent's history copied in and | ||
| 139 | // every line re-stamped to the fork instant, then the ancestors' session | ||
| 140 | // metas. Counting those again multiplied usage ~1.85x on real data (#5758). | ||
| 141 | describe("forked rollouts", () => { | ||
| 142 | const meta = (overrides: { | ||
| 143 | id: string; | ||
| 144 | timestamp: string; | ||
| 145 | forkedFromId?: string; | ||
| 146 | spawnParentId?: string; | ||
| 147 | }) => | ||
| 148 | JSON.stringify({ | ||
| 149 | type: "session_meta", | ||
| 150 | timestamp: overrides.timestamp, | ||
| 151 | payload: { | ||
| 152 | type: "session_meta", | ||
| 153 | id: overrides.id, | ||
| 154 | ...(overrides.forkedFromId === undefined | ||
| 155 | ? {} | ||
| 156 | : { forked_from_id: overrides.forkedFromId }), | ||
| 157 | ...(overrides.spawnParentId === undefined | ||
| 158 | ? {} | ||
| 159 | : { | ||
| 160 | source: { | ||
| 161 | subagent: { thread_spawn: { parent_thread_id: overrides.spawnParentId } }, | ||
| 162 | }, | ||
| 163 | }), | ||
| 164 | }, | ||
| 165 | }); | ||
| 166 | const stamped = (timestamp: string, line: string) => { | ||
| 167 | const parsed = JSON.parse(line) as { timestamp: string }; | ||
| 168 | parsed.timestamp = timestamp; | ||
| 169 | return JSON.stringify(parsed); | ||
| 170 | }; | ||
| 171 | |||
| 172 | it("keeps the child session id over copied ancestor metas", () => { | ||
| 173 | const state = initialCodexScanState(); | ||
| 174 | parseCodexLine(meta({ id: "child", timestamp: "2026-08-01T05:00:00.000Z" }), state); | ||
| 175 | parseCodexLine(meta({ id: "parent", timestamp: "2026-08-01T05:00:00.000Z" }), state); | ||
| 176 | parseCodexLine(turnContext, state); | ||
| 177 | const record = parseCodexLine(tokenCount(100, 0, 10, 0), state); | ||
| 178 | |||
| 179 | expect(record?.sessionId).toBe("child"); | ||
| 180 | }); | ||
| 181 | |||
| 182 | it("drops the re-stamped copied burst and keeps the first real event", () => { | ||
| 183 | const state = initialCodexScanState(); | ||
| 184 | const forkInstant = "2026-08-01T05:00:00.000Z"; | ||
| 185 | parseCodexLine(meta({ id: "child", timestamp: forkInstant, forkedFromId: "parent" }), state); | ||
| 186 | parseCodexLine(meta({ id: "parent", timestamp: forkInstant }), state); | ||
| 187 | parseCodexLine(stamped(forkInstant, turnContext), state); | ||
| 188 | |||
| 189 | // Copied history: written in one burst at the fork instant. | ||
| 190 | expect( | ||
| 191 | parseCodexLine(stamped("2026-08-01T05:00:00.001Z", tokenCount(100, 0, 10, 0)), state), | ||
| 192 | ).toBeNull(); | ||
| 193 | expect( | ||
| 194 | parseCodexLine(stamped("2026-08-01T05:00:00.002Z", tokenCount(200, 0, 20, 0)), state), | ||
| 195 | ).toBeNull(); | ||
| 196 | |||
| 197 | // The child's first genuine turn lands seconds later and must count. | ||
| 198 | const real = parseCodexLine( | ||
| 199 | stamped("2026-08-01T05:00:06.000Z", tokenCount(300, 0, 30, 0)), | ||
| 200 | state, | ||
| 201 | ); | ||
| 202 | expect(real).not.toBeNull(); | ||
| 203 | expect(real?.totals.outputTokens).toBe(30); | ||
| 204 | |||
| 205 | // Suppression never restarts, even for closely spaced later events. | ||
| 206 | const next = parseCodexLine( | ||
| 207 | stamped("2026-08-01T05:00:06.100Z", tokenCount(400, 0, 40, 0)), | ||
| 208 | state, | ||
| 209 | ); | ||
| 210 | expect(next).not.toBeNull(); | ||
| 211 | }); | ||
| 212 | |||
| 213 | it("recognizes subagent spawns without forked_from_id", () => { | ||
| 214 | const state = initialCodexScanState(); | ||
| 215 | const spawnInstant = "2026-08-01T05:00:00.000Z"; | ||
| 216 | parseCodexLine( | ||
| 217 | meta({ id: "child", timestamp: spawnInstant, spawnParentId: "parent" }), | ||
| 218 | state, | ||
| 219 | ); | ||
| 220 | parseCodexLine(stamped(spawnInstant, turnContext), state); | ||
| 221 | expect( | ||
| 222 | parseCodexLine(stamped("2026-08-01T05:00:00.001Z", tokenCount(100, 0, 10, 0)), state), | ||
| 223 | ).toBeNull(); | ||
| 224 | }); | ||
| 225 | |||
| 226 | it("does not suppress anything in a rollout that is not a fork", () => { | ||
| 227 | const state = initialCodexScanState(); | ||
| 228 | parseCodexLine(meta({ id: "root", timestamp: "2026-08-01T05:00:00.000Z" }), state); | ||
| 229 | parseCodexLine(stamped("2026-08-01T05:00:00.100Z", turnContext), state); | ||
| 230 | const record = parseCodexLine( | ||
| 231 | stamped("2026-08-01T05:00:00.200Z", tokenCount(100, 0, 10, 0)), | ||
| 232 | state, | ||
| 233 | ); | ||
| 234 | expect(record).not.toBeNull(); | ||
| 235 | }); | ||
| 236 | }); | ||
| 137 | }); | 237 | }); |
| 138 | 238 | ||
| 139 | describe("totalTokens", () => { | 239 | describe("totalTokens", () => { |
apps/server/src/usage/usageTranscripts.ts+56-2| ... | @@ -151,10 +151,42 @@ export interface CodexScanState { | ... | @@ -151,10 +151,42 @@ export interface CodexScanState { |
| 151 | model: string; | 151 | model: string; |
| 152 | sessionId: string; | 152 | sessionId: string; |
| 153 | lastUsageSignature: string | null; | 153 | lastUsageSignature: string | null; |
| 154 | sawSessionMeta: boolean; | ||
| 155 | /** While true, leading usage events are re-stamped copies of parent history. */ | ||
| 156 | suppressingForkCopies: boolean; | ||
| 157 | forkCopyAnchorMs: number; | ||
| 154 | } | 158 | } |
| 155 | 159 | ||
| 156 | export function initialCodexScanState(): CodexScanState { | 160 | export function initialCodexScanState(): CodexScanState { |
| 157 | return { model: "", sessionId: "", lastUsageSignature: null }; | 161 | return { |
| 162 | model: "", | ||
| 163 | sessionId: "", | ||
| 164 | lastUsageSignature: null, | ||
| 165 | sawSessionMeta: false, | ||
| 166 | suppressingForkCopies: false, | ||
| 167 | forkCopyAnchorMs: 0, | ||
| 168 | }; | ||
| 169 | } | ||
| 170 | |||
| 171 | /** | ||
| 172 | * A forked or subagent rollout opens with the parent's full history copied in, | ||
| 173 | * every line re-stamped to the fork instant. Those copies are written in one | ||
| 174 | * synchronous burst (observed gaps 0-40ms), while the child's first genuine | ||
| 175 | * usage event only lands after a real model turn (observed 5s+). One second of | ||
| 176 | * separation splits the two cleanly; `ccusage` uses the same threshold. | ||
| 177 | */ | ||
| 178 | const FORK_COPY_MAX_GAP_MS = 1000; | ||
| 179 | |||
| 180 | /** Whether a `session_meta` payload marks the rollout as a fork or subagent. */ | ||
| 181 | function isForkedSessionMeta(payload: Record<string, unknown>): boolean { | ||
| 182 | if (typeof payload["forked_from_id"] === "string") return true; | ||
| 183 | const source = payload["source"]; | ||
| 184 | if (typeof source !== "object" || source === null) return false; | ||
| 185 | const subagent = (source as Record<string, unknown>)["subagent"]; | ||
| 186 | if (typeof subagent !== "object" || subagent === null) return false; | ||
| 187 | const spawn = (subagent as Record<string, unknown>)["thread_spawn"]; | ||
| 188 | if (typeof spawn !== "object" || spawn === null) return false; | ||
| 189 | return typeof (spawn as Record<string, unknown>)["parent_thread_id"] === "string"; | ||
| 158 | } | 190 | } |
| 159 | 191 | ||
| 160 | /** | 192 | /** |
| ... | @@ -181,8 +213,18 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord | ... | @@ -181,8 +213,18 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord |
| 181 | const payloadType = payloadRecord["type"]; | 213 | const payloadType = payloadRecord["type"]; |
| 182 | 214 | ||
| 183 | if (record["type"] === "session_meta") { | 215 | if (record["type"] === "session_meta") { |
| 216 | // Only the first meta describes this file's own session. A forked rollout | ||
| 217 | // repeats the ancestors' metas right after it; letting those through would | ||
| 218 | // reassign every subsequent record to an ancestor session. | ||
| 219 | if (state.sawSessionMeta) return null; | ||
| 220 | state.sawSessionMeta = true; | ||
| 184 | const id = payloadRecord["id"] ?? payloadRecord["session_id"]; | 221 | const id = payloadRecord["id"] ?? payloadRecord["session_id"]; |
| 185 | if (typeof id === "string") state.sessionId = id; | 222 | if (typeof id === "string") state.sessionId = id; |
| 223 | const metaTimestampMs = parseTimestampMs(record["timestamp"]); | ||
| 224 | if (metaTimestampMs !== null && isForkedSessionMeta(payloadRecord)) { | ||
| 225 | state.suppressingForkCopies = true; | ||
| 226 | state.forkCopyAnchorMs = metaTimestampMs; | ||
| 227 | } | ||
| 186 | return null; | 228 | return null; |
| 187 | } | 229 | } |
| 188 | 230 | ||
| ... | @@ -213,6 +255,17 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord | ... | @@ -213,6 +255,17 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord |
| 213 | if (signature === state.lastUsageSignature) return null; | 255 | if (signature === state.lastUsageSignature) return null; |
| 214 | state.lastUsageSignature = signature; | 256 | state.lastUsageSignature = signature; |
| 215 | 257 | ||
| 258 | // In a forked rollout the copied parent history was already counted from the | ||
| 259 | // parent's own file. Drop the leading burst; the first usage event separated | ||
| 260 | // from its predecessor by a real turn's worth of time ends it for good. | ||
| 261 | if (state.suppressingForkCopies) { | ||
| 262 | if (timestampMs - state.forkCopyAnchorMs < FORK_COPY_MAX_GAP_MS) { | ||
| 263 | state.forkCopyAnchorMs = timestampMs; | ||
| 264 | return null; | ||
| 265 | } | ||
| 266 | state.suppressingForkCopies = false; | ||
| 267 | } | ||
| 268 | |||
| 216 | const inputTokens = int(lastRecord["input_tokens"]); | 269 | const inputTokens = int(lastRecord["input_tokens"]); |
| 217 | const cachedInputTokens = int(lastRecord["cached_input_tokens"]); | 270 | const cachedInputTokens = int(lastRecord["cached_input_tokens"]); |
| 218 | const cacheCreationTokens = int(lastRecord["cache_write_input_tokens"]); | 271 | const cacheCreationTokens = int(lastRecord["cache_write_input_tokens"]); |
| ... | @@ -238,7 +291,8 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord | ... | @@ -238,7 +291,8 @@ export function parseCodexLine(line: string, state: CodexScanState): UsageRecord |
| 238 | totals, | 291 | totals, |
| 239 | // Codex does not report cost in the rollout. | 292 | // Codex does not report cost in the rollout. |
| 240 | reportedCostUsd: null, | 293 | reportedCostUsd: null, |
| 241 | // Rollout files are unique per session, so events need no global dedup. | 294 | // Events surviving the fork-copy suppression above are unique to this |
| 295 | // rollout, so they need no global dedup. | ||
| 242 | dedupeKey: null, | 296 | dedupeKey: null, |
| 243 | }; | 297 | }; |
| 244 | } | 298 | } |