diff --git a/openwiki-facts/source-facts.json b/openwiki-facts/source-facts.json index 9db685fc..5555612f 100644 --- a/openwiki-facts/source-facts.json +++ b/openwiki-facts/source-facts.json @@ -198,95 +198,95 @@ }, { "name": "parseClaudeIncremental", - "evidence": "src/lib/rollout.js:208" + "evidence": "src/lib/rollout.js:211" }, { "name": "parseGeminiIncremental", - "evidence": "src/lib/rollout.js:331" + "evidence": "src/lib/rollout.js:334" }, { "name": "parseOpencodeIncremental", - "evidence": "src/lib/rollout.js:452" + "evidence": "src/lib/rollout.js:455" }, { "name": "parseOpenclawIncremental", - "evidence": "src/lib/rollout.js:605" + "evidence": "src/lib/rollout.js:608" }, { "name": "parseOpencodeDbIncremental", - "evidence": "src/lib/rollout.js:2478" + "evidence": "src/lib/rollout.js:2556" }, { "name": "parseCursorApiIncremental", - "evidence": "src/lib/rollout.js:2659" + "evidence": "src/lib/rollout.js:2737" }, { "name": "parseKiroIncremental", - "evidence": "src/lib/rollout.js:2917" + "evidence": "src/lib/rollout.js:2995" }, { "name": "parseHermesIncremental", - "evidence": "src/lib/rollout.js:3172" + "evidence": "src/lib/rollout.js:3250" }, { "name": "parseKiroCliIncremental", - "evidence": "src/lib/rollout.js:3791" + "evidence": "src/lib/rollout.js:3869" }, { "name": "parseKimiIncremental", - "evidence": "src/lib/rollout.js:4421" + "evidence": "src/lib/rollout.js:4499" }, { "name": "parseKimiCodeIncremental", - "evidence": "src/lib/rollout.js:4612" + "evidence": "src/lib/rollout.js:4690" }, { "name": "parseCodebuddyIncremental", - "evidence": "src/lib/rollout.js:4836" + "evidence": "src/lib/rollout.js:4914" }, { "name": "parseRoocodeIncremental", - "evidence": "src/lib/rollout.js:5339" + "evidence": "src/lib/rollout.js:5417" }, { "name": "parseZedIncremental", - "evidence": "src/lib/rollout.js:5656" + "evidence": "src/lib/rollout.js:5734" }, { "name": "parseGooseIncremental", - "evidence": "src/lib/rollout.js:5973" + "evidence": "src/lib/rollout.js:6051" }, { "name": "parseDroidIncremental", - "evidence": "src/lib/rollout.js:6350" + "evidence": "src/lib/rollout.js:6428" }, { "name": "parseKilocodeIncremental", - "evidence": "src/lib/rollout.js:6597" + "evidence": "src/lib/rollout.js:6675" }, { "name": "parseOmpIncremental", - "evidence": "src/lib/rollout.js:6733" + "evidence": "src/lib/rollout.js:6811" }, { "name": "parsePiIncremental", - "evidence": "src/lib/rollout.js:6986" + "evidence": "src/lib/rollout.js:7064" }, { "name": "parseCraftIncremental", - "evidence": "src/lib/rollout.js:7270" + "evidence": "src/lib/rollout.js:7348" }, { "name": "parseCopilotIncremental", - "evidence": "src/lib/rollout.js:7620" + "evidence": "src/lib/rollout.js:7698" }, { "name": "parseGrokBuildIncremental", - "evidence": "src/lib/rollout.js:8086" + "evidence": "src/lib/rollout.js:8164" }, { "name": "parseAntigravityIncremental", - "evidence": "src/lib/rollout.js:8258" + "evidence": "src/lib/rollout.js:8336" } ] } diff --git a/src/lib/codex-rollout-parser.js b/src/lib/codex-rollout-parser.js index 60a25eb2..d655903d 100644 --- a/src/lib/codex-rollout-parser.js +++ b/src/lib/codex-rollout-parser.js @@ -349,6 +349,15 @@ async function parseCodexRolloutFile(filePath, { fromIso, toIso, from = null, to let pendingSkills = []; let pendingExecEnds = []; // exec_command_end payloads since last token_count + // Codex subagent-fork replay handling (issue #75). A rollout spawned via + // session_meta.source.subagent.thread_spawn replays the parent thread's + // history before an "inter_agent_communication_metadata" boundary. While in + // replay we advance the delta baseline but do not attribute anything, so the + // context breakdown reflects only the genuine child turns. Whole-file read, + // so no cross-call cursor is needed — a boundary-less (partial) fork simply + // stays in replay and contributes nothing. + let accounting = true; + const totals = emptyTotals(); const byTool = new Map(); // tool_name -> {name,calls,totals} const bySkill = new Map(); // skill_name -> {name,calls,totals} @@ -525,20 +534,36 @@ async function parseCodexRolloutFile(filePath, { fromIso, toIso, from = null, to } catch { continue; } - const ts = typeof obj?.timestamp === "string" ? obj.timestamp : null; - if (!ts) continue; - if (fromIso && ts < fromIso) continue; - if (toIso && ts > toIso) continue; - if (!isTimestampInRequestedDayRange(ts, { from, to, timeZoneContext })) continue; - + // Stateful fork-control records are handled BEFORE any timestamp/date + // filtering. They must never be dropped by a query window: if a date filter + // removed the boundary while keeping replay records on one side of it, the + // replayed parent history would leak back into the child's totals (the exact + // #75 overcount). This mirrors the pre-gate ordering in rollout.js. if (obj.type === "session_meta") { const p = obj.payload || {}; sessionId = p.id || sessionId; cwd = p.cwd || cwd; cliVersion = p.cli_version || cliVersion; provider = p.model_provider || provider; + if (p.source?.subagent?.thread_spawn) accounting = false; + continue; + } + + // Boundary between replayed parent history and genuine child turns. + if (obj.type === "inter_agent_communication_metadata") { + accounting = true; + pendingCalls = []; + pendingSkills = []; + pendingExecEnds = []; + continue; } + const ts = typeof obj?.timestamp === "string" ? obj.timestamp : null; + if (!ts) continue; + if (fromIso && ts < fromIso) continue; + if (toIso && ts > toIso) continue; + if (!isTimestampInRequestedDayRange(ts, { from, to, timeZoneContext })) continue; + if (obj.type === "turn_context") { const p = obj.payload || {}; if (typeof p.cwd === "string") cwd = p.cwd; @@ -547,14 +572,17 @@ async function parseCodexRolloutFile(filePath, { fromIso, toIso, from = null, to } if (obj.type === "response_item" && obj.payload?.type === "function_call") { - pendingCalls.push(obj.payload); - const skill = extractSkillNameFromFunctionCall(obj.payload); - if (skill) pendingSkills.push(skill); + // Skip replayed parent tool calls; only child turns are attributed. + if (accounting) { + pendingCalls.push(obj.payload); + const skill = extractSkillNameFromFunctionCall(obj.payload); + if (skill) pendingSkills.push(skill); + } continue; } if (obj.type === "event_msg" && obj.payload?.type === "exec_command_end") { - pendingExecEnds.push(obj.payload); + if (accounting) pendingExecEnds.push(obj.payload); continue; } @@ -564,7 +592,11 @@ async function parseCodexRolloutFile(filePath, { fromIso, toIso, from = null, to const lastUsage = info?.last_token_usage; const totalUsage = info?.total_token_usage; const delta = pickDelta(lastUsage, totalUsage, prevTotals); + // Advance the delta baseline through replay so the first child delta is + // measured from the last replayed cumulative, but do not attribute the + // replayed parent history to this child. if (totalUsage && typeof totalUsage === "object") prevTotals = totalUsage; + if (!accounting) continue; if (delta) attributeTurn(delta); continue; } @@ -573,6 +605,16 @@ async function parseCodexRolloutFile(filePath, { fromIso, toIso, from = null, to rl.close(); stream.close?.(); + // Saw a subagent fork but never its boundary: the whole file was treated as + // replay and contributes nothing. Safe (never over-count) but silent — + // surface it under debug so a malformed/renamed boundary is diagnosable. + if (!accounting && process.env.TOKENTRACKER_DEBUG) { + process.stderr.write( + `[codex-fork] ${filePath}: subagent fork with no ` + + `inter_agent_communication_metadata boundary; context breakdown empty\n`, + ); + } + return { sessionId, cwd, diff --git a/src/lib/rollout.js b/src/lib/rollout.js index 7d81bbea..38e637c8 100644 --- a/src/lib/rollout.js +++ b/src/lib/rollout.js @@ -137,6 +137,7 @@ async function parseRolloutIncremental({ const startOffset = sameFile ? prev.offset || 0 : 0; const lastTotal = sameFile ? prev.lastTotal || null : null; const lastModel = sameFile ? prev.lastModel || null : null; + const forkState = sameFile ? prev.codexFork || null : null; const projectContext = projectEnabled ? await resolveProjectContextForFile({ @@ -155,6 +156,7 @@ async function parseRolloutIncremental({ startOffset, lastTotal, lastModel, + forkState, hourlyState, touchedBuckets, source: fileSource, @@ -172,6 +174,7 @@ async function parseRolloutIncremental({ offset: result.endOffset, lastTotal: result.lastTotal, lastModel: result.lastModel, + codexFork: result.codexFork || null, head: headSig, updatedAt: new Date().toISOString(), }; @@ -782,6 +785,7 @@ async function parseRolloutFile({ startOffset, lastTotal, lastModel, + forkState, hourlyState, touchedBuckets, source, @@ -796,7 +800,8 @@ async function parseRolloutFile({ const st = await fs.stat(filePath); const endOffset = st.size; if (startOffset >= endOffset) { - return { endOffset, lastTotal, lastModel, eventsAggregated: 0 }; + const carriedFork = forkState === "child" || forkState === "replay" ? forkState : null; + return { endOffset, lastTotal, lastModel, eventsAggregated: 0, codexFork: carriedFork }; } const stream = fssync.createReadStream(filePath, { encoding: "utf8", start: startOffset }); @@ -809,14 +814,37 @@ async function parseRolloutFile({ let currentProjectKey = projectKey || null; let eventsAggregated = 0; + // Codex subagent-fork accounting (issue #75). A rollout spawned via + // session_meta.source.subagent.thread_spawn replays the parent thread's + // token_count history before a deterministic + // "inter_agent_communication_metadata" boundary, then the genuine child + // turns. codexFork tracks that lifecycle across incremental reads: + // null -> not a fork (or not yet known); accounting on + // "replay" -> fork detected, boundary not seen; accounting off + // "child" -> boundary passed; accounting on + // The cumulative total_token_usage counter is continuous across the + // boundary, so we still advance the delta baseline through the replay + // (without accumulating) and let the first child delta be measured from the + // last replayed cumulative. + let codexFork = forkState === "child" || forkState === "replay" ? forkState : null; + let accounting = codexFork !== "replay"; + for await (const line of rl) { if (!line) continue; + // Format contract with Codex (issue #75): fork detection and the replay + // boundary are matched by these exact record-type substrings. If a future + // Codex format renames/reshapes them, this prefilter drops the line and the + // fork silently reverts to counting replayed history — update these literals + // (and the tests in test/codex-fork-history.test.js) as a conscious change. const maybeTokenCount = line.includes('"token_count"'); + const maybeBoundary = + !maybeTokenCount && line.includes('"inter_agent_communication_metadata"'); const maybeTurnContext = !maybeTokenCount && + !maybeBoundary && (line.includes('"turn_context"') || line.includes('"session_meta"')) && - (line.includes('"model"') || line.includes('"cwd"')); - if (!maybeTokenCount && !maybeTurnContext) continue; + (line.includes('"model"') || line.includes('"cwd"') || line.includes('"thread_spawn"')); + if (!maybeTokenCount && !maybeTurnContext && !maybeBoundary) continue; let obj; try { @@ -825,11 +853,27 @@ async function parseRolloutFile({ continue; } + // Boundary between replayed parent history and genuine child turns. + if (obj?.type === "inter_agent_communication_metadata") { + accounting = true; + if (codexFork === "replay") codexFork = "child"; + continue; + } + if ( (obj?.type === "turn_context" || obj?.type === "session_meta") && obj?.payload && typeof obj.payload === "object" ) { + if ( + obj.type === "session_meta" && + obj.payload.source?.subagent?.thread_spawn && + codexFork !== "child" + ) { + // Fork detected before any boundary in this file: enter replay mode. + codexFork = "replay"; + accounting = false; + } if (typeof obj.payload.model === "string") { model = obj.payload.model; } @@ -864,6 +908,17 @@ async function parseRolloutFile({ const totalUsage = info.total_token_usage; const delta = pickDelta(lastUsage, totalUsage, totals); + if (!accounting) { + // Replay mode (subagent fork before the boundary): advance the delta + // baseline so the first child delta is measured from the last replayed + // cumulative, but do not attribute the replayed parent history to this + // child. Advance unconditionally here — pickDelta can return null for a + // zero/duplicate delta, and we still want the baseline to track the + // replay. This branch is scoped to forks so the shared non-fork path + // below keeps its original guard order untouched. + if (totalUsage && typeof totalUsage === "object") totals = totalUsage; + continue; + } if (!delta) continue; delta.conversation_count = 1; @@ -891,7 +946,30 @@ async function parseRolloutFile({ eventsAggregated += 1; } - return { endOffset, lastTotal: totals, lastModel: model, eventsAggregated }; + // Fail closed for a partially-written fork: the boundary never arrived, so + // the replay is not yet safely separable from the child. Do not advance the + // cursor or the persisted baseline — the completed file is re-read next pass. + // This degrades safely (0 tokens, never over-count) but is silent, so surface + // it under debug: a file that stays here across many passes is a boundary + // that was malformed/renamed, not merely still being written. + if (codexFork === "replay") { + if (process.env.TOKENTRACKER_DEBUG) { + process.stderr.write( + `[codex-fork] ${filePath}: subagent fork has no ` + + `inter_agent_communication_metadata boundary yet; deferring ` + + `(0 tokens ingested, cursor held at ${startOffset})\n`, + ); + } + return { + endOffset: startOffset, + lastTotal, + lastModel: model, + eventsAggregated: 0, + codexFork: "replay", + }; + } + + return { endOffset, lastTotal: totals, lastModel: model, eventsAggregated, codexFork }; } async function parseClaudeFile({ diff --git a/test/codex-fork-history.test.js b/test/codex-fork-history.test.js new file mode 100644 index 00000000..5e40016d --- /dev/null +++ b/test/codex-fork-history.test.js @@ -0,0 +1,328 @@ +// Regression tests for issue #75 — Codex subagent fork history overcount. +// +// A spawned Codex subagent rollout begins with +// session_meta.payload.source.subagent.thread_spawn, then REPLAYS the parent +// thread's token_count history (cumulative total_token_usage counters that +// keep climbing), then a deterministic boundary record +// (type: "inter_agent_communication_metadata"), then the genuine child turns. +// +// The cumulative counter is CONTINUOUS across the boundary (verified against +// real local rollouts): the last replayed cumulative and the first child +// cumulative differ only by the child's genuine first turn. So the correct +// accounting is baseline-tracking: advance the delta baseline through the +// replay WITHOUT accumulating, flip accumulation on at the boundary, and let +// the first child delta be (firstChildCumulative - lastReplayCumulative). +// +// These tests use synthetic rollouts whose structure mirrors the verified +// real files. No network, no external fixtures — strictly local. + +const assert = require("node:assert/strict"); +const os = require("node:os"); +const path = require("node:path"); +const fs = require("node:fs/promises"); +const { test } = require("node:test"); + +const { parseRolloutIncremental } = require("../src/lib/rollout"); +const { parseCodexRolloutFile } = require("../src/lib/codex-rollout-parser"); + +// --- builders --------------------------------------------------------------- + +function forkSessionMetaLine({ cwd = "/work/child", parent = "parent-thread-1" } = {}) { + return JSON.stringify({ + type: "session_meta", + timestamp: "2026-07-20T10:51:35.000Z", + payload: { + id: "child-thread-1", + cwd, + model_provider: "openai", + source: { subagent: { thread_spawn: { parent_thread_id: parent, depth: 1 } } }, + }, + }); +} + +function normalSessionMetaLine({ cwd = "/work/main" } = {}) { + return JSON.stringify({ + type: "session_meta", + timestamp: "2026-07-20T10:51:35.000Z", + payload: { id: "main-thread-1", cwd, model_provider: "openai" }, + }); +} + +function boundaryLine({ ts = "2026-07-20T10:52:00.000Z" } = {}) { + return JSON.stringify({ + type: "inter_agent_communication_metadata", + timestamp: ts, + payload: { trigger_turn: true }, + }); +} + +// cumulative is the total_token_usage object (Codex reports input inclusive of +// cached). last is that turn's last_token_usage (present in real data). +function tokenCountLine({ ts, last, cumulative }) { + return JSON.stringify({ + type: "event_msg", + timestamp: ts, + payload: { + type: "token_count", + info: { last_token_usage: last, total_token_usage: cumulative }, + }, + }); +} + +function u(input, cached, output) { + return { + input_tokens: input, + cached_input_tokens: cached, + output_tokens: output, + reasoning_output_tokens: 0, + total_tokens: input + output, + }; +} + +async function readJsonLines(filePath) { + const raw = await fs.readFile(filePath, "utf8").catch(() => ""); + return raw + .split("\n") + .map((l) => l.trim()) + .filter(Boolean) + .map((l) => JSON.parse(l)); +} + +// The queue is a running-total log keyed by (source, model, hour_start): each +// incremental run re-emits the FULL bucket total, so the LATEST entry per +// bucket is authoritative. Sum a field across the latest entry of each bucket. +function sumField(rows, field) { + const latest = new Map(); + for (const r of rows) { + latest.set(`${r.source}|${r.model}|${r.hour_start}`, r); + } + return Array.from(latest.values()).reduce((s, r) => s + Number(r[field] || 0), 0); +} + +// Replay cumulative history (cached-heavy, climbing) shared by fork scenarios. +// Last replayed cumulative = input 600 / cached 560 / output 60 (total 660). +const REPLAY = [ + tokenCountLine({ ts: "2026-07-20T10:51:40.000Z", last: u(100, 90, 10), cumulative: u(100, 90, 10) }), + tokenCountLine({ ts: "2026-07-20T10:51:45.000Z", last: u(200, 190, 20), cumulative: u(300, 280, 30) }), + tokenCountLine({ ts: "2026-07-20T10:51:50.000Z", last: u(300, 280, 30), cumulative: u(600, 560, 60) }), +]; +// Genuine child turns, continuing from the replayed baseline (continuous). +// C1 cumulative 650/600/70; C2 cumulative 700/640/90. +const CHILD = [ + tokenCountLine({ ts: "2026-07-20T10:52:05.000Z", last: u(50, 40, 10), cumulative: u(650, 600, 70) }), + tokenCountLine({ ts: "2026-07-20T10:52:10.000Z", last: u(50, 40, 20), cumulative: u(700, 640, 90) }), +]; +// Expected genuine child accounting (delta of cumulatives, then normalized so +// input excludes cached): C1 delta input50/cached40/out10 -> input10; C2 delta +// input50/cached40/out20 -> input10. Totals: cached 80, output 30, total 130. +const EXPECT_CHILD = { cached_input_tokens: 80, output_tokens: 30, total_tokens: 130 }; + +async function withTmp(fn) { + const tmp = await fs.mkdtemp(path.join(os.tmpdir(), "tt-fork-")); + try { + return await fn(tmp); + } finally { + await fs.rm(tmp, { recursive: true, force: true }); + } +} + +// --- 1. normal (non-fork) rollout is unaffected ----------------------------- + +test("#75 normal rollout counts every token_count as before", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + const lines = [normalSessionMetaLine(), ...REPLAY]; // here REPLAY records are genuine + await fs.writeFile(rolloutPath, lines.join("\n") + "\n", "utf8"); + + const res = await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + const queued = await readJsonLines(queuePath); + // Full cumulative counted (cached climbs to 560). + assert.equal(sumField(queued, "cached_input_tokens"), 560); + assert.equal(sumField(queued, "total_tokens"), 660); + assert.equal(res.eventsAggregated, 3); + }); +}); + +// --- 1b. non-fork null-delta record must not re-baseline the cumulative ----- +// Guards the CRITICAL from review: the fork fix must not change the shared +// non-fork path. A transient cumulative dip (corrupt/out-of-order record) makes +// pickDelta return null with a valid totalUsage; the baseline must NOT advance +// down to the dip, or the recovery record would over-count everything since. + +test("#75 non-fork cumulative dip does not re-baseline (no over-count on recovery)", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + const lines = [ + normalSessionMetaLine(), + tokenCountLine({ ts: "2026-07-20T10:51:40.000Z", last: u(1000, 0, 0), cumulative: u(1000, 0, 0) }), + // Dip: all-zero cumulative -> pickDelta returns null (valid totalUsage). + tokenCountLine({ ts: "2026-07-20T10:51:45.000Z", last: u(0, 0, 0), cumulative: u(0, 0, 0) }), + // Recovery: baseline must still be 1000, so this delta is 10, not 1010. + tokenCountLine({ ts: "2026-07-20T10:51:50.000Z", last: u(10, 0, 0), cumulative: u(1010, 0, 0) }), + ]; + await fs.writeFile(rolloutPath, lines.join("\n") + "\n", "utf8"); + + await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + const queued = await readJsonLines(queuePath); + // 1000 (first turn) + 10 (recovery delta). A downward re-baseline gives 2010. + assert.equal(sumField(queued, "total_tokens"), 1010); + }); +}); + +// --- 2. complete fork: replay excluded, only child counted ------------------ + +test("#75 complete subagent fork counts only post-boundary child turns", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + const lines = [forkSessionMetaLine(), ...REPLAY, boundaryLine(), ...CHILD]; + await fs.writeFile(rolloutPath, lines.join("\n") + "\n", "utf8"); + + await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + const queued = await readJsonLines(queuePath); + assert.equal( + sumField(queued, "cached_input_tokens"), + EXPECT_CHILD.cached_input_tokens, + "replayed parent cached history must not be counted", + ); + assert.equal(sumField(queued, "total_tokens"), EXPECT_CHILD.total_tokens); + // Fork resolved -> persisted state is child. + assert.equal(cursors.files[rolloutPath].codexFork, "child"); + }); +}); + +// --- 3. partial fork (no boundary yet): fail closed, cursor not advanced ----- + +test("#75 partial fork without boundary ingests nothing and does not advance cursor", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + const partial = [forkSessionMetaLine(), ...REPLAY]; // boundary not written yet + await fs.writeFile(rolloutPath, partial.join("\n") + "\n", "utf8"); + + const res = await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + const queued = await readJsonLines(queuePath); + assert.equal(queued.length, 0, "no genuine child turns yet"); + assert.equal(res.eventsAggregated, 0); + // Fail closed: cursor stays at 0 so the completed file is re-read later. + assert.equal(cursors.files[rolloutPath].offset, 0); + assert.equal(cursors.files[rolloutPath].codexFork, "replay"); + }); +}); + +// --- 4. resume: boundary + child appended after a partial read -------------- + +test("#75 fork resumes correctly once the boundary is written", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + + // Pass 1: partial (replay only). + await fs.writeFile(rolloutPath, [forkSessionMetaLine(), ...REPLAY].join("\n") + "\n", "utf8"); + await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + assert.equal((await readJsonLines(queuePath)).length, 0); + + // Pass 2: boundary + child now present (full file rewritten as it grows). + await fs.writeFile( + rolloutPath, + [forkSessionMetaLine(), ...REPLAY, boundaryLine(), ...CHILD].join("\n") + "\n", + "utf8", + ); + await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + + const queued = await readJsonLines(queuePath); + assert.equal(sumField(queued, "cached_input_tokens"), EXPECT_CHILD.cached_input_tokens); + assert.equal(sumField(queued, "total_tokens"), EXPECT_CHILD.total_tokens); + assert.equal(cursors.files[rolloutPath].codexFork, "child"); + }); +}); + +// --- 5. incremental resume after boundary (child grows across reads) -------- + +test("#75 child turns appended in a later incremental read are counted once", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + + // Pass 1: session_meta + replay + boundary + first child turn only. + await fs.writeFile( + rolloutPath, + [forkSessionMetaLine(), ...REPLAY, boundaryLine(), CHILD[0]].join("\n") + "\n", + "utf8", + ); + await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + assert.equal(cursors.files[rolloutPath].codexFork, "child"); + + // Pass 2: append the second child turn (append-only growth). + await fs.appendFile(rolloutPath, CHILD[1] + "\n", "utf8"); + await parseRolloutIncremental({ rolloutFiles: [rolloutPath], cursors, queuePath }); + + const queued = await readJsonLines(queuePath); + // Both child turns counted exactly once; replay still excluded. + assert.equal(sumField(queued, "cached_input_tokens"), EXPECT_CHILD.cached_input_tokens); + assert.equal(sumField(queued, "total_tokens"), EXPECT_CHILD.total_tokens); + }); +}); + +// --- 7. context-breakdown drill-down parser excludes replay ----------------- + +test("#75 context-breakdown parser attributes only child turns for a fork", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const lines = [forkSessionMetaLine(), ...REPLAY, boundaryLine(), ...CHILD]; + await fs.writeFile(rolloutPath, lines.join("\n") + "\n", "utf8"); + + const res = await parseCodexRolloutFile(rolloutPath, {}); + // Only the two genuine child turns, measured from the boundary baseline. + assert.equal(res.totals.total_tokens, EXPECT_CHILD.total_tokens); + assert.equal(res.totals.cached_input_tokens, EXPECT_CHILD.cached_input_tokens); + assert.equal(res.turnCount, 2); + }); +}); + +// --- 7b. drill-down under a day-scoped window still excludes replay --------- +// Production calls parseCodexRolloutFile with from/to day keys. Fork-control +// records are handled before the date gate, so replay stays excluded. + +test("#75 context-breakdown parser excludes replay under a day-scoped window", async () => { + await withTmp(async (tmp) => { + const rolloutPath = path.join(tmp, "rollout.jsonl"); + const lines = [forkSessionMetaLine(), ...REPLAY, boundaryLine(), ...CHILD]; + await fs.writeFile(rolloutPath, lines.join("\n") + "\n", "utf8"); + + // All records fall on 2026-07-20; query that exact day. + const res = await parseCodexRolloutFile(rolloutPath, { from: "2026-07-20", to: "2026-07-20" }); + assert.equal(res.totals.total_tokens, EXPECT_CHILD.total_tokens); + assert.equal(res.totals.cached_input_tokens, EXPECT_CHILD.cached_input_tokens); + assert.equal(res.turnCount, 2); + }); +}); + +// --- 6. two subagents forked from the same parent are isolated -------------- + +test("#75 multiple subagent forks each count only their own child turns", async () => { + await withTmp(async (tmp) => { + const fileA = path.join(tmp, "rollout-a.jsonl"); + const fileB = path.join(tmp, "rollout-b.jsonl"); + const queuePath = path.join(tmp, "queue.jsonl"); + const cursors = { version: 1, files: {}, updatedAt: null }; + + const forkDoc = [forkSessionMetaLine(), ...REPLAY, boundaryLine(), ...CHILD].join("\n") + "\n"; + await fs.writeFile(fileA, forkDoc, "utf8"); + await fs.writeFile(fileB, forkDoc, "utf8"); + + await parseRolloutIncremental({ rolloutFiles: [fileA, fileB], cursors, queuePath }); + const queued = await readJsonLines(queuePath); + // Two children -> exactly twice the single-child totals, no replay bleed. + assert.equal(sumField(queued, "cached_input_tokens"), EXPECT_CHILD.cached_input_tokens * 2); + assert.equal(sumField(queued, "total_tokens"), EXPECT_CHILD.total_tokens * 2); + }); +});