From a6bf81f756203b12848b9214cbec4d90a12e8c7f Mon Sep 17 00:00:00 2001 From: ihearttokyo <164558075+ihearttokyo@users.noreply.github.com> Date: Fri, 24 Jul 2026 22:53:43 -0400 Subject: [PATCH] feat(codex): add tool-excluded active throughput metrics --- src/codex-cache.ts | 5 +- src/codex-throughput.ts | 521 +++++++++++++++++++++++++++++++++ src/dashboard.tsx | 16 +- src/main.ts | 106 +++++++ src/model-breakdown.ts | 5 + src/parser.ts | 14 + src/providers/codex.ts | 272 ++++++++++++++++- src/providers/types.ts | 3 + src/session-cache.ts | 8 +- src/types.ts | 5 +- tests/cli-codex-tps.test.ts | 46 +++ tests/codex-throughput.test.ts | 124 ++++++++ tests/providers/codex.test.ts | 216 ++++++++++++++ 13 files changed, 1326 insertions(+), 15 deletions(-) create mode 100644 src/codex-throughput.ts create mode 100644 tests/cli-codex-tps.test.ts create mode 100644 tests/codex-throughput.test.ts diff --git a/src/codex-cache.ts b/src/codex-cache.ts index 0eb59b4a..6146e8e9 100644 --- a/src/codex-cache.ts +++ b/src/codex-cache.ts @@ -11,9 +11,10 @@ import type { ParsedProviderCall } from './providers/types.js' // v5: also attribute CLI-wrapped MCP calls (`mcp-cli call server tool`) that // Codex logs as a plain exec_command (issue #478 follow-up). Force a re-parse // so sessions cached under v4 pick up the CLI-MCP attribution. -// v6: rich-session-capture — per-call locAdded/locRemoved/editFailed from +// v6/v7: rich-session-capture — per-call locAdded/locRemoved/editFailed from // patch_apply_end. Sessions cached under v5 lack these fields; re-parse to add. -const CODEX_CACHE_VERSION = 7 +// v8: persist native MCP timing and compact invocation attribution. +const CODEX_CACHE_VERSION = 8 const CACHE_FILE = 'codex-results.json' type FileFingerprint = { mtimeMs: number; sizeBytes: number } diff --git a/src/codex-throughput.ts b/src/codex-throughput.ts new file mode 100644 index 00000000..4796206d --- /dev/null +++ b/src/codex-throughput.ts @@ -0,0 +1,521 @@ +import { open, stat } from 'node:fs/promises' +import { StringDecoder } from 'node:string_decoder' + +export type CodexThroughputPoint = { + timestamp: string + model?: string + outputTokens: number + reasoningTokens: number + generatedTokens: number + taskGeneratedTokens?: number + elapsedSeconds?: number + generatedTokensPerSecond?: number + activeDurationSeconds?: number + activeGeneratedTokensPerSecond?: number + toolWaitSeconds?: number +} + +type TokenUsage = { + output_tokens?: number + reasoning_output_tokens?: number + total_tokens?: number +} + +type RolloutLine = { + type?: string + timestamp?: string + payload?: { + type?: string + turn_id?: string + call_id?: string + started_at?: number + duration_ms?: number + duration?: { secs?: number; nanos?: number } | string + model?: string + forked_from_id?: string + info?: { + last_token_usage?: TokenUsage + total_token_usage?: TokenUsage + } + } +} + +const CHUNK_BYTES = 64 * 1024 +const MAX_PENDING_LINE_CHARS = 4 * 1024 * 1024 +const TRUNCATION_MARKER = '__CODEBURN_TRUNCATED_LINE__' + +function rawString(source: string, field: string): string | undefined { + const match = new RegExp(`"${field}"\\s*:\\s*"((?:\\\\.|[^"\\\\])*)"`).exec(source) + if (!match) return undefined + try { return JSON.parse(`"${match[1]}"`) as string } catch { return undefined } +} + +function rawNumber(source: string, field: string): number | undefined { + const match = new RegExp(`"${field}"\\s*:\\s*(-?\\d+(?:\\.\\d+)?)`).exec(source) + if (!match) return undefined + const value = Number(match[1]) + return Number.isFinite(value) ? value : undefined +} + +function compactUsage(source: string, field: 'last_token_usage' | 'total_token_usage'): TokenUsage | undefined { + const index = source.indexOf(`"${field}"`) + if (index < 0) return undefined + const body = source.slice(index, index + 4096) + return { + output_tokens: rawNumber(body, 'output_tokens'), + reasoning_output_tokens: rawNumber(body, 'reasoning_output_tokens'), + total_tokens: rawNumber(body, 'total_tokens'), + } +} + +function parseRawDurationValue(value: string): number | undefined { + const objectMatch = /^\s*\{\s*"secs"\s*:\s*(-?\d+(?:\.\d+)?)\s*,\s*"nanos"\s*:\s*(-?\d+(?:\.\d+)?)\s*\}/.exec(value) + if (objectMatch) { + const seconds = Number(objectMatch[1]) + const nanos = Number(objectMatch[2]) + if (Number.isFinite(seconds) && Number.isFinite(nanos)) return seconds * 1000 + nanos / 1e6 + } + const stringMatch = /^\s*"(\d+(?:\.\d+)?)(ms|s)?"/.exec(value) + if (stringMatch) { + const parsed = Number(stringMatch[1]) + if (Number.isFinite(parsed)) return parsed * (stringMatch[2] === 's' ? 1000 : 1) + } + const numberMatch = /^\s*(-?\d+(?:\.\d+)?)/.exec(value) + if (numberMatch) { + const parsed = Number(numberMatch[1]) + if (Number.isFinite(parsed)) return parsed + } + return undefined +} + +function durationMs(payload: RolloutLine['payload']): number | undefined { + if (!payload) return undefined + if (typeof payload.duration_ms === 'number' && Number.isFinite(payload.duration_ms)) return payload.duration_ms + if (typeof payload.duration === 'object' && payload.duration) { + const seconds = payload.duration.secs + const nanos = payload.duration.nanos + if (typeof seconds === 'number' && typeof nanos === 'number' && Number.isFinite(seconds) && Number.isFinite(nanos)) { + return seconds * 1000 + nanos / 1e6 + } + } + if (typeof payload.duration === 'string') { + const match = /^(\d+(?:\.\d+)?)(ms|s)?$/.exec(payload.duration.trim()) + if (match) return Number(match[1]) * (match[2] === 's' ? 1000 : 1) + } + return undefined +} + +function mergeToolIntervals(intervals: Array<[number, number]>, durationMs: number, taskStartedAt?: number, taskCompletedAt?: number): number { + const windowStart = taskStartedAt ?? (taskCompletedAt !== undefined ? taskCompletedAt - durationMs : undefined) + const windowEnd = windowStart !== undefined ? windowStart + durationMs : undefined + const clipped = intervals.map(([start, end]) => [ + windowStart !== undefined ? Math.max(start, windowStart) : start, + windowEnd !== undefined ? Math.min(end, windowEnd) : end, + ] as [number, number]).filter(([start, end]) => end > start) + const merged = clipped.sort((a, b) => a[0] - b[0]).reduce>((result, interval) => { + const previous = result.at(-1) + if (previous && interval[0] <= previous[1]) previous[1] = Math.max(previous[1], interval[1]) + else result.push([...interval]) + return result + }, []) + return Math.min(durationMs, merged.reduce((total, [start, end]) => total + end - start, 0)) +} + +function parseLine(line: string): RolloutLine | null { + const payloadStart = line.indexOf('"payload"') + const payloadHead = payloadStart >= 0 ? line.slice(payloadStart) : line + if (line.length > 256 * 1024 || line.startsWith(TRUNCATION_MARKER)) { + const payloadType = rawString(payloadHead, 'type') + const infoStart = payloadHead.indexOf('"info"') + const info = infoStart >= 0 ? payloadHead.slice(infoStart) : '' + return { + type: rawString(line, 'type'), + timestamp: rawString(line, 'timestamp'), + payload: { + type: payloadType, + turn_id: rawString(payloadHead, 'turn_id'), + call_id: rawString(payloadHead, 'call_id'), + started_at: rawNumber(payloadHead, 'started_at'), + duration_ms: rawNumber(payloadHead, 'duration_ms'), + duration: rawString(payloadHead, 'duration') ?? (rawNumber(payloadHead, 'secs') !== undefined + ? { secs: rawNumber(payloadHead, 'secs'), nanos: rawNumber(payloadHead, 'nanos') } + : undefined), + model: rawString(payloadHead, 'model'), + forked_from_id: rawString(payloadHead, 'forked_from_id'), + info: { + last_token_usage: compactUsage(info, 'last_token_usage'), + total_token_usage: compactUsage(info, 'total_token_usage'), + }, + }, + } + } + try { + return JSON.parse(line) as RolloutLine + } catch { + return null + } +} + +/** + * Estimate generated tokens/sec from a Codex rollout's persisted checkpoints. + * Codex JSONL has no per-token timestamps, so this is deliberately a + * checkpoint-to-checkpoint estimate, not live decode speed. + */ +type ThroughputState = { + model?: string + previousTotal?: number + previousOutput: number + previousReasoning: number + previousTimestamp?: number + currentTaskGenerated: number + currentTaskToolIntervals: Array<[number, number]> + currentTaskStartedAt?: number + toolStarts: Map + latestPoint?: CodexThroughputPoint + points: CodexThroughputPoint[] + forkCutoffMs?: number +} + +function newThroughputState(): ThroughputState { + return { + previousOutput: 0, + previousReasoning: 0, + currentTaskGenerated: 0, + currentTaskToolIntervals: [], + toolStarts: new Map(), + points: [], + } +} + +/** + * Incrementally parses a rollout. Watch mode feeds only newly appended bytes + * to this reader, so a growing JSONL file is not reparsed from byte zero. + */ +export class CodexThroughputReader { + private offset = 0 + private pending = '' + private decoder = new StringDecoder('utf8') + private pendingDurationMs: number | undefined + private scanDepth = 0 + private scanPayloadDepth: number | undefined + private scanInString = false + private scanEscape = false + private scanString = '' + private scanLastString = '' + private scanAwaitingColon = false + private scanCurrentKey: string | undefined + private scanCapture: { mode: 'string' | 'object' | 'primitive'; text: string; depth: number } | undefined + private state = newThroughputState() + + reset(): void { + this.offset = 0 + this.pending = '' + this.decoder = new StringDecoder('utf8') + this.pendingDurationMs = undefined + this.scanDepth = 0 + this.scanPayloadDepth = undefined + this.scanInString = false + this.scanEscape = false + this.scanString = '' + this.scanLastString = '' + this.scanAwaitingColon = false + this.scanCurrentKey = undefined + this.scanCapture = undefined + this.state = newThroughputState() + } + + private finishDurationCapture(): void { + if (!this.scanCapture) return + const value = this.scanCapture.mode === 'string' ? `"${this.scanCapture.text}"` : this.scanCapture.text + const parsed = parseRawDurationValue(value) + if (parsed !== undefined && this.pendingDurationMs === undefined) this.pendingDurationMs = parsed + this.scanCapture = undefined + } + + private scanDurationSegment(source: string): void { + for (let i = 0; i < source.length; i++) { + const char = source[i]! + if (this.scanInString) { + if (this.scanEscape) { + this.scanEscape = false + if (this.scanCapture?.mode === 'object') this.scanCapture.text += char + else if (this.scanCapture?.mode === 'string') this.scanCapture.text += char + else this.scanString += char + continue + } + if (char === '\\') { + this.scanEscape = true + if (this.scanCapture?.mode === 'object' || this.scanCapture?.mode === 'string') this.scanCapture.text += char + continue + } + if (char === '"') { + if (this.scanCapture?.mode === 'object') this.scanCapture.text += char + this.scanInString = false + if (this.scanCapture?.mode === 'string') this.finishDurationCapture() + else if (this.scanCapture?.mode === 'object') { + this.scanAwaitingColon = false + this.scanCurrentKey = undefined + } else { + this.scanLastString = this.scanString + this.scanAwaitingColon = true + } + continue + } + if (this.scanCapture?.mode === 'object' || this.scanCapture?.mode === 'string') this.scanCapture.text += char + else this.scanString += char + continue + } + + if (this.scanCapture?.mode === 'primitive') { + if (char === ',' || char === '}' || char === ']') this.finishDurationCapture() + else { this.scanCapture.text += char; continue } + } + if (this.scanAwaitingColon) { + if (/\s/.test(char)) continue + if (char === ':') { + this.scanCurrentKey = this.scanLastString + this.scanAwaitingColon = false + continue + } + this.scanAwaitingColon = false + } + if (char === '"') { + this.scanString = '' + if (this.scanCapture?.mode === 'object') this.scanCapture.text += char + if (this.scanCurrentKey === 'duration' && this.scanPayloadDepth === this.scanDepth) { + this.scanCapture = { mode: 'string', text: '', depth: this.scanDepth } + this.scanCurrentKey = undefined + } + this.scanInString = true + continue + } + if (char === '{' || char === '[') { + if (this.scanCurrentKey === 'payload' && char === '{' && this.scanPayloadDepth === undefined) { + this.scanPayloadDepth = this.scanDepth + 1 + } + if (this.scanCurrentKey === 'duration' && this.scanPayloadDepth === this.scanDepth) { + this.scanCapture = { mode: 'object', text: char, depth: this.scanDepth + 1 } + this.scanCurrentKey = undefined + } else if (this.scanCapture?.mode === 'object') { + this.scanCapture.text += char + } + this.scanDepth++ + continue + } + if (char === '}' || char === ']') { + if (this.scanCapture?.mode === 'object') this.scanCapture.text += char + this.scanDepth = Math.max(0, this.scanDepth - 1) + if (this.scanCapture?.mode === 'object' && this.scanDepth < this.scanCapture.depth) this.finishDurationCapture() + continue + } + if (this.scanCurrentKey === 'duration' && this.scanPayloadDepth === this.scanDepth && !/\s/.test(char)) { + this.scanCapture = { mode: 'primitive', text: char, depth: this.scanDepth } + this.scanCurrentKey = undefined + continue + } + if (this.scanCapture?.mode === 'object') this.scanCapture.text += char + } + } + + private processLine(line: string, durationOverride?: number): void { + const entry = parseLine(line) + if (!entry) return + if (durationOverride !== undefined && (line.startsWith(TRUNCATION_MARKER) || line.length > 256 * 1024) && entry.type === 'event_msg' && (entry.payload?.type === 'mcp_tool_call_end' || entry.payload?.type === 'task_complete')) { + entry.payload = { ...entry.payload, duration_ms: durationOverride } + } + const state = this.state + if (entry.type === 'session_meta') { + if (entry.payload?.model) state.model = entry.payload.model + if (entry.payload?.forked_from_id && entry.timestamp) { + const timestamp = Date.parse(entry.timestamp) + if (Number.isFinite(timestamp)) state.forkCutoffMs = timestamp + 5000 + } + return + } + if (entry.type === 'turn_context' && entry.payload?.model) state.model = entry.payload.model + const entryTimestamp = entry.timestamp ? Date.parse(entry.timestamp) : NaN + const isForkReplay = state.forkCutoffMs !== undefined && Number.isFinite(entryTimestamp) && entryTimestamp < state.forkCutoffMs + if (isForkReplay && ( + entry.payload?.type === 'task_started' || + entry.payload?.type === 'task_complete' || + entry.payload?.type === 'function_call' || + entry.payload?.type === 'function_call_output' || + entry.payload?.type === 'custom_tool_call' || + entry.payload?.type === 'custom_tool_call_output' || + entry.payload?.type === 'mcp_tool_call_end' || + entry.payload?.type === 'patch_apply_end' || + entry.payload?.type === 'token_count' + )) return + if (entry.type === 'event_msg' && entry.payload?.type === 'task_started') { + state.currentTaskGenerated = 0 + state.currentTaskToolIntervals = [] + const startedAt = entry.timestamp ? Date.parse(entry.timestamp) : NaN + state.currentTaskStartedAt = Number.isFinite(startedAt) ? startedAt : undefined + state.toolStarts.clear() + } + if (entry.type === 'response_item' && (entry.payload?.type === 'function_call' || entry.payload?.type === 'custom_tool_call') && entry.payload.call_id && entry.timestamp) { + const started = Date.parse(entry.timestamp) + if (Number.isFinite(started)) state.toolStarts.set(entry.payload.call_id, started) + } + if (entry.type === 'response_item' && (entry.payload?.type === 'function_call_output' || entry.payload?.type === 'custom_tool_call_output') && entry.payload.call_id && entry.timestamp) { + const ended = Date.parse(entry.timestamp) + const started = state.toolStarts.get(entry.payload.call_id) + if (started !== undefined && Number.isFinite(ended) && ended > started) state.currentTaskToolIntervals.push([started, ended]) + state.toolStarts.delete(entry.payload.call_id) + } + if (entry.type === 'event_msg' && entry.payload?.type === 'mcp_tool_call_end' && entry.timestamp) { + const ended = Date.parse(entry.timestamp) + const elapsed = durationMs(entry.payload) + if (Number.isFinite(ended) && elapsed !== undefined && elapsed > 0) state.currentTaskToolIntervals.push([ended - elapsed, ended]) + } + if (entry.type === 'event_msg' && entry.payload?.type === 'task_complete') { + const taskDurationMs = durationMs(entry.payload) + if (state.latestPoint && typeof taskDurationMs === 'number' && taskDurationMs > 0 && state.currentTaskGenerated > 0) { + state.latestPoint.taskGeneratedTokens = state.currentTaskGenerated + const completedAt = entry.timestamp ? Date.parse(entry.timestamp) : undefined + const toolWaitMs = mergeToolIntervals(state.currentTaskToolIntervals, taskDurationMs, state.currentTaskStartedAt, Number.isFinite(completedAt) ? completedAt : undefined) + const activeMs = taskDurationMs - toolWaitMs + if (activeMs > 0) { + state.latestPoint.activeDurationSeconds = activeMs / 1000 + state.latestPoint.toolWaitSeconds = toolWaitMs / 1000 + state.latestPoint.activeGeneratedTokensPerSecond = state.currentTaskGenerated / (activeMs / 1000) + } + } + } + if (entry.type !== 'event_msg' || entry.payload?.type !== 'token_count') return + const info = entry.payload.info + if (!info || !entry.timestamp) return + const last = info.last_token_usage + const total = info.total_token_usage + const cumulative = total?.total_tokens + if (cumulative !== undefined && cumulative === state.previousTotal) return + let outputTokens = last?.output_tokens ?? 0 + let reasoningTokens = last?.reasoning_output_tokens ?? 0 + if (!last && total && cumulative !== undefined && state.previousTotal !== undefined) { + outputTokens = Math.max(0, (total.output_tokens ?? 0) - state.previousOutput) + reasoningTokens = Math.max(0, (total.reasoning_output_tokens ?? 0) - state.previousReasoning) + } + if (cumulative !== undefined) { + state.previousTotal = cumulative + state.previousOutput = total?.output_tokens ?? state.previousOutput + state.previousReasoning = total?.reasoning_output_tokens ?? state.previousReasoning + } + const generatedTokens = outputTokens + reasoningTokens + if (generatedTokens <= 0) return + const timestampMs = Date.parse(entry.timestamp) + if (!Number.isFinite(timestampMs)) return + const point: CodexThroughputPoint = { + timestamp: entry.timestamp, + model: state.model, + outputTokens, + reasoningTokens, + generatedTokens, + } + state.currentTaskGenerated += generatedTokens + state.latestPoint = point + if (state.previousTimestamp !== undefined && timestampMs > state.previousTimestamp) { + const elapsedSeconds = (timestampMs - state.previousTimestamp) / 1000 + point.elapsedSeconds = elapsedSeconds + point.generatedTokensPerSecond = generatedTokens / elapsedSeconds + } + state.previousTimestamp = timestampMs + state.points.push(point) + if (state.points.length > 10000) state.points.splice(0, state.points.length - 10000) + } + + async update(filePath: string, limit = 10, finalize = false): Promise { + const info = await stat(filePath) + if (info.size < this.offset) this.reset() + const bytesToRead = info.size - this.offset + if (bytesToRead > 0) { + const file = await open(filePath, 'r') + try { + let position = this.offset + while (position < info.size) { + const buffer = Buffer.allocUnsafe(Math.min(CHUNK_BYTES, info.size - position)) + const { bytesRead } = await file.read(buffer, 0, buffer.length, position) + if (bytesRead === 0) break + position += bytesRead + this.offset = position + let chunk = this.decoder.write(buffer.subarray(0, bytesRead)) + while (chunk.length > 0) { + const newlineIndex = chunk.search(/\r?\n/) + const segment = newlineIndex >= 0 ? chunk.slice(0, newlineIndex) : chunk + this.pending += segment + this.scanDurationSegment(segment) + if (newlineIndex < 0) break + const newlineLength = chunk[newlineIndex] === '\r' ? 2 : 1 + const line = this.pending + const durationOverride = this.pendingDurationMs + this.pending = '' + this.pendingDurationMs = undefined + this.scanDepth = 0 + this.scanPayloadDepth = undefined + this.scanInString = false + this.scanEscape = false + this.scanString = '' + this.scanLastString = '' + this.scanAwaitingColon = false + this.scanCurrentKey = undefined + this.scanCapture = undefined + this.processLine(line, durationOverride) + chunk = chunk.slice(newlineIndex + newlineLength) + } + if (this.pending.length > MAX_PENDING_LINE_CHARS) { + const body = this.pending.startsWith(TRUNCATION_MARKER) + ? this.pending.slice(TRUNCATION_MARKER.length) + : this.pending + this.pending = TRUNCATION_MARKER + body.slice(0, 256 * 1024) + body.slice(-256 * 1024) + } + } + } finally { + await file.close() + } + } + if (finalize && this.pending) { + this.processLine(this.pending, this.pendingDurationMs) + this.pending = '' + this.pendingDurationMs = undefined + } + return limit > 0 ? this.state.points.slice(-limit) : this.state.points.slice() + } +} + +export async function readCodexThroughput(filePath: string, limit = 10): Promise { + return new CodexThroughputReader().update(filePath, limit, true) +} + +export async function newestCodexSession(sessions: Array<{ path: string }>): Promise { + let newest: { path: string; mtimeMs: number } | undefined + for (const session of sessions) { + try { + const info = await stat(session.path) + if (!newest || info.mtimeMs > newest.mtimeMs) newest = { path: session.path, mtimeMs: info.mtimeMs } + } catch { + // A session can disappear while Codex rotates or archives it. + } + } + return newest?.path +} + +export function renderCodexThroughput(points: CodexThroughputPoint[], filePath: string): string { + const latest = points.at(-1) + if (!latest) return `No token_count checkpoints found in ${filePath}.` + const lines = [ + 'CodeBurn Codex throughput estimate', + `Session: ${filePath}`, + `Latest checkpoint: ${latest.timestamp}`, + `Latest checkpoint tokens: ${latest.generatedTokens.toLocaleString()} (${latest.outputTokens.toLocaleString()} output + ${latest.reasoningTokens.toLocaleString()} reasoning)`, + ] + if (latest.taskGeneratedTokens !== undefined) lines.push(`Completed task total: ${latest.taskGeneratedTokens.toLocaleString()} generated tokens`) + if (latest.activeGeneratedTokensPerSecond !== undefined) { + lines.push(`Active throughput: ${latest.activeGeneratedTokensPerSecond.toFixed(1)} generated tokens/sec over ${latest.activeDurationSeconds!.toFixed(1)}s`) + lines.push(`Excluded tool wait: ${latest.toolWaitSeconds!.toFixed(1)}s`) + } else if (latest.generatedTokensPerSecond !== undefined) { + lines.push(`Checkpoint estimate: ${latest.generatedTokensPerSecond.toFixed(1)} generated tokens/sec over ${latest.elapsedSeconds!.toFixed(1)}s`) + } else { + lines.push('Throughput: unavailable (waiting for a completed turn)') + } + lines.push('Note: offline JSONL estimate; tool intervals are removed, but server/prompt latency may remain.') + return lines.join('\n') +} diff --git a/src/dashboard.tsx b/src/dashboard.tsx index 1d9b6d25..0fad8cd4 100644 --- a/src/dashboard.tsx +++ b/src/dashboard.tsx @@ -46,7 +46,10 @@ export function showEmptyState(projectCount: number, scrollableHistory: boolean, return historyProjectCount === 0 && !historyLoading } -const MIN_WIDE = 90 +// The By Model panel now carries six numeric columns. Keep panels stacked until +// each half has enough room for those columns instead of truncating Tok/s at +// ordinary 100–120 column terminals. +const MIN_WIDE = 130 const ORANGE = '#FF8C42' const DIM = '#555555' const GOLD = '#FFD700' @@ -439,6 +442,7 @@ const MODEL_COL_COST = 8 const MODEL_COL_CACHE = 7 const MODEL_COL_CALLS = 7 const MODEL_COL_ONESHOT = 7 +const MODEL_COL_TPS = 7 const MODEL_NAME_WIDTH = 14 const MIN_EDIT_TURNS_FOR_RATE = 5 @@ -448,6 +452,7 @@ function ModelBreakdown({ projects, pw, bw }: { projects: ProjectSummary[]; pw: const modelTotals = aggregateModelTotals(projects) const modelEfficiency = aggregateModelEfficiency(projects) const anyEstimated = Object.values(modelTotals).some(d => d.estimatedCostUSD > 0) + const anyActiveTiming = Object.values(modelTotals).some(d => d.activeDurationMs > 0 && d.activeGeneratedTokens > 0) const sorted = Object.entries(modelTotals).sort(([, a], [, b]) => b.costUSD - a.costUSD) const maxCost = sorted[0]?.[1]?.costUSD ?? 0 const unpriced = findUnpricedModels(Object.entries(modelTotals).map(([model, d]) => ({ @@ -459,7 +464,7 @@ function ModelBreakdown({ projects, pw, bw }: { projects: ProjectSummary[]; pw: return ( - {''.padEnd(bw + 1 + MODEL_NAME_WIDTH)}{'cost'.padStart(MODEL_COL_COST)}{'cache'.padStart(MODEL_COL_CACHE)}{'calls'.padStart(MODEL_COL_CALLS)}{'1-shot'.padStart(MODEL_COL_ONESHOT)} + {''.padEnd(bw + 1 + MODEL_NAME_WIDTH)}{'cost'.padStart(MODEL_COL_COST)}{'cache'.padStart(MODEL_COL_CACHE)}{'calls'.padStart(MODEL_COL_CALLS)}{'1-shot'.padStart(MODEL_COL_ONESHOT)}{'Tok/s'.padStart(MODEL_COL_TPS)} {sorted.map(([model, data], i) => { const totalInput = data.freshInput + data.cacheRead + data.cacheWrite const cacheHit = totalInput > 0 ? (data.cacheRead / totalInput) * 100 : 0 @@ -468,6 +473,9 @@ function ModelBreakdown({ projects, pw, bw }: { projects: ProjectSummary[]; pw: const oneShotLabel = efficiency && efficiency.editTurns >= MIN_EDIT_TURNS_FOR_RATE && efficiency.oneShotRate !== null ? `${efficiency.oneShotRate.toFixed(1)}%` : '-' + const tpsLabel = data.activeDurationMs > 0 && data.activeGeneratedTokens > 0 + ? (data.activeGeneratedTokens / (data.activeDurationMs / 1000)).toFixed(1) + : '-' return ( @@ -476,6 +484,7 @@ function ModelBreakdown({ projects, pw, bw }: { projects: ProjectSummary[]; pw: {cacheLabel.padStart(MODEL_COL_CACHE)} {String(data.calls).padStart(MODEL_COL_CALLS)} {oneShotLabel.padStart(MODEL_COL_ONESHOT)} + {tpsLabel.padStart(MODEL_COL_TPS)} ) })} @@ -487,6 +496,9 @@ function ModelBreakdown({ projects, pw, bw }: { projects: ProjectSummary[]; pw: {anyEstimated && ( ~ estimated cost (priced from estimated tokens) )} + {anyActiveTiming && ( + ~ Tok/s: generated tokens / active time; tool wait excluded + )} ) } diff --git a/src/main.ts b/src/main.ts index ae6867cf..d24789ef 100644 --- a/src/main.ts +++ b/src/main.ts @@ -5,6 +5,7 @@ import { exportCsv, exportJson, type PeriodExport } from './export.js' import { findUnpricedModels, loadPricing, setModelAliases, setPriceOverrides, setLocalModelSavings, setProxyPaths, normalizeProxyPath } from './models.js' import { parseAllSessions, filterProjectsByName, filterProjectsByDateRange, clearSessionCache, setInteractiveScanUI } from './parser.js' import { allProviderNames, getAllProviders } from './providers/index.js' +import { getProvider } from './providers/index.js' import { convertCost, formatCost } from './currency.js' import { renderStatusBar } from './format.js' import { toDateString } from './daily-cache.js' @@ -46,6 +47,7 @@ import { createRequire } from 'node:module' const require = createRequire(import.meta.url) const { version } = require('../package.json') import { loadCurrency, getCurrency, isValidCurrencyCode } from './currency.js' +import { CodexThroughputReader, newestCodexSession, renderCodexThroughput } from './codex-throughput.js' // A downstream reader that closes the pipe early (`| head`, quitting `less`, or // a missing command) makes stdout writes fail with EPIPE. Exit cleanly rather @@ -68,6 +70,22 @@ function parseInteger(value: string): number { return parseInt(value, 10) } +function parseCodexTpsLimit(value: string): number { + const parsed = Number(value) + if (!Number.isFinite(parsed) || !Number.isInteger(parsed) || parsed < 1 || parsed > 10000) { + throw new Error('limit must be an integer from 1 to 10000') + } + return parsed +} + +function parseCodexTpsWatch(value: string): number { + const parsed = Number(value) + if (!Number.isFinite(parsed) || parsed < 0 || (parsed > 0 && parsed < 1) || parsed > 3600) { + throw new Error('watch must be 0 or at least 1 second (up to 3600 seconds)') + } + return parsed +} + type PriceOverrideConfig = NonNullable[string] type PriceOverrideOptions = { @@ -1843,6 +1861,94 @@ program await runContextCommand(session, opts) }) +program + .command('codex-tps [session]') + .description('Retrospective Codex generated-tokens/sec estimate from rollout checkpoints (not live decode speed)') + .option('--json', 'JSON output') + .option('--limit ', 'Number of recent checkpoints to scan', parseCodexTpsLimit, 10) + .option('--watch ', 'Refresh continuously while Codex writes checkpoints', parseCodexTpsWatch, 0) + .action(async (session: string | undefined, opts: { json?: boolean; limit: number; watch: number }) => { + const intervalMs = Math.max(0, opts.watch) * 1000 + if (opts.json && intervalMs > 0) { + process.stderr.write('codeburn codex-tps: --json cannot be combined with --watch; use text watch output or one-shot JSON.\n') + process.exitCode = 2 + return + } + const provider = await getProvider('codex') + if (!provider) { + process.stderr.write('codeburn codex-tps: Codex provider is unavailable.\n') + process.exitCode = 1 + return + } + let cachedPath: string | undefined = session + let throughputReader: CodexThroughputReader | undefined + let lastFileState: { size: number; mtimeMs: number } | undefined + let lastDiscoveryMs = 0 + let refreshInFlight = false + const render = async (): Promise => { + if (refreshInFlight) return + refreshInFlight = true + try { + let filePath = session ?? cachedPath + // Keep an idle watcher on its chosen rollout. A full active+archive + // discovery can be hundreds of milliseconds on large histories, so + // only re-scan slowly to notice rotation; disappearance still triggers + // an immediate discovery on the next tick. + if (!session && (!filePath || Date.now() - lastDiscoveryMs >= 60_000)) { + lastDiscoveryMs = Date.now() + filePath = await newestCodexSession(await provider.discoverSessions()) + } + if (!filePath) { + process.stderr.write('codeburn codex-tps: no Codex rollout sessions found.\n') + if (intervalMs === 0) process.exitCode = 1 + return + } + const previousPath = cachedPath + cachedPath = filePath + if (previousPath !== filePath || !throughputReader) throughputReader = new CodexThroughputReader() + const fileInfo = await import('node:fs/promises').then(fs => fs.stat(filePath)).catch(() => null) + if (!fileInfo) { + process.stderr.write(`codeburn codex-tps: session file not found: ${filePath}\n`) + if (intervalMs === 0) process.exitCode = 1 + if (!session) cachedPath = undefined + return + } + if (intervalMs > 0 && lastFileState && fileInfo.size === lastFileState.size && fileInfo.mtimeMs === lastFileState.mtimeMs) return + lastFileState = { size: fileInfo.size, mtimeMs: fileInfo.mtimeMs } + const points = await throughputReader!.update(filePath, opts.limit, intervalMs === 0) + if (opts.json) { + process.stdout.write(JSON.stringify({ session: filePath, points, live: intervalMs > 0 }, null, 2) + '\n') + } else { + if (intervalMs > 0) process.stdout.write('\x1b[2J\x1b[H') + process.stdout.write(renderCodexThroughput(points, filePath) + (intervalMs > 0 ? '\nWatching for new Codex checkpoints... (Ctrl-C to stop)\n' : '\n')) + } + } finally { + refreshInFlight = false + } + } + try { + await render() + } catch (error) { + const message = error instanceof Error ? error.message : String(error) + process.stderr.write(`codeburn codex-tps: refresh failed: ${message}\n`) + if (intervalMs === 0) { + process.exitCode = 1 + return + } + } + if (intervalMs > 0) { + await new Promise((resolve) => { + const timer = setInterval(() => { + void render().catch(error => { + const message = error instanceof Error ? error.message : String(error) + process.stderr.write(`codeburn codex-tps: refresh failed: ${message}\n`) + }) + }, intervalMs) + process.once('SIGINT', () => { clearInterval(timer); resolve() }) + }) + } + }) + program .command('compare') .description('Compare two AI models side-by-side') diff --git a/src/model-breakdown.ts b/src/model-breakdown.ts index 5801be80..799338e8 100644 --- a/src/model-breakdown.ts +++ b/src/model-breakdown.ts @@ -8,6 +8,8 @@ export interface ModelTotals { freshInput: number cacheRead: number cacheWrite: number + activeDurationMs: number + activeGeneratedTokens: number } /// Aggregate per-model usage across every session, keyed by the friendly display @@ -24,6 +26,7 @@ export function aggregateModelTotals(projects: ProjectSummary[]): Record b[0 const toolNameMap: Record = { exec_command: 'Bash', + // Codex Desktop's custom-tool transport uses the shorter `exec` name for + // the same shell tool that CLI rollouts record as `exec_command`. + exec: 'Bash', read_file: 'Read', write_file: 'Edit', apply_diff: 'Edit', @@ -96,6 +99,11 @@ type CodexEntry = { timestamp?: string payload?: { type?: string + turn_id?: string + call_id?: string + started_at?: number + duration_ms?: number + duration?: { secs?: number; nanos?: number } | string role?: string cwd?: string model_provider?: string @@ -104,6 +112,7 @@ type CodexEntry = { forked_from_id?: string model?: string name?: string + invocation?: { server?: string; tool?: string } content?: Array<{ type?: string; text?: string }> info?: { model?: string @@ -196,11 +205,128 @@ function getRawJsonStringField(head: string, field: string): string | undefined } } +function getRawJsonNumberField(head: string, field: string): number | undefined { + const match = new RegExp(`"${field}"\\s*:\\s*(-?\\d+(?:\\.\\d+)?)`).exec(head) + if (!match) return undefined + const value = Number(match[1]) + return Number.isFinite(value) ? value : undefined +} + +function getRawPayloadFieldWindow(source: Buffer, field: string, windowBytes = 4096): string | undefined { + const payloadKey = Buffer.from('"payload"') + const payloadIndex = source.indexOf(payloadKey) + if (payloadIndex < 0) return undefined + let payloadStart = source.indexOf(0x7b, payloadIndex + payloadKey.length) // { + if (payloadStart < 0) return undefined + + let depth = 0 + let inString = false + let escaped = false + for (let i = payloadStart; i < source.length; i++) { + const byte = source[i]! + if (inString) { + if (escaped) escaped = false + else if (byte === 0x5c) escaped = true // \\ + else if (byte === 0x22) inString = false // " + continue + } + if (byte === 0x22) { + const keyStart = i + 1 + let keyEnd = keyStart + let keyEscaped = false + for (; keyEnd < source.length; keyEnd++) { + const keyByte = source[keyEnd]! + if (keyEscaped) { keyEscaped = false; continue } + if (keyByte === 0x5c) { keyEscaped = true; continue } + if (keyByte === 0x22) break + } + if (depth === 1 && keyEnd < source.length) { + const key = source.subarray(keyStart, keyEnd).toString('utf-8') + let valueStart = keyEnd + 1 + while (valueStart < source.length && (source[valueStart] === 0x20 || source[valueStart] === 0x09 || source[valueStart] === 0x0a || source[valueStart] === 0x0d)) valueStart++ + if (source[valueStart] === 0x3a && key === field) { + return source.subarray(i, Math.min(source.length, i + windowBytes)).toString('utf-8') + } + } + i = keyEnd + inString = false + continue + } + if (byte === 0x22) inString = true + else if (byte === 0x7b || byte === 0x5b) depth++ // { or [ + else if (byte === 0x7d || byte === 0x5d) depth-- // } or ] + if (depth < 0) break + } + return undefined +} + +function getRawDurationMs(head: string): number | undefined { + const objectMatch = /"duration"\s*:\s*\{\s*"secs"\s*:\s*(-?\d+(?:\.\d+)?)\s*,\s*"nanos"\s*:\s*(-?\d+(?:\.\d+)?)\s*\}/.exec(head) + if (objectMatch) { + const seconds = Number(objectMatch[1]) + const nanos = Number(objectMatch[2]) + if (Number.isFinite(seconds) && Number.isFinite(nanos)) return seconds * 1000 + nanos / 1e6 + } + const text = getRawJsonStringField(head, 'duration') + if (text) { + const match = /^(\d+(?:\.\d+)?)(ms|s)?$/.exec(text.trim()) + if (match) { + const value = Number(match[1]) + if (Number.isFinite(value)) return value * (match[2] === 's' ? 1000 : 1) + } + } + return undefined +} + +function durationValueMs(value: unknown): number | undefined { + if (typeof value === 'number' && Number.isFinite(value)) return value + if (typeof value === 'object' && value) { + const record = value as Record + const seconds = record['secs'] + const nanos = record['nanos'] + if (typeof seconds === 'number' && typeof nanos === 'number' && Number.isFinite(seconds) && Number.isFinite(nanos)) { + return seconds * 1000 + nanos / 1e6 + } + } + if (typeof value === 'string') { + const match = /^(\d+(?:\.\d+)?)(ms|s)?$/.exec(value.trim()) + if (match) { + const parsed = Number(match[1]) + if (Number.isFinite(parsed)) return parsed * (match[2] === 's' ? 1000 : 1) + } + } + return undefined +} + +function getRawTokenUsage(head: string, field: 'last_token_usage' | 'total_token_usage'): CodexTokenUsage | undefined { + const match = new RegExp(`"${field}"\\s*:\\s*\\{([^}]*)\\}`).exec(head) + if (!match) return undefined + const body = match[1]! + return { + input_tokens: getRawJsonNumberField(body, 'input_tokens'), + cached_input_tokens: getRawJsonNumberField(body, 'cached_input_tokens'), + output_tokens: getRawJsonNumberField(body, 'output_tokens'), + reasoning_output_tokens: getRawJsonNumberField(body, 'reasoning_output_tokens'), + total_tokens: getRawJsonNumberField(body, 'total_tokens'), + } +} + function payloadHead(head: string): string { const idx = head.indexOf('"payload"') return idx === -1 ? head : head.slice(idx) } +function getRawInvocation(head: string): { server?: string; tool?: string } | undefined { + const idx = head.indexOf('"invocation"') + if (idx === -1) return undefined + // Server/tool are shallow fields and precede the potentially huge arguments + // object in Codex MCP records. Limit this scan to keep compact parsing cheap. + const invocationHead = head.slice(idx, idx + 8192) + const server = getRawJsonStringField(invocationHead, 'server') + const tool = getRawJsonStringField(invocationHead, 'tool') + return server || tool ? { server, tool } : undefined +} + function countJsonStringBytes(source: Buffer, valueStart: number): number { let count = 0 for (let i = valueStart; i < source.length; i++) { @@ -270,6 +396,31 @@ function parseCodexLine(line: string | Buffer): CodexEntry | null { const pHead = payloadHead(head) const payloadType = getRawJsonStringField(pHead, 'type') const role = getRawJsonStringField(pHead, 'role') + // task_complete appends the potentially huge final assistant message before + // its duration fields. Fall back to the full Buffer only for this event so + // timing metadata is not lost when the compact head stops early. + const needsTimingTail = type === 'event_msg' && (payloadType === 'task_complete' || payloadType === 'mcp_tool_call_end') + const timingTail = needsTimingTail && line.length > RAW_HEAD_BYTES + ? line.subarray(Math.max(0, line.length - 16 * 1024)).toString('utf-8') + : pHead + const timingNumber = (field: string): number | undefined => + getRawJsonNumberField(pHead, field) ?? getRawJsonNumberField(timingTail, field) + // MCP records can place a large invocation.arguments object before duration + // and a large result after it. Searching a small window around the field + // avoids materializing the middle of the Buffer while still preserving wait + // timing for those records. + const payloadDuration = payloadType === 'mcp_tool_call_end' + ? getRawDurationMs(getRawPayloadFieldWindow(line, 'duration') ?? '') + : undefined + const timingDuration = payloadDuration ?? getRawDurationMs(pHead) ?? getRawDurationMs(timingTail) + const compactModel = getRawJsonStringField(pHead, 'model') + const compactModelName = getRawJsonStringField(pHead, 'model_name') + const compactLastUsage = getRawTokenUsage(pHead, 'last_token_usage') + const compactTotalUsage = getRawTokenUsage(pHead, 'total_token_usage') + const compactInfo = compactModel || compactModelName || compactLastUsage || compactTotalUsage + ? { model: compactModel, model_name: compactModelName, last_token_usage: compactLastUsage, total_token_usage: compactTotalUsage } + : undefined + const invocation = getRawInvocation(pHead) ?? getRawInvocation(timingTail) const entry: CodexEntry = { type, @@ -284,6 +435,12 @@ function parseCodexLine(line: string | Buffer): CodexEntry | null { forked_from_id: getRawJsonStringField(pHead, 'forked_from_id'), model: getRawJsonStringField(pHead, 'model'), name: getRawJsonStringField(pHead, 'name'), + invocation, + call_id: getRawJsonStringField(pHead, 'call_id'), + turn_id: getRawJsonStringField(pHead, 'turn_id'), + duration_ms: timingNumber('duration_ms') ?? timingDuration, + started_at: timingNumber('started_at'), + info: compactInfo, }, } @@ -296,26 +453,48 @@ function parseCodexLine(line: string | Buffer): CodexEntry | null { return entry } -async function discoverSessionFile(filePath: string): Promise { +type DiscoveredCodexSession = { + source: SessionSource + sessionId?: string +} + +async function discoverSessionFile(filePath: string): Promise { const s = await stat(filePath).catch(() => null) if (!s?.isFile()) return null const cachedProject = await getCachedCodexProject(filePath) + const { valid, meta } = await isValidCodexSession(filePath) if (cachedProject) { - return { path: filePath, project: cachedProject, provider: 'codex' } + return { + source: { path: filePath, project: cachedProject, provider: 'codex' }, + sessionId: valid ? meta?.payload?.session_id : undefined, + } } - const { valid, meta } = await isValidCodexSession(filePath) if (!valid || !meta) return null const cwd = meta.payload?.cwd ?? 'unknown' - return { path: filePath, project: sanitizeProject(cwd), provider: 'codex' } + return { + source: { path: filePath, project: sanitizeProject(cwd), provider: 'codex' }, + sessionId: meta.payload?.session_id, + } } async function discoverSessionsInDir(codexDir: string): Promise { const sources: SessionSource[] = [] + // A rollout can exist in both roots during/after archiving. The active root + // is scanned first, and session_id keeps the archived copy from resurfacing. + const seenSessionIds = new Set() const sessionsDir = join(codexDir, 'sessions') + const addSession = (discovered: DiscoveredCodexSession | null): void => { + if (!discovered) return + const sessionId = discovered.sessionId?.trim() + if (sessionId && seenSessionIds.has(sessionId)) return + if (sessionId) seenSessionIds.add(sessionId) + sources.push(discovered.source) + } + const years = await readdir(sessionsDir).catch(() => [] as string[]) for (const year of years) { @@ -336,8 +515,7 @@ async function discoverSessionsInDir(codexDir: string): Promise for (const file of files) { if (!file.startsWith('rollout-') || !file.endsWith('.jsonl')) continue const filePath = join(dayDir, file) - const source = await discoverSessionFile(filePath) - if (source) sources.push(source) + addSession(await discoverSessionFile(filePath)) } } } @@ -349,8 +527,7 @@ async function discoverSessionsInDir(codexDir: string): Promise const archivedFiles = await readdir(archivedDir).catch(() => [] as string[]) for (const file of archivedFiles) { if (!file.startsWith('rollout-') || !file.endsWith('.jsonl')) continue - const source = await discoverSessionFile(join(archivedDir, file)) - if (source) sources.push(source) + addSession(await discoverSessionFile(join(archivedDir, file))) } return sources @@ -410,6 +587,11 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars let currentTurnId = `${sessionId}:t0` let sawAnyLine = false const results: ParsedProviderCall[] = [] + let taskResultStart = 0 + let taskGeneratedTokens = 0 + let taskToolIntervals: Array<[number, number]> = [] + let taskStartedAt: number | undefined + const openToolStarts = new Map() // Stream the session file line by line. Heavy Codex sessions can exceed // 250 MB on disk; reading the entire file into a string would either hit @@ -437,7 +619,29 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars continue } - if (entry.type === 'response_item' && entry.payload?.type === 'function_call') { + const isForkReplay = Boolean(forkCutoff && entry.timestamp && entry.timestamp < forkCutoff) + if (isForkReplay && ( + entry.payload?.type === 'task_started' || + entry.payload?.type === 'task_complete' || + entry.payload?.type === 'function_call' || + entry.payload?.type === 'function_call_output' || + entry.payload?.type === 'custom_tool_call' || + entry.payload?.type === 'custom_tool_call_output' || + entry.payload?.type === 'mcp_tool_call_end' || + entry.payload?.type === 'patch_apply_end' + )) continue + + if (entry.type === 'event_msg' && entry.payload?.type === 'task_started') { + taskResultStart = results.length + taskGeneratedTokens = 0 + taskToolIntervals = [] + const startedAt = entry.timestamp ? Date.parse(entry.timestamp) : NaN + taskStartedAt = Number.isFinite(startedAt) ? startedAt : undefined + openToolStarts.clear() + continue + } + + if (entry.type === 'response_item' && (entry.payload?.type === 'function_call' || entry.payload?.type === 'custom_tool_call')) { const rawName = entry.payload.name ?? '' const mapped = toolNameMap[rawName] ?? rawName pendingTools.push(mapped) @@ -459,10 +663,53 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars pendingToolSequence.push([{ tool: mcpTool }]) } } + const callId = entry.payload.call_id + const started = entry.timestamp ? Date.parse(entry.timestamp) : NaN + if (callId && Number.isFinite(started)) openToolStarts.set(callId, started) pendingToolSequence.push([call]) continue } + if (entry.type === 'response_item' && (entry.payload?.type === 'function_call_output' || entry.payload?.type === 'custom_tool_call_output')) { + const callId = entry.payload.call_id + const ended = entry.timestamp ? Date.parse(entry.timestamp) : NaN + const started = callId ? openToolStarts.get(callId) : undefined + if (started !== undefined && Number.isFinite(ended) && ended > started) taskToolIntervals.push([started, ended]) + if (callId) openToolStarts.delete(callId) + continue + } + + if (entry.type === 'event_msg' && entry.payload?.type === 'task_complete') { + const durationMs = entry.payload.duration_ms + if (typeof durationMs === 'number' && durationMs > 0 && taskGeneratedTokens > 0 && taskResultStart < results.length) { + const completedAt = entry.timestamp ? Date.parse(entry.timestamp) : NaN + const windowStart = taskStartedAt ?? (Number.isFinite(completedAt) ? completedAt - durationMs : undefined) + const windowEnd = windowStart !== undefined ? windowStart + durationMs : undefined + const clipped = taskToolIntervals.map(([start, end]) => [ + windowStart !== undefined ? Math.max(start, windowStart) : start, + windowEnd !== undefined ? Math.min(end, windowEnd) : end, + ] as [number, number]).filter(([start, end]) => end > start) + const merged = clipped.sort((a, b) => a[0] - b[0]).reduce>((acc, interval) => { + const previous = acc.at(-1) + if (previous && interval[0] <= previous[1]) previous[1] = Math.max(previous[1], interval[1]) + else acc.push([...interval]) + return acc + }, []) + const toolWaitMs = Math.min(durationMs, merged.reduce((sum, interval) => sum + interval[1] - interval[0], 0)) + const activeMs = durationMs - toolWaitMs + if (activeMs <= 0) continue + for (let i = taskResultStart; i < results.length; i++) { + const call = results[i]! + const generated = call.outputTokens + call.reasoningTokens + if (generated <= 0) continue + call.activeGeneratedTokens = generated + call.activeDurationMs = activeMs * (generated / taskGeneratedTokens) + call.toolWaitMs = toolWaitMs * (generated / taskGeneratedTokens) + } + } + continue + } + if (entry.type === 'event_msg' && entry.payload?.type === 'patch_apply_end') { pendingTools.push('Edit') const p = entry.payload as Record @@ -490,6 +737,11 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars // attributed. Rebuild the canonical `mcp____` name the // classifier recognizes. if (entry.type === 'event_msg' && entry.payload?.type === 'mcp_tool_call_end') { + const endedAt = entry.timestamp ? Date.parse(entry.timestamp) : NaN + const durationMs = entry.payload.duration_ms ?? durationValueMs(entry.payload.duration) + if (typeof durationMs === 'number' && durationMs > 0 && Number.isFinite(endedAt)) { + taskToolIntervals.push([endedAt - durationMs, endedAt]) + } const inv = (entry.payload as Record)['invocation'] as Record | undefined const server = typeof inv?.['server'] === 'string' ? inv['server'] as string : '' const tool = typeof inv?.['tool'] === 'string' ? inv['tool'] as string : '' @@ -568,6 +820,7 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars ...(pendingLocRemoved ? { locRemoved: pendingLocRemoved } : {}), ...(pendingEditFailed ? { editFailed: pendingEditFailed } : {}), }) + taskGeneratedTokens += estOutput pendingTools = [] pendingToolSequence = [] @@ -685,6 +938,7 @@ function createParser(source: SessionSource, seenKeys: Set): SessionPars ...(pendingLocRemoved ? { locRemoved: pendingLocRemoved } : {}), ...(pendingEditFailed ? { editFailed: pendingEditFailed } : {}), }) + taskGeneratedTokens += outputTokens + reasoningTokens pendingTools = [] pendingToolSequence = [] diff --git a/src/providers/types.ts b/src/providers/types.ts index b5396a20..8f9e902c 100644 --- a/src/providers/types.ts +++ b/src/providers/types.ts @@ -53,6 +53,9 @@ export type ParsedProviderCall = { // Exact provider-recorded cwd, kept separately because projectPath may later // canonicalize a linked worktree to its main repository. workingDirectory?: string + activeDurationMs?: number + activeGeneratedTokens?: number + toolWaitMs?: number } // A directory or database file that a provider's discoverSessions() scans. diff --git a/src/session-cache.ts b/src/session-cache.ts index 461dd461..2682756f 100644 --- a/src/session-cache.ts +++ b/src/session-cache.ts @@ -51,6 +51,9 @@ export type CachedCall = { toolErrors?: number // Codex: count of this call's patch applications with success === false. editFailed?: number + activeDurationMs?: number + activeGeneratedTokens?: number + toolWaitMs?: number } export type CachedTurn = { @@ -213,7 +216,7 @@ export const PROVIDER_PARSE_VERSIONS: Record = { // rich-session-capture-v1: per-call LOC deltas + editFailed from // patch_apply_end. (The codex-results.json CODEX_CACHE_VERSION is bumped in // lockstep so the pre-session-cache layer re-parses too.) - codex: 'mcp-attribution-v2-est-cost-rich-capture-v1-cross-provider-pr-v1', + codex: 'mcp-attribution-v5-est-cost-active-timing-mcp-wait-rich-capture-v1-cross-provider-pr-v1', cursor: 'composer-anchored-crediting-v1-est-cost', 'cursor-agent': 'workspaceless-transcript-v1', copilot: 'cli-shutdown-cost-v1-skills', @@ -337,6 +340,9 @@ function validateCall(c: unknown): c is CachedCall { && (o['speed'] === 'standard' || o['speed'] === 'fast') && isOptionalNum(o['costUSD']) && isOptionalBool(o['isEstimated']) + && isOptionalNum(o['activeDurationMs']) + && isOptionalNum(o['activeGeneratedTokens']) + && isOptionalNum(o['toolWaitMs']) && isStringArray(o['tools']) && isStringArray(o['bashCommands']) && isStringArray(o['skills']) diff --git a/src/types.ts b/src/types.ts index d5624cfb..a51ae672 100644 --- a/src/types.ts +++ b/src/types.ts @@ -153,6 +153,9 @@ export type ParsedApiCall = { /// Count of this call's tool results flagged `is_error` (Claude tool_result /// blocks). Bash stderr alone is NOT counted (warnings go there). Omitted at 0. toolErrors?: number + activeDurationMs?: number + activeGeneratedTokens?: number + toolWaitMs?: number } export type ToolCall = { @@ -268,7 +271,7 @@ export type SessionSummary = { /// from a provider that never captures branches (→ contributes nothing). /// Claude only; absent otherwise. everHadBranch?: boolean - modelBreakdown: Record + modelBreakdown: Record toolBreakdown: Record mcpBreakdown: Record bashBreakdown: Record diff --git a/tests/cli-codex-tps.test.ts b/tests/cli-codex-tps.test.ts new file mode 100644 index 00000000..4b39027a --- /dev/null +++ b/tests/cli-codex-tps.test.ts @@ -0,0 +1,46 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { spawnSync } from 'node:child_process' +import { afterEach, describe, expect, it } from 'vitest' + +const homes: string[] = [] + +afterEach(async () => { + while (homes.length) await rm(homes.pop()!, { recursive: true, force: true }) +}) + +function runCli(args: string[], home: string) { + return spawnSync(process.execPath, ['--import', 'tsx', 'src/cli.ts', ...args], { + cwd: process.cwd(), + env: { ...process.env, HOME: home, CODEX_HOME: join(home, '.codex'), TZ: 'UTC' }, + encoding: 'utf-8', + timeout: 30_000, + }) +} + +describe('codex-tps CLI validation', () => { + it('rejects sub-second watch intervals', async () => { + const home = await mkdtemp(join(tmpdir(), 'codeburn-tps-cli-')) + homes.push(home) + const result = runCli(['codex-tps', '--watch', '0.1'], home) + expect(result.status).toBe(1) + expect(result.stderr).toContain('watch must be 0 or at least 1 second') + }) + + it('rejects JSON watch output instead of concatenating invalid JSON documents', async () => { + const home = await mkdtemp(join(tmpdir(), 'codeburn-tps-cli-')) + homes.push(home) + const result = runCli(['codex-tps', '--json', '--watch', '1'], home) + expect(result.status).toBe(2) + expect(result.stderr).toContain('--json cannot be combined with --watch') + }) + + it('returns a nonzero status for a missing explicit rollout', async () => { + const home = await mkdtemp(join(tmpdir(), 'codeburn-tps-cli-')) + homes.push(home) + const result = runCli(['codex-tps', join(home, 'missing.jsonl')], home) + expect(result.status).toBe(1) + expect(result.stderr).toContain('session file not found') + }) +}) diff --git a/tests/codex-throughput.test.ts b/tests/codex-throughput.test.ts new file mode 100644 index 00000000..1e0fd4da --- /dev/null +++ b/tests/codex-throughput.test.ts @@ -0,0 +1,124 @@ +import { appendFile, mkdtemp, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' +import { CodexThroughputReader, readCodexThroughput, renderCodexThroughput } from '../src/codex-throughput.js' + +describe('Codex throughput prototype', () => { + it('estimates generated tokens/sec between token_count checkpoints', async () => { + const dir = await mkdtemp(join(tmpdir(), 'codeburn-tps-')) + const path = join(dir, 'rollout.jsonl') + await writeFile(path, [ + JSON.stringify({ type: 'session_meta', timestamp: '2026-07-25T00:00:00.000Z', payload: { model: 'gpt-5.6-sol' } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:00.000Z', payload: { type: 'task_started' } }), + JSON.stringify({ type: 'response_item', timestamp: '2026-07-25T00:00:02.000Z', payload: { type: 'function_call', call_id: 'tool-1' } }), + JSON.stringify({ type: 'response_item', timestamp: '2026-07-25T00:00:05.000Z', payload: { type: 'function_call_output', call_id: 'tool-1' } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:05.000Z', payload: { type: 'mcp_tool_call_end', duration: { secs: 3, nanos: 0 } } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:10.000Z', payload: { type: 'token_count', info: { last_token_usage: { output_tokens: 80, reasoning_output_tokens: 20 }, total_token_usage: { total_tokens: 100, output_tokens: 80, reasoning_output_tokens: 20 } } } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:15.000Z', payload: { type: 'token_count', info: { last_token_usage: { output_tokens: 40, reasoning_output_tokens: 10 }, total_token_usage: { total_tokens: 150, output_tokens: 120, reasoning_output_tokens: 30 } } } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:16.000Z', payload: { type: 'task_complete', duration_ms: 10000 } }), + ].join('\n')) + + const points = await readCodexThroughput(path) + expect(points).toHaveLength(2) + expect(points[1]).toMatchObject({ generatedTokens: 50, elapsedSeconds: 5, generatedTokensPerSecond: 10, activeDurationSeconds: 7, activeGeneratedTokensPerSecond: 21.428571428571427, toolWaitSeconds: 3, model: 'gpt-5.6-sol' }) + expect(renderCodexThroughput(points, path)).toContain('21.4 generated tokens/sec') + }) + + it('parses only appended complete lines while watching a growing rollout', async () => { + const dir = await mkdtemp(join(tmpdir(), 'codeburn-tps-watch-')) + const path = join(dir, 'rollout.jsonl') + const first = JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:00.000Z', payload: { type: 'token_count', info: { last_token_usage: { output_tokens: 8, reasoning_output_tokens: 2 }, total_token_usage: { total_tokens: 10, output_tokens: 8, reasoning_output_tokens: 2 } } } }) + const second = JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:01.000Z', payload: { type: 'token_count', info: { last_token_usage: { output_tokens: 4, reasoning_output_tokens: 1 }, total_token_usage: { total_tokens: 15, output_tokens: 12, reasoning_output_tokens: 3 } } } }) + await writeFile(path, first.slice(0, 40)) + const reader = new CodexThroughputReader() + expect(await reader.update(path)).toEqual([]) + await appendFile(path, first.slice(40) + '\n' + second + '\n') + const points = await reader.update(path) + expect(points).toHaveLength(2) + expect(points[1]).toMatchObject({ generatedTokens: 5, generatedTokensPerSecond: 5 }) + }) + + it('ignores replayed pre-fork checkpoints before estimating new work', async () => { + const dir = await mkdtemp(join(tmpdir(), 'codeburn-tps-fork-')) + const path = join(dir, 'rollout.jsonl') + const line = (timestamp: string, payload: Record) => JSON.stringify({ type: 'event_msg', timestamp, payload }) + await writeFile(path, [ + JSON.stringify({ type: 'session_meta', timestamp: '2026-07-25T00:00:00.000Z', payload: { model: 'gpt-5.6-sol', forked_from_id: 'parent' } }), + line('2026-07-25T00:00:01.000Z', { type: 'task_started' }), + line('2026-07-25T00:00:02.000Z', { type: 'token_count', info: { last_token_usage: { output_tokens: 100 }, total_token_usage: { total_tokens: 100, output_tokens: 100 } } }), + line('2026-07-25T00:00:03.000Z', { type: 'task_complete', duration_ms: 1000 }), + line('2026-07-25T00:00:06.000Z', { type: 'task_started' }), + line('2026-07-25T00:00:08.000Z', { type: 'token_count', info: { last_token_usage: { output_tokens: 20 }, total_token_usage: { total_tokens: 20, output_tokens: 20 } } }), + line('2026-07-25T00:00:10.000Z', { type: 'task_complete', duration_ms: 4000 }), + ].join('\n')) + + const points = await readCodexThroughput(path) + expect(points).toHaveLength(1) + expect(points[0]).toMatchObject({ generatedTokens: 20, activeGeneratedTokensPerSecond: 5 }) + }) + + it('keeps oversized rollout lines bounded while extracting token usage', async () => { + const dir = await mkdtemp(join(tmpdir(), 'codeburn-tps-large-')) + const path = join(dir, 'rollout.jsonl') + const largeResult = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-07-25T00:00:01.000Z', + payload: { + type: 'token_count', + info: { last_token_usage: { output_tokens: 12 }, total_token_usage: { total_tokens: 12, output_tokens: 12 } }, + result: 'x'.repeat(5 * 1024 * 1024), + }, + }) + await writeFile(path, largeResult) + const points = await readCodexThroughput(path) + expect(points).toHaveLength(1) + expect(points[0]?.generatedTokens).toBe(12) + }) + + it('keeps MCP duration when arguments and result surround the middle field', async () => { + const dir = await mkdtemp(join(tmpdir(), 'codeburn-tps-mcp-large-')) + const path = join(dir, 'rollout.jsonl') + const mcp = JSON.stringify({ + type: 'event_msg', timestamp: '2026-07-25T00:00:05.000Z', + payload: { + type: 'mcp_tool_call_end', + invocation: { server: 'github', tool: 'get_issue', arguments: { body: 'x'.repeat(5 * 1024 * 1024) } }, + duration: { secs: 3, nanos: 0 }, + result: { duration: '9s', text: 'x'.repeat(5 * 1024 * 1024) }, + }, + }) + await writeFile(path, [ + JSON.stringify({ type: 'session_meta', timestamp: '2026-07-25T00:00:00.000Z', payload: { model: 'gpt-5.6-sol' } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:00.000Z', payload: { type: 'task_started' } }), + mcp, + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:08.000Z', payload: { type: 'token_count', info: { last_token_usage: { output_tokens: 100 }, total_token_usage: { total_tokens: 100, output_tokens: 100 } } } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:10.000Z', payload: { type: 'task_complete', duration_ms: 10000 } }), + ].join('\n')) + const points = await readCodexThroughput(path) + expect(points[0]).toMatchObject({ toolWaitSeconds: 3, activeDurationSeconds: 7, activeGeneratedTokensPerSecond: 100 / 7 }) + }) + + it('keeps a streamed string MCP duration when arguments and result surround the middle field', async () => { + const dir = await mkdtemp(join(tmpdir(), 'codeburn-tps-mcp-string-large-')) + const path = join(dir, 'rollout.jsonl') + const mcp = JSON.stringify({ + type: 'event_msg', timestamp: '2026-07-25T00:00:05.000Z', + payload: { + type: 'mcp_tool_call_end', + invocation: { server: 'github', tool: 'get_issue', arguments: { body: 'x'.repeat(5 * 1024 * 1024) } }, + duration: '3s', + result: { duration: '9s', text: 'x'.repeat(5 * 1024 * 1024) }, + }, + }) + await writeFile(path, [ + JSON.stringify({ type: 'session_meta', timestamp: '2026-07-25T00:00:00.000Z', payload: { model: 'gpt-5.6-sol' } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:00.000Z', payload: { type: 'task_started' } }), + mcp, + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:08.000Z', payload: { type: 'token_count', info: { last_token_usage: { output_tokens: 100 }, total_token_usage: { total_tokens: 100, output_tokens: 100 } } } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-07-25T00:00:10.000Z', payload: { type: 'task_complete', duration_ms: 10000 } }), + ].join('\n')) + const points = await readCodexThroughput(path) + expect(points[0]).toMatchObject({ toolWaitSeconds: 3, activeDurationSeconds: 7, activeGeneratedTokensPerSecond: 100 / 7 }) + }) +}) diff --git a/tests/providers/codex.test.ts b/tests/providers/codex.test.ts index 9595e356..0b59ecbb 100644 --- a/tests/providers/codex.test.ts +++ b/tests/providers/codex.test.ts @@ -160,6 +160,56 @@ describe('codex provider - session discovery', () => { }]) }) + it('deduplicates the same session_id across active and archived roots', async () => { + const sharedLines = [ + sessionMeta({ cwd: '/Users/test/shared', session_id: 'sess-shared' }), + tokenCount({ last: { input: 100, output: 50 }, total: { total: 150 } }), + ] + const activePath = await writeSession(tmpDir, '2026-04-14', 'rollout-shared.jsonl', sharedLines) + const archivedCopyPath = await writeArchivedSession(tmpDir, 'rollout-shared.jsonl', sharedLines) + const distinctPath = await writeArchivedSession(tmpDir, 'rollout-distinct.jsonl', [ + sessionMeta({ cwd: '/Users/test/distinct', session_id: 'sess-distinct' }), + tokenCount({ last: { input: 200, output: 50 }, total: { total: 250 } }), + ]) + + const provider = createCodexProvider(tmpDir) + const sessions = await provider.discoverSessions() + const paths = sessions.map(session => session.path) + + expect(sessions).toHaveLength(2) + expect(paths).toEqual(expect.arrayContaining([activePath, distinctPath])) + expect(paths).not.toContain(archivedCopyPath) + }) + + it('does not double-count usage for an archived copy while counting distinct sessions', async () => { + const sharedLines = [ + sessionMeta({ session_id: 'sess-shared' }), + tokenCount({ last: { input: 100, output: 50 }, total: { total: 150 } }), + ] + await writeSession(tmpDir, '2026-04-14', 'rollout-shared.jsonl', sharedLines) + await writeArchivedSession(tmpDir, 'rollout-shared-copy.jsonl', sharedLines) + await writeArchivedSession(tmpDir, 'rollout-distinct.jsonl', [ + sessionMeta({ session_id: 'sess-distinct' }), + tokenCount({ last: { input: 200, output: 50 }, total: { total: 250 } }), + ]) + + const provider = createCodexProvider(tmpDir) + const sessions = await provider.discoverSessions() + const seenKeys = new Set() + const calls: ParsedProviderCall[] = [] + for (const session of sessions) { + for await (const call of provider.createSessionParser(session, seenKeys).parse()) { + calls.push(call) + } + } + + expect(calls.map(call => call.sessionId).sort()).toEqual(['sess-distinct', 'sess-shared']) + expect(calls.reduce( + (total, call) => total + call.inputTokens + call.cachedInputTokens + call.outputTokens + call.reasoningTokens, + 0, + )).toBe(400) + }) + it('returns empty for non-existent directory', async () => { const provider = createCodexProvider('/nonexistent/path/that/does/not/exist') const sessions = await provider.discoverSessions() @@ -332,6 +382,89 @@ describe('codex provider - JSONL parsing', () => { expect(call.deduplicationKey).toContain('codex:') }) + it('parses large rollout lines and computes active timing for custom tool calls', async () => { + const largeTokenLine = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T10:01:10Z', + payload: { + type: 'token_count', + info: { + last_token_usage: { input_tokens: 100, cached_input_tokens: 0, output_tokens: 100, reasoning_output_tokens: 20, total_tokens: 220 }, + total_token_usage: { input_tokens: 100, cached_input_tokens: 0, output_tokens: 100, reasoning_output_tokens: 20, total_tokens: 220 }, + }, + rate_limits: { filler: 'x'.repeat(40_000) }, + }, + }) + const largeCompleteLine = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T10:01:11Z', + payload: { type: 'task_complete', last_agent_message: 'x'.repeat(40_000), duration_ms: 10_000 }, + }) + const filePath = await writeSession(tmpDir, '2026-04-14', 'rollout-timing.jsonl', [ + sessionMeta({ session_id: 'sess-timing', model: 'gpt-5.5' }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:00Z', payload: { type: 'task_started', turn_id: 'turn-1' } }), + userMessage('run the tool'), + JSON.stringify({ type: 'response_item', timestamp: '2026-04-14T10:00:02Z', payload: { type: 'custom_tool_call', call_id: 'call-1', name: 'exec' } }), + JSON.stringify({ type: 'response_item', timestamp: '2026-04-14T10:00:05Z', payload: { type: 'custom_tool_call_output', call_id: 'call-1', output: 'done' } }), + largeTokenLine, + largeCompleteLine, + ]) + + const provider = createCodexProvider(tmpDir) + const source = { path: filePath, project: 'test', provider: 'codex' } + const calls: ParsedProviderCall[] = [] + for await (const call of provider.createSessionParser(source, new Set()).parse()) calls.push(call) + + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ + outputTokens: 100, + reasoningTokens: 20, + tools: ['Bash'], + activeDurationMs: 7000, + activeGeneratedTokens: 120, + toolWaitMs: 3000, + }) + }) + + it('keeps estimated output parsing for large token lines without usage info', async () => { + // Some rollout variants put token_count metadata beyond the compact head + // or omit `info` entirely. The line must still reach the character-based + // estimate path rather than being interpreted as an empty usage object. + const largeTokenLine = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T10:01:10Z', + payload: { type: 'token_count' }, + filler: 'x'.repeat(40_000), + }) + const assistantLine = JSON.stringify({ + type: 'response_item', + timestamp: '2026-04-14T10:01:05Z', + payload: { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'generated response '.repeat(100) }], + }, + }) + const filePath = await writeSession(tmpDir, '2026-04-14', 'rollout-estimated-large.jsonl', [ + sessionMeta({ session_id: 'sess-estimated-large', model: 'gpt-5.5' }), + userMessage('summarize the result'), + assistantLine, + largeTokenLine, + ]) + + const provider = createCodexProvider(tmpDir) + const source = { path: filePath, project: 'test', provider: 'codex' } + const calls: ParsedProviderCall[] = [] + for await (const call of provider.createSessionParser(source, new Set()).parse()) calls.push(call) + + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ + model: 'gpt-5.5', + costIsEstimated: true, + }) + expect(calls[0]!.outputTokens).toBeGreaterThan(0) + }) + it('attributes MCP calls emitted as event_msg/mcp_tool_call_end', async () => { const filePath = await writeSession(tmpDir, '2026-04-14', 'rollout-mcp.jsonl', [ sessionMeta({ session_id: 'sess-mcp', model: 'gpt-5.5' }), @@ -356,6 +489,89 @@ describe('codex provider - JSONL parsing', () => { expect(calls[0]!.tools).toEqual(['mcp__github__get_issue']) }) + it('subtracts native MCP wait time from active timing', async () => { + const filePath = await writeSession(tmpDir, '2026-04-14', 'rollout-mcp-timing.jsonl', [ + sessionMeta({ session_id: 'sess-mcp-timing', model: 'gpt-5.5' }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:00Z', payload: { type: 'task_started' } }), + userMessage('look up the issue'), + JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T10:00:05Z', + payload: { + type: 'mcp_tool_call_end', + call_id: 'mcp-1', + invocation: { server: 'github', tool: 'get_issue', arguments: {} }, + duration: { secs: 3, nanos: 0 }, + }, + }), + tokenCount({ + timestamp: '2026-04-14T10:00:08Z', + last: { input: 300, output: 100 }, + total: { total: 400 }, + }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:10Z', payload: { type: 'task_complete', duration_ms: 10_000 } }), + ]) + + const provider = createCodexProvider(tmpDir) + const source = { path: filePath, project: 'test', provider: 'codex' } + const calls: ParsedProviderCall[] = [] + for await (const call of provider.createSessionParser(source, new Set()).parse()) calls.push(call) + + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ activeDurationMs: 7000, toolWaitMs: 3000 }) + }) + + it('keeps MCP attribution on large result lines', async () => { + const largeMcpLine = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T10:00:05Z', + payload: { + type: 'mcp_tool_call_end', + call_id: 'mcp-large', + invocation: { server: 'github', tool: 'get_issue', arguments: { duration: '1s', body: 'x'.repeat(100_000) } }, + duration: { secs: 3, nanos: 0 }, + result: { Ok: { content: [{ type: 'text', text: 'x'.repeat(40_000) }] } }, + }, + }) + const filePath = await writeSession(tmpDir, '2026-04-14', 'rollout-mcp-large.jsonl', [ + sessionMeta({ session_id: 'sess-mcp-large', model: 'gpt-5.5' }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:00Z', payload: { type: 'task_started' } }), + userMessage('look up the issue'), + largeMcpLine, + tokenCount({ timestamp: '2026-04-14T10:00:08Z', last: { input: 300, output: 100 }, total: { total: 400 } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:10Z', payload: { type: 'task_complete', duration_ms: 10_000 } }), + ]) + + const provider = createCodexProvider(tmpDir) + const source = { path: filePath, project: 'test', provider: 'codex' } + const calls: ParsedProviderCall[] = [] + for await (const call of provider.createSessionParser(source, new Set()).parse()) calls.push(call) + + expect(calls).toHaveLength(1) + expect(calls[0]).toMatchObject({ tools: ['mcp__github__get_issue'], activeDurationMs: 7000, toolWaitMs: 3000 }) + }) + + it('omits active timing when recorded tool wait consumes the task duration', async () => { + const filePath = await writeSession(tmpDir, '2026-04-14', 'rollout-degenerate-timing.jsonl', [ + sessionMeta({ session_id: 'sess-degenerate-timing', model: 'gpt-5.5' }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:00Z', payload: { type: 'task_started' } }), + userMessage('wait for the tool'), + JSON.stringify({ type: 'response_item', timestamp: '2026-04-14T10:00:00Z', payload: { type: 'custom_tool_call', call_id: 'call-1', name: 'exec' } }), + JSON.stringify({ type: 'response_item', timestamp: '2026-04-14T10:00:10Z', payload: { type: 'custom_tool_call_output', call_id: 'call-1', output: 'done' } }), + tokenCount({ timestamp: '2026-04-14T10:00:12Z', last: { input: 300, output: 100 }, total: { total: 400 } }), + JSON.stringify({ type: 'event_msg', timestamp: '2026-04-14T10:00:13Z', payload: { type: 'task_complete', duration_ms: 10_000 } }), + ]) + + const provider = createCodexProvider(tmpDir) + const source = { path: filePath, project: 'test', provider: 'codex' } + const calls: ParsedProviderCall[] = [] + for await (const call of provider.createSessionParser(source, new Set()).parse()) calls.push(call) + + expect(calls).toHaveLength(1) + expect(calls[0]!.activeDurationMs).toBeUndefined() + expect(calls[0]!.toolWaitMs).toBeUndefined() + }) + it('attributes CLI-wrapped MCP calls (mcp-cli call server tool) to MCP + Bash', async () => { const execStr = (command: string) => JSON.stringify({ type: 'response_item',