Skip to content

fix(codex): exclude subagent fork replay history from token accounting (#75) #77

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pitimon merged 1 commit into from
Jul 21, 2026
Merged

fix(codex): exclude subagent fork replay history from token accounting (#75) #77

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 23 additions & 23 deletions openwiki-facts/source-facts.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
}
]
}
Expand Down
62 changes: 52 additions & 10 deletions src/lib/codex-rollout-parser.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -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;
Expand All @@ -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;
}

Expand All @@ -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;
}
Expand All @@ -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,
Expand Down
86 changes: 82 additions & 4 deletions src/lib/rollout.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand All @@ -155,6 +156,7 @@ async function parseRolloutIncremental({
startOffset,
lastTotal,
lastModel,
forkState,
hourlyState,
touchedBuckets,
source: fileSource,
Expand All @@ -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(),
};
Expand Down Expand Up @@ -782,6 +785,7 @@ async function parseRolloutFile({
startOffset,
lastTotal,
lastModel,
forkState,
hourlyState,
touchedBuckets,
source,
Expand All @@ -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 });
Expand All @@ -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 {
Expand All @@ -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;
}
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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({
Expand Down
Loading