fix(codex): restore decode behaviour lost in the extraction

packages/core/src/providers/codex/decode.ts

@@ -37,6 +37,9 @@ function normalizeContentBlocks<T extends { type?: string; text?: string }>(
 
 export const codexToolNameMap: Record<string, string> = {
   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)

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

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'])
+  })
+})