diff --git a/packages/core/src/providers/codex/decode.ts b/packages/core/src/providers/codex/decode.ts index 5bcab263..f3bcfb9d 100644 --- a/packages/core/src/providers/codex/decode.ts +++ b/packages/core/src/providers/codex/decode.ts @@ -37,6 +37,9 @@ function normalizeContentBlocks( export const codexToolNameMap: 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', @@ -95,11 +98,45 @@ 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 +} + +// Extract the token buckets of a token_count payload from a compact head scan. +// Mirrors the full-JSON `info` shape so the Buffer path and the string path +// produce identical payloads for the decoder's token-count branch. +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++) { @@ -172,6 +209,28 @@ export 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, and mcp_tool_call_end can place a large result after + // them. Mirror the pre-extraction decoder: for those events fall back to a + // tail window of the line so fields past the compact head are not lost. + 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 + // Synthesize the fields the full-JSON path would carry, so the Buffer path + // and the string path produce identical payloads for the same line: `info` + // (token buckets — latent: token_count lines are small and never route here) + // and `invocation` (live: large MCP records with a huge invocation.arguments + // object exceed LARGE_STREAM_LINE_BYTES and lose their mcp__server__tool + // attribution without this). + 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, @@ -186,6 +245,8 @@ export function parseCodexLine(line: string | Buffer): CodexEntry | null { forked_from_id: getRawJsonStringField(pHead, 'forked_from_id'), model: getRawJsonStringField(pHead, 'model'), name: getRawJsonStringField(pHead, 'name'), + invocation, + info: compactInfo, }, } @@ -317,7 +378,24 @@ export function decodeCodex({ records, state: prevState, seenKeys: liveSeen, ses continue } - if (entry.type === 'response_item' && entry.payload?.type === 'function_call') { + // Forked sessions replay the parent's event history clustered at the fork + // creation time. Skip replayed events (within 5s of fork) so the parent's + // tool calls, patch applies and MCP tool ends do not leak into the child's + // turn (inflating tool counts, edit LOC and failed-edit counts). The + // token_count branch applies the same cutoff for token/cost dedup. + const isForkReplay = Boolean(s.forkCutoff && entry.timestamp && entry.timestamp < s.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 === 'response_item' && (entry.payload?.type === 'function_call' || entry.payload?.type === 'custom_tool_call')) { const rawName = entry.payload.name ?? '' const mapped = codexToolNameMap[rawName] ?? rawName s.pendingTools.push(mapped) diff --git a/packages/core/src/providers/codex/types.ts b/packages/core/src/providers/codex/types.ts index b361cdae..97cc2f8f 100644 --- a/packages/core/src/providers/codex/types.ts +++ b/packages/core/src/providers/codex/types.ts @@ -28,6 +28,7 @@ export type CodexEntry = { forked_from_id?: string model?: string name?: string + invocation?: { server?: string; tool?: string } content?: Array<{ type?: string; text?: string }> info?: { model?: string diff --git a/packages/core/tests/providers/codex-decode.test.ts b/packages/core/tests/providers/codex-decode.test.ts index 164fb029..17924ce8 100644 --- a/packages/core/tests/providers/codex-decode.test.ts +++ b/packages/core/tests/providers/codex-decode.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' -import { decodeCodex } from '../../src/providers/codex/index.js' +import { codexToolNameMap, decodeCodex, parseCodexLine } from '../../src/providers/codex/index.js' import type { CodexDecodeState } from '../../src/providers/codex/index.js' import type { DecodeContext } from '../../src/contracts.js' @@ -28,6 +28,23 @@ function assistantMessage(text: string, timestamp: string) { function functionCall(name: string, timestamp: string) { return JSON.stringify({ type: 'response_item', timestamp, payload: { type: 'function_call', name } }) } +function customToolCall(name: string, timestamp: string) { + return JSON.stringify({ type: 'response_item', timestamp, payload: { type: 'custom_tool_call', name } }) +} +function patchApplyEnd(opts: { success: boolean; added: number; file: string; timestamp: string }) { + return JSON.stringify({ + type: 'event_msg', + timestamp: opts.timestamp, + payload: { + type: 'patch_apply_end', + success: opts.success, + changes: { [opts.file]: { unified_diff: '+a\n'.repeat(opts.added) } }, + }, + }) +} +function mcpToolCallEnd(server: string, tool: string, timestamp: string) { + return JSON.stringify({ type: 'event_msg', timestamp, payload: { type: 'mcp_tool_call_end', invocation: { server, tool } } }) +} function tokenCount(opts: { timestamp: string; last?: { input?: number; cached?: number; output?: number; reasoning?: number }; total?: { input?: number; cached?: number; output?: number; reasoning?: number; total?: number }; noInfo?: boolean }) { const info = opts.noInfo ? undefined : { last_token_usage: opts.last ? { input_tokens: opts.last.input ?? 0, cached_input_tokens: opts.last.cached ?? 0, output_tokens: opts.last.output ?? 0, reasoning_output_tokens: opts.last.reasoning ?? 0, total_tokens: (opts.last.input ?? 0) + (opts.last.output ?? 0) } : undefined, @@ -65,7 +82,7 @@ const CORPUS: string[] = [ const FORK_BOUNDARY_INDEX = 7 // the fork's session_meta const MID_PARENT_INDEX = 4 // between A's two turns -function decodeCold(records: string[]) { +function decodeCold(records: (string | Buffer)[]) { return decodeCodex({ records, context }).calls } @@ -127,3 +144,130 @@ describe('codex decoder — round-trip resume invariant', () => { expect(parentAndForkFresh.length).toBeGreaterThan(threaded.length) }) }) + +describe('codex decoder — decode fidelity restoration (GAP 1-4)', () => { + it('GAP 1: custom_tool_call events feed the turn tools and tool sequence', () => { + const calls = decodeCold([ + sessionMeta({ session_id: 'sess-custom' }), + userMessage('use the custom tool', '2026-04-14T12:00:01Z'), + customToolCall('my_custom_tool', '2026-04-14T12:00:02Z'), + tokenCount({ timestamp: '2026-04-14T12:00:03Z', last: { input: 100 }, total: { input: 100, total: 100 } }), + ]) + expect(calls).toHaveLength(1) + expect(calls[0]!.tools).toEqual(['my_custom_tool']) + expect(calls[0]!.toolSequence).toEqual([[{ tool: 'my_custom_tool' }]]) + }) + + it("GAP 2: fork replay does not leak the parent's tool/patch/MCP events into the child turn", () => { + // Parent: one turn with a FAILED edit alongside Bash + MCP. The child + // replays that history verbatim inside the 5s fork window, then does its + // own genuine work (a read + a successful edit on a different file). The + // child's turn must carry exactly its own tools/sequence/LOC — none of the + // replayed Bash, failed edit or MCP end. The child fixture has real tool + // events of its own so the test bites in BOTH directions: an + // over-aggressive skip that eats everything empties the child turn (child + // asserts fail); a neutralized skip leaks the parent's replay into it + // (parent asserts still pass, child asserts fail). + const calls = decodeCold([ + // Parent session: one turn with a Bash call, a FAILED edit and an MCP call. + sessionMeta({ session_id: 'sess-parent', timestamp: '2026-04-14T10:00:00Z' }), + userMessage('parent turn', '2026-04-14T10:00:01Z'), + functionCall('exec_command', '2026-04-14T10:00:02Z'), + patchApplyEnd({ success: false, added: 2, file: 'src/a.ts', timestamp: '2026-04-14T10:00:03Z' }), + mcpToolCallEnd('srv', 't1', '2026-04-14T10:00:04Z'), + tokenCount({ timestamp: '2026-04-14T10:00:05Z', last: { input: 500 }, total: { input: 500, total: 500 } }), + // Fork created at 10:05:00 → cutoff 10:05:05. The parent's history is + // replayed clustered inside the window (10:05:01-04) and must be skipped + // wholesale — a replayed FAILED patch and MCP end must not leak into the + // child's turn (which would inflate tools, locAdded and editFailed). + sessionMeta({ session_id: 'sess-fork', forked_from_id: 'sess-parent', timestamp: '2026-04-14T10:05:00Z' }), + functionCall('exec_command', '2026-04-14T10:05:01Z'), + patchApplyEnd({ success: false, added: 2, file: 'src/a.ts', timestamp: '2026-04-14T10:05:02Z' }), + mcpToolCallEnd('srv', 't1', '2026-04-14T10:05:03Z'), + tokenCount({ timestamp: '2026-04-14T10:05:04Z', last: { input: 500 }, total: { input: 500, total: 500 } }), + // Child's own turn, past the cutoff: genuine tool events of its own. + userMessage('child turn', '2026-04-14T10:05:20Z'), + functionCall('read_file', '2026-04-14T10:05:20Z'), + patchApplyEnd({ success: true, added: 3, file: 'src/b.ts', timestamp: '2026-04-14T10:05:20Z' }), + tokenCount({ timestamp: '2026-04-14T10:05:21Z', last: { input: 300 }, total: { input: 800, total: 800 } }), + ]) + expect(calls).toHaveLength(2) + // The parent's turn keeps its own tools, failed-edit flag and LOC. + expect(calls[0]!.inputTokens).toBe(500) + expect(calls[0]!.tools).toEqual(['Bash', 'Edit', 'mcp__srv__t1']) + expect(calls[0]!.toolSequence).toEqual([ + [{ tool: 'Bash' }], + [{ tool: 'Edit', file: 'src/a.ts' }], + [{ tool: 'mcp__srv__t1' }], + ]) + expect(calls[0]!.locAdded).toBe(2) + expect(calls[0]!.editFailed).toBe(1) + // The child's turn sees NONE of the replayed events: exactly its own read + + // successful edit, no leaked Bash / failed edit / MCP end. + expect(calls[1]!.inputTokens).toBe(300) + expect(calls[1]!.tools).toEqual(['Read', 'Edit']) + expect(calls[1]!.toolSequence).toEqual([ + [{ tool: 'Read' }], + [{ tool: 'Edit', file: 'src/b.ts' }], + ]) + expect(calls[1]!.locAdded).toBe(3) + expect(calls[1]!.locRemoved).toBeUndefined() + expect(calls[1]!.editFailed).toBeUndefined() + }) + + it('GAP 3: Buffer path synthesizes payload.info and payload.invocation', () => { + // info on the Buffer path is a latent gap: a token_count line is a handful + // of numbers and never exceeds LARGE_STREAM_LINE_BYTES (32KB), so it always + // arrives as a string. The Buffer branch must still preserve it if hit. + const tokenLine = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T13:00:00Z', + payload: { + type: 'token_count', + info: { + last_token_usage: { input_tokens: 111, output_tokens: 7, total_tokens: 118 }, + total_token_usage: { input_tokens: 111, total_tokens: 118 }, + }, + }, + }) + const tokenEntry = parseCodexLine(Buffer.from(tokenLine)) + expect(tokenEntry?.payload?.info?.last_token_usage?.input_tokens).toBe(111) + expect(tokenEntry?.payload?.info?.total_token_usage?.total_tokens).toBe(118) + + // invocation is LIVE: an mcp_tool_call_end carrying a huge + // invocation.arguments object exceeds the 32KB threshold, routes to the + // Buffer path, and without invocation extraction the mcp__server__tool + // name is lost from the turn. + const bigArgs = 'y'.repeat(40 * 1024) + const mcpLine = JSON.stringify({ + type: 'event_msg', + timestamp: '2026-04-14T13:00:01Z', + payload: { type: 'mcp_tool_call_end', invocation: { server: 'srv', tool: 'big', arguments: { blob: bigArgs } } }, + }) + expect(Buffer.byteLength(mcpLine)).toBeGreaterThan(32 * 1024) + const mcpEntry = parseCodexLine(Buffer.from(mcpLine)) + expect(mcpEntry?.payload?.invocation).toEqual({ server: 'srv', tool: 'big' }) + + // Decode-level: the large MCP record is attributed as mcp__srv__big. + const calls = decodeCold([ + sessionMeta({ session_id: 'sess-big' }), + userMessage('big mcp', '2026-04-14T13:00:10Z'), + Buffer.from(mcpLine), + tokenCount({ timestamp: '2026-04-14T13:00:11Z', last: { input: 50 }, total: { input: 50, total: 50 } }), + ]) + expect(calls).toHaveLength(1) + expect(calls[0]!.tools).toEqual(['mcp__srv__big']) + }) + + it("GAP 4: 'exec' maps to 'Bash' for the Codex Desktop custom-tool transport", () => { + expect(codexToolNameMap['exec']).toBe('Bash') + const calls = decodeCold([ + sessionMeta({ session_id: 'sess-exec' }), + userMessage('run it', '2026-04-14T14:00:01Z'), + functionCall('exec_command', '2026-04-14T14:00:02Z'), + customToolCall('exec', '2026-04-14T14:00:03Z'), + tokenCount({ timestamp: '2026-04-14T14:00:04Z', last: { input: 50 }, total: { input: 50, total: 50 } }), + ]) + expect(calls[0]!.tools).toEqual(['Bash', 'Bash']) + }) +})