From 7bfc37d29074d99d6d8f86462c7afea14f9b2364 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 10 Sep 2026 01:19:15 -0700 Subject: [PATCH 1/2] fix(codex): record command, file-change and submitted-turn facts An audit question asks which commands ran, what they exited with, when they ran, which files changed, and what the person actually typed. The Codex adapter dropped every `item_completed` event, so none of those facts reached a span, and it labelled harness-injected context as a human turn. - Emit one CHAIN span per `CommandExecution` item (command, cwd, exit code, process id, output, and the item's own start and end times) and one per changed path in a `FileChange` item. Each item joins the one tool call whose window contains its whole run; an item that outlives every call or falls inside two stays under the session root and says so. Item shapes the adapter cannot represent are counted on the root, never guessed. - Mark the inner spans `traces.tool_call.level=inner` so tool-call counts, loop detection and conversation text keep reading the model-issued level. - Accept a script receipt whose "Wall time" line has no colon, which left script outcomes UNSET. - Treat Codex's context blocks (``, ``, ``, the `` wrappers, the injected warnings and the rest of `CONTEXTUAL_USER_FRAGMENT_MATCHERS`) as injected, and take the human turn from the record Codex writes for submitted input: the legacy `user_message` event or the current `item_completed`/`UserMessage` item, paired with its response-item copy so one turn stays one span. - Drop the parent session from `childSessionIds`: a child that messages its parent named it as a `send_message` target. Co-Authored-By: Claude Opus 5 (1M context) --- src/adapters/actor.ts | 57 ++++- src/adapters/codex-format.ts | 195 +++++++++++++++++ src/adapters/codex.ts | 334 ++++++++++++++++++++++++++++-- src/adapters/tool-io.ts | 13 ++ src/live.ts | 4 +- src/session-relationship.ts | 12 +- tests/codex-command-facts.test.ts | 301 +++++++++++++++++++++++++++ tests/codex-facts-fixture.ts | 204 ++++++++++++++++++ tests/codex-tool-status.test.ts | 3 + 9 files changed, 1095 insertions(+), 28 deletions(-) create mode 100644 tests/codex-command-facts.test.ts create mode 100644 tests/codex-facts-fixture.ts diff --git a/src/adapters/actor.ts b/src/adapters/actor.ts index 7549abe..8d51d5f 100644 --- a/src/adapters/actor.ts +++ b/src/adapters/actor.ts @@ -160,12 +160,63 @@ export function claudeActor(args: { return 'human' } +/** + * Context blocks Codex writes as user-role messages and never reports as a user + * turn. The list mirrors `CONTEXTUAL_USER_FRAGMENT_MATCHERS` (openai/codex + * `codex-rs/core/src/context/contextual_user_message.rs`); the default matcher + * accepts a trimmed block that starts with the open marker and ends with the + * close marker, ignoring ASCII case (openai/codex + * `codex-rs/context-fragments/src/fragment.rs`). + */ +const CODEX_CONTEXT_BLOCKS: ReadonlyArray = [ + ['# AGENTS.md instructions', ''], + ['', ''], + ['', ''], + ['', ''], + ['', ''], + ['', ''], + ['', ''], + [''], + ['', ''], + ['', ''], +] + +/** Harness warnings Codex also injects as user-role text, matched by prefix. */ +const CODEX_CONTEXT_PREFIXES = [ + 'Warning: apply_patch was requested via ', + 'Warning: Your account was flagged for potentially high-risk cyber activity', + 'Warning: The maximum number of unified exec processes you can keep open is', +] as const + +const CODEX_EXTERNAL_CONTEXT = /^]+)>[\s\S]*<\/external_\1>$/ + +function startsWithIgnoringCase(text: string, prefix: string): boolean { + return text.slice(0, prefix.length).toLowerCase() === prefix.toLowerCase() +} + +function endsWithIgnoringCase(text: string, suffix: string): boolean { + return text.slice(-suffix.length).toLowerCase() === suffix.toLowerCase() +} + +/** Whether one text block of a Codex user-role message is harness context. */ +export function isCodexContextBlock(block: string): boolean { + const text = block.trim() + if (CODEX_CONTEXT_PREFIXES.some((prefix) => text.startsWith(prefix))) return true + if (CODEX_EXTERNAL_CONTEXT.test(text)) return true + return CODEX_CONTEXT_BLOCKS.some( + ([open, close]) => startsWithIgnoringCase(text, open) && endsWithIgnoringCase(text, close), + ) +} + /** * Derive the actor for a Codex user message. Codex has no sidechain/userType, - * so it's text-only: synthetic markers → injected, first-turn agent-spawn - * brief → injected, otherwise human. + * so it's text-only: a Codex context block → injected, synthetic markers → + * injected, first-turn agent-spawn brief → injected, otherwise human. + * `blocks` are the message's separate text blocks; Codex treats the whole + * message as context when any one block is. */ -export function codexActor(args: { text: string; isFirstUserTurn?: boolean }): Actor { +export function codexActor(args: { text: string; blocks?: readonly string[]; isFirstUserTurn?: boolean }): Actor { + if ((args.blocks ?? [args.text]).some(isCodexContextBlock)) return 'injected' if (textIsCmdOrInject(args.text)) return 'injected' if (textIsSynthetic(args.text)) return 'injected' if (args.isFirstUserTurn && looksLikeAgentPrompt(args.text)) return 'injected' diff --git a/src/adapters/codex-format.ts b/src/adapters/codex-format.ts index 0b181b2..ee7f328 100644 --- a/src/adapters/codex-format.ts +++ b/src/adapters/codex-format.ts @@ -44,6 +44,8 @@ export interface CodexLine { author?: string recipient?: string namespace?: string + /** `user_message` event text. */ + message?: unknown item?: { type?: string id?: string @@ -121,6 +123,188 @@ export function codexSubagentActivity(line: CodexLine): CodexSubagentActivity | } } +/** A command Codex ran, from an `item_completed` event whose item is `CommandExecution`. */ +export interface CodexCommandExecution { + readonly itemId: string + /** Recorded argv (current builds) or command string, verbatim. */ + readonly command: readonly string[] | string + readonly cwd?: string + readonly processId?: string + readonly source?: string + readonly status?: string + readonly exitCode?: number + readonly output?: string | { readonly stdout?: string; readonly stderr?: string } + readonly outputFields: readonly string[] + readonly startedAtMs?: number + readonly completedAtMs?: number +} + +export interface CodexFileChangeEntry { + readonly path: string + /** Codex's change type (`add`, `delete`, `update`), verbatim. */ + readonly kind: string + readonly movePath?: string +} + +/** Files a patch changed, from an `item_completed` event whose item is `FileChange`. */ +export interface CodexFileChange { + readonly itemId: string + readonly changes: readonly CodexFileChangeEntry[] + readonly status?: string + readonly startedAtMs?: number + readonly completedAtMs?: number +} + +/** A turn a client submitted, from an `item_completed` event whose item is `UserMessage`. */ +export interface CodexUserMessage { + readonly itemId: string + readonly text: string +} + +/** + * One `item_completed` event, normalized. `skipped` carries the label the adapter + * counts when an item produces no span: the item type, or `:malformed` + * when a required field is missing. + */ +export type CodexCompletedItem = + | { readonly type: 'CommandExecution'; readonly item: object; readonly command: CodexCommandExecution } + | { readonly type: 'FileChange'; readonly item: object; readonly fileChange: CodexFileChange } + | { readonly type: 'UserMessage'; readonly item: object; readonly userMessage: CodexUserMessage } + | { readonly type: 'skipped'; readonly label: string } + +type JsonRecord = Record + +function recordValue(value: unknown): JsonRecord | undefined { + return value !== null && typeof value === 'object' && !Array.isArray(value) ? value as JsonRecord : undefined +} + +function nonEmptyString(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined +} + +/** Codex defaults a missing `completed_at_ms` to 0, so only a positive time is a recorded time. */ +function epochMs(value: unknown): number | undefined { + return typeof value === 'number' && Number.isFinite(value) && value > 0 ? value : undefined +} + +function commandValue(value: unknown): readonly string[] | string | undefined { + if (typeof value === 'string') return value.length > 0 ? value : undefined + if (!Array.isArray(value) || value.length === 0) return undefined + return value.every((part) => typeof part === 'string') ? value as string[] : undefined +} + +function commandOutput(item: JsonRecord): Pick { + if (typeof item.aggregated_output === 'string') { + return { output: item.aggregated_output, outputFields: ['aggregated_output'] } + } + const stdout = typeof item.stdout === 'string' ? item.stdout : undefined + const stderr = typeof item.stderr === 'string' ? item.stderr : undefined + if (stdout === undefined && stderr === undefined) return { outputFields: [] } + return { + output: { ...(stdout === undefined ? {} : { stdout }), ...(stderr === undefined ? {} : { stderr }) }, + outputFields: [...(stdout === undefined ? [] : ['stdout']), ...(stderr === undefined ? [] : ['stderr'])], + } +} + +/** The rollout records a map keyed by path; `codex exec --json` records an array of `{path, kind}`. */ +function fileChangeEntries(value: unknown): CodexFileChangeEntry[] | undefined { + const entries: CodexFileChangeEntry[] = [] + if (Array.isArray(value)) { + for (const raw of value) { + const change = recordValue(raw) + const path = nonEmptyString(change?.path) + const kind = nonEmptyString(change?.kind) ?? nonEmptyString(change?.type) + if (!path || !kind) return undefined + const movePath = nonEmptyString(change?.move_path) + entries.push({ path, kind, ...(movePath ? { movePath } : {}) }) + } + } else { + const changes = recordValue(value) + if (!changes) return undefined + for (const [path, raw] of Object.entries(changes)) { + const change = recordValue(raw) + const kind = nonEmptyString(change?.type) ?? nonEmptyString(change?.kind) + if (path.length === 0 || !kind) return undefined + const movePath = nonEmptyString(change?.move_path) + entries.push({ path, kind, ...(movePath ? { movePath } : {}) }) + } + } + if (entries.length === 0) return undefined + return entries.sort((left, right) => (left.path < right.path ? -1 : left.path > right.path ? 1 : 0)) +} + +/** The text a `UserMessage` item carries; other input parts (images, audio) have no text. */ +function userMessageText(content: unknown): string { + if (!Array.isArray(content)) return '' + return content + .map((part) => { + const entry = recordValue(part) + return entry?.type === 'text' && typeof entry.text === 'string' ? entry.text : '' + }) + .join('') +} + +/** + * Normalize Codex's `item_completed` event for the command, file-change and + * user-message items. + * + * Field names follow `CommandExecutionItem`, `FileChangeItem`, `UserMessageItem` + * and `ItemCompletedEvent` in openai/codex `codex-rs/protocol`. The format drifts, + * so a required field that is missing or mistyped yields a counted `skipped` + * label rather than a span built from guessed values. + */ +export function codexCompletedItem(line: CodexLine): CodexCompletedItem | undefined { + if (line.type !== 'event_msg' || line.payload?.type !== 'item_completed') return undefined + const payload = line.payload as JsonRecord + const item = recordValue(payload.item) + const type = nonEmptyString(item?.type) ?? 'unknown' + if (!item || (type !== 'CommandExecution' && type !== 'FileChange' && type !== 'UserMessage')) { + return { type: 'skipped', label: type } + } + const itemId = nonEmptyString(item.id) + if (type === 'UserMessage') { + const text = userMessageText(item.content) + // An image-only or audio-only turn carries no text to record as a turn span. + if (!itemId || !text) return { type: 'skipped', label: `${type}:${itemId ? 'no_text' : 'malformed'}` } + return { type, item, userMessage: { itemId, text } } + } + const startedAtMs = epochMs(payload.started_at_ms) ?? epochMs(item.started_at_ms) + const completedAtMs = epochMs(payload.completed_at_ms) ?? epochMs(item.completed_at_ms) + const status = nonEmptyString(item.status) + const times = { + ...(startedAtMs === undefined ? {} : { startedAtMs }), + ...(completedAtMs === undefined ? {} : { completedAtMs }), + } + if (type === 'FileChange') { + const changes = fileChangeEntries(item.changes) + if (!itemId || !changes) return { type: 'skipped', label: `${type}:malformed` } + return { type, item, fileChange: { itemId, changes, ...(status ? { status } : {}), ...times } } + } + const command = commandValue(item.command) + if (!itemId || !command) return { type: 'skipped', label: `${type}:malformed` } + const cwd = nonEmptyString(item.cwd) + const processId = typeof item.process_id === 'number' && Number.isSafeInteger(item.process_id) + ? String(item.process_id) + : nonEmptyString(item.process_id) + const source = nonEmptyString(item.source) + const exitCode = typeof item.exit_code === 'number' && Number.isSafeInteger(item.exit_code) ? item.exit_code : undefined + return { + type, + item, + command: { + itemId, + command, + ...(cwd ? { cwd } : {}), + ...(processId ? { processId } : {}), + ...(source ? { source } : {}), + ...(status ? { status } : {}), + ...(exitCode === undefined ? {} : { exitCode }), + ...commandOutput(item), + ...times, + }, + } +} + export function contentToString(content: unknown): string { if (typeof content === 'string') return content if (Array.isArray(content)) { @@ -135,6 +319,17 @@ export function contentToString(content: unknown): string { return '' } +/** Each text block of a message, unjoined, so a per-block classifier sees block boundaries. */ +export function contentTextBlocks(content: unknown): string[] { + if (typeof content === 'string') return [content] + if (!Array.isArray(content)) return [] + return content.flatMap((item) => ( + item && typeof item === 'object' && typeof (item as { text?: unknown }).text === 'string' + ? [(item as { text: string }).text] + : [] + )) +} + export function timestampFromEpochMs(value: unknown): string | undefined { if (typeof value !== 'number' || !Number.isFinite(value)) return undefined const date = new Date(value) diff --git a/src/adapters/codex.ts b/src/adapters/codex.ts index 660266c..6edd98d 100644 --- a/src/adapters/codex.ts +++ b/src/adapters/codex.ts @@ -14,7 +14,7 @@ * Shared by the codex-acp wrapper via alias (same rollout format). */ -import { sourceOf, textSources } from '../source-location.js' +import { type SourceReferences, sourceOf, textSources } from '../source-location.js' import { readdir, stat } from 'node:fs/promises' import { homedir } from 'node:os' @@ -34,11 +34,16 @@ import type { SpawnedChildResolution, } from '../types.js' import { codexActor } from './actor.js' -import { capText, userPromptSpan } from './conversation.js' +import { ACTOR_ATTR, capText, userPromptSpan } from './conversation.js' import { + type CodexCommandExecution, + type CodexCompletedItem, + codexCompletedItem, + type CodexFileChange, type CodexLine, codexSubagentActivity, type CodexTokenUsage, + contentTextBlocks, contentToString, latestTimestamp, multiAgentOperation, @@ -57,7 +62,7 @@ import { isCodexTaskBoundary, resolveCodexParentTask, } from './codex-task-scope.js' -import { recordToolOutput, toolIoAttributes } from './tool-io.js' +import { INNER_TOOL_CALL_LEVEL, recordToolOutput, TOOL_CALL_LEVEL_ATTR, toolIoAttributes } from './tool-io.js' export { CodexTaskScopeError } from './codex-task-scope.js' @@ -167,8 +172,9 @@ function explicitOutputError(value: unknown, timeoutIsError = true): boolean | u const exitCode = numericStatus(header.match(/^Process exited with code[ \t]+(-?\d+)[ \t]*$/im)?.[1]) if (exitCode !== undefined) return exitCode !== 0 } - if (/^Script completed\s*\nWall time:/i.test(header)) return false - if (/^Script failed\s*\nWall time:/i.test(header)) return true + // Receipts print either "Wall time: 1.2 seconds" or "Wall time 1.2 seconds". + if (/^Script completed\s*\nWall time\b/i.test(header)) return false + if (/^Script failed\s*\nWall time\b/i.test(header)) return true const scriptExitCode = numericStatus(header.match(/^Script error:[ \t]*\r?\nExit code:[ \t]*(-?\d+)[ \t]*(?:\r?\n|$)/i)?.[1]) if (scriptExitCode !== undefined) return scriptExitCode !== 0 const commandExitCode = numericStatus(text.match(/^Command failed with exit code[ \t]+(-?\d+)\.?$/i)?.[1]) @@ -380,6 +386,189 @@ function closeSpanAt(target: OtlpSpan, sourceEndTime: string): void { target.end_time = sourceEndTime } +/** A completed item the adapter turns into spans; user messages take the turn path instead. */ +type CompletedItem = + & Extract + & { readonly recordTime: string } + +/** The time window of a model-issued call: its call record to its output record. */ +interface ToolWindow { + readonly startMs: number + readonly endMs: number +} + +/** + * How an item was placed. Codex item IDs never equal call IDs, so an item joins + * the one call whose window contains the item's whole run. An item that ran + * past every window (a command left running, then polled) or sits inside two + * windows (parallel calls) stays under the session root instead of a guess. + */ +type ItemJoin = 'call' | 'unmatched' | 'ambiguous' + +function joinItem( + windows: ReadonlyMap, + startMs: number, + endMs: number, +): { parent?: OtlpSpan; join: ItemJoin } { + const matches: OtlpSpan[] = [] + for (const [toolSpan, window] of windows) { + if (window.startMs <= startMs && endMs <= window.endMs) matches.push(toolSpan) + } + if (matches.length === 1) return { parent: matches[0], join: 'call' } + return { join: matches.length === 0 ? 'unmatched' : 'ambiguous' } +} + +function itemTimes( + item: { readonly startedAtMs?: number; readonly completedAtMs?: number }, + recordTime: string, +): { start: string; end: string; timeSource?: 'completed_only' | 'record' } { + const end = timestampFromEpochMs(item.completedAtMs) ?? recordTime + const start = timestampFromEpochMs(item.startedAtMs) ?? end + if (item.startedAtMs !== undefined && item.completedAtMs !== undefined) return { start, end } + return { start, end, timeSource: item.completedAtMs === undefined ? 'record' : 'completed_only' } +} + +function itemStatus( + status: string | undefined, + noun: string, + exitCode?: number, +): { code: OtlpSpan['status']['code']; message?: string } { + if (exitCode !== undefined && exitCode !== 0) return { code: 'ERROR', message: `${noun} exited ${exitCode}` } + if (status === 'failed' || status === 'declined') return { code: 'ERROR', message: `${noun} ${status}` } + if (exitCode === 0 || status === 'completed') return { code: 'OK' } + return { code: 'UNSET' } +} + +function itemSources(item: object, fields: readonly string[]) { + return fields.flatMap((field) => { + const reference = sourceOf(item, field) + return reference ? [reference] : [] + }) +} + +function innerItemAttributes( + type: CompletedItem['type'], + itemId: string, + join: ItemJoin, + timeSource: string | undefined, + status: string | undefined, +): Record { + return { + // Existing OTLP importers classify this marker as a container, not a call. + 'span.type': 'tool.execution', + [TOOL_CALL_LEVEL_ATTR]: INNER_TOOL_CALL_LEVEL, + 'traces.codex.item_type': type, + 'traces.codex.item_id': itemId, + 'traces.codex.item_join': join, + ...(timeSource ? { 'traces.codex.item_time_source': timeSource } : {}), + ...(status ? { 'traces.codex.item_status': status } : {}), + } +} + +interface ItemSpanContext { + readonly traceId: string + readonly rootId: string + readonly windows: ReadonlyMap +} + +/** + * One CHAIN span per command. Command text, cwd, and output stay in the tool + * I/O keys, which metadata-only upload strips and external redactors scrub. + */ +function commandSpan(context: ItemSpanContext, item: object, command: CodexCommandExecution, recordTime: string): OtlpSpan { + const { start, end, timeSource } = itemTimes(command, recordTime) + const { parent, join } = joinItem(context.windows, Date.parse(start), Date.parse(end)) + const status = itemStatus(command.status, 'command', command.exitCode) + const commandSpan = span({ + traceId: context.traceId, + spanId: `command:${command.itemId}`, + parentSpanId: parent?.span_id ?? context.rootId, + name: 'command.execution', + kind: 'CHAIN', + startTime: start, + status: status.code, + statusMessage: status.message, + service: SERVICE, + agent: SERVICE, + extra: { + ...toolIoAttributes({ + input: { command: command.command, ...(command.cwd ? { cwd: command.cwd } : {}) }, + inputSource: itemSources(item, ['command', ...(command.cwd ? ['cwd'] : [])]), + output: command.output, + outputSource: itemSources(item, command.outputFields), + }), + ...innerItemAttributes('CommandExecution', command.itemId, join, timeSource, command.status), + ...(command.exitCode === undefined ? {} : { 'process.exit_code': command.exitCode }), + ...(command.processId ? { 'traces.codex.process_id': command.processId } : {}), + ...(command.source ? { 'traces.codex.command_source': command.source } : {}), + }, + }) + closeSpanAt(commandSpan, end) + return commandSpan +} + +/** One CHAIN span per changed path; the path stays in `input.value` for the same reason as commands. */ +function fileChangeSpans(context: ItemSpanContext, item: object, fileChange: CodexFileChange, recordTime: string): OtlpSpan[] { + const { start, end, timeSource } = itemTimes(fileChange, recordTime) + const { parent, join } = joinItem(context.windows, Date.parse(start), Date.parse(end)) + const status = itemStatus(fileChange.status, 'file change') + return fileChange.changes.map((change, index) => { + const changeSpan = span({ + traceId: context.traceId, + spanId: `file-change:${fileChange.itemId}:${index}`, + parentSpanId: parent?.span_id ?? context.rootId, + name: 'file.change', + kind: 'CHAIN', + startTime: start, + status: status.code, + statusMessage: status.message, + service: SERVICE, + agent: SERVICE, + extra: { + ...toolIoAttributes({ + input: { path: change.path, kind: change.kind, ...(change.movePath ? { move_path: change.movePath } : {}) }, + inputSource: sourceOf(item, 'changes'), + }), + ...innerItemAttributes('FileChange', fileChange.itemId, join, timeSource, fileChange.status), + 'traces.codex.file_change_kind': change.kind, + }, + }) + closeSpanAt(changeSpan, end) + return changeSpan + }) +} + +/** A user turn awaiting its second record: Codex logs each typed turn as a response item and a `user_message` event. */ +interface UserTurnCandidate { + readonly span: OtlpSpan + readonly key: string + readonly task: number +} + +/** + * Legacy Codex prepends context to the submitted message and marks the typed + * text with this line (openai/codex `codex-rs/protocol/src/protocol.rs` + * `USER_MESSAGE_BEGIN`), so the two records of one turn can differ by a prefix. + */ +const USER_MESSAGE_BEGIN = '## My request for Codex:' + +function userTurnKey(text: string): string { + const begin = text.indexOf(USER_MESSAGE_BEGIN) + const typed = begin === -1 ? text : text.slice(begin + USER_MESSAGE_BEGIN.length) + return typed.trim().replace(/\s+/g, ' ') +} + +/** Remove and return the latest candidate with the same text in the same task. */ +function takeUserTurn(candidates: UserTurnCandidate[], key: string, task: number): UserTurnCandidate | undefined { + for (let index = candidates.length - 1; index >= 0; index -= 1) { + const candidate = candidates[index]! + if (candidate.key === key && candidate.task === task) return candidates.splice(index, 1)[0] + } + return undefined +} + +const USER_MESSAGE_EVENT_ATTR = 'traces.codex.user_message_event' + const verificationCommand = /\b(?:pnpm|npm|yarn|bun)\s+(?:run\s+)?(?:test|typecheck|lint|build|check)(?::[A-Za-z0-9:_-]+)?\b|\b(?:vitest|jest|pytest|tsc|biome|eslint|sha256sum|pdfinfo|pdftotext)\b|\bgo\s+test\b|\bcargo\s+(?:test|check|clippy|build)\b|\bgit\s+(?:status|diff|show|merge-tree)\b|\bgh-drew\s+pr\s+(?:view|checks)\b/i @@ -666,6 +855,59 @@ export class CodexAdapter implements HarnessTraceAdapter { let lastCumulativeTokenUsage: string | undefined let lastTimestamp: string | undefined const awaitingModel = model ? [] : [root] + const toolWindows = new Map() + const completedItems: CompletedItem[] = [] + const completedItemKeys = new Set() + const skippedItemCounts = new Map() + const countSkippedItem = (label: string): void => { + skippedItemCounts.set(label, (skippedItemCounts.get(label) ?? 0) + 1) + } + // Pairs the two records of one typed turn. A task index scopes the pairing, + // so the same short reply in two turns stays two turns. + let taskIndex = 0 + const unpairedUserItems: UserTurnCandidate[] = [] + const unpairedUserEvents: UserTurnCandidate[] = [] + const tasksWithUserEvents = new Set() + /** + * Record one turn Codex reports as submitted input. Codex reports these + * turns and never its own context blocks: the legacy `user_message` event + * and the current `item_completed`/`UserMessage` item both come from the + * same filter (openai/codex `codex-rs/core/src/event_mapping.rs`), and the + * rollout carries one or the other by history mode (openai/codex + * `codex-rs/rollout/src/policy.rs`). The response-item copy of the same turn + * pairs with this record instead of becoming a second span. + */ + const recordSubmittedTurn = (raw: string, ts: string, contentSource: SourceReferences): void => { + const prompt = capText(raw) + if (!prompt) return + tasksWithUserEvents.add(taskIndex) + const key = userTurnKey(raw) + const recorded = takeUserTurn(unpairedUserItems, key, taskIndex) + if (recorded) { + recorded.span.attributes[USER_MESSAGE_EVENT_ATTR] = true + return + } + const actor = sessionRole === 'child' + ? 'agent' + : codexActor({ text: prompt, isFirstUserTurn: !sawUserTurn }) + sawUserTurn = true + const turnSpan = userPromptSpan({ + traceId, + spanId: `msg:${step}:user`, + parentSpanId: rootId, + startTime: ts, + content: prompt, + contentSource, + service: SERVICE, + agent: SERVICE, + step, + actor, + }) + turnSpan.attributes[USER_MESSAGE_EVENT_ATTR] = true + spans.push(turnSpan) + unpairedUserEvents.push({ span: turnSpan, key, task: taskIndex }) + step += 1 + } const ensureSubagentSpan = ( threadId: string, agentPath: string, @@ -748,6 +990,7 @@ export class CodexAdapter implements HarnessTraceAdapter { lastTimestamp = latestTimestamp(lastTimestamp, l.timestamp) const ts = validTimestamp(l.timestamp) ?? lastTimestamp ?? root.start_time if (l.type === 'event_msg' && l.payload?.type === 'task_started') { + taskIndex += 1 activeTaskTurnId = l.payload.turn_id ?? null root.status = { code: 'UNSET' } } else if (l.type === 'event_msg' && l.payload?.type === 'task_complete') { @@ -848,6 +1091,7 @@ export class CodexAdapter implements HarnessTraceAdapter { const name = String(t.attributes['tool.name'] ?? '') const { status, pollOutcome } = outputStatus(name, l.payload) closeSpanAt(t, ts) + toolWindows.set(t, { startMs: Date.parse(t.start_time), endMs: Date.parse(ts) }) t.status = status if (pollOutcome) t.attributes['traces.poll.outcome'] = pollOutcome recordToolOutput(t, l.payload.output, sourceOf(l.payload, 'output')) @@ -864,9 +1108,31 @@ export class CodexAdapter implements HarnessTraceAdapter { if (requestId) t.attributes['traces.codex.agent_request_id'] = requestId } } + } else if (l.type === 'event_msg' && l.payload?.type === 'user_message') { + recordSubmittedTurn( + typeof l.payload.message === 'string' ? l.payload.message : '', + ts, + textSources(l.payload, 'message'), + ) } else if (l.type === 'event_msg') { const activity = codexSubagentActivity(l) - if (!activity) continue + if (!activity) { + const completed = codexCompletedItem(l) + if (completed?.type === 'skipped') { + countSkippedItem(completed.label) + } else if (completed?.type === 'UserMessage') { + recordSubmittedTurn(completed.userMessage.text, ts, textSources(completed.item, 'content')) + } else if (completed) { + const itemId = completed.type === 'CommandExecution' ? completed.command.itemId : completed.fileChange.itemId + const itemKey = `${completed.type}:${itemId}` + if (completedItemKeys.has(itemKey)) countSkippedItem(`${completed.type}:duplicate`) + else { + completedItemKeys.add(itemKey) + completedItems.push({ ...completed, recordTime: ts }) + } + } + continue + } const threadId = activity.agentThreadId const eventTime = timestampFromEpochMs(activity.occurredAtMs) ?? ts const eventCallSpan = toolByCallId.get(activity.eventId ?? '') @@ -953,26 +1219,30 @@ export class CodexAdapter implements HarnessTraceAdapter { } else if (l.type === 'response_item' && l.payload?.type === 'message' && l.payload.role === 'user') { // The human's prompt text. Codex drops the user turn from token events, // so capture it here as its own CHAIN span (no text → no span). - const prompt = textOf(l.payload.content) + const raw = contentToString(l.payload.content) + const prompt = capText(raw) if (prompt) { + const key = userTurnKey(raw) + // The user_message event already recorded this turn. + if (takeUserTurn(unpairedUserEvents, key, taskIndex)) continue const actor = sessionRole === 'child' ? 'agent' - : codexActor({ text: prompt, isFirstUserTurn: !sawUserTurn }) + : codexActor({ text: prompt, blocks: contentTextBlocks(l.payload.content), isFirstUserTurn: !sawUserTurn }) sawUserTurn = true - spans.push( - userPromptSpan({ - traceId, - spanId: `msg:${step}:user`, - parentSpanId: rootId, - startTime: ts, - content: prompt, - contentSource: textSources(l.payload, 'content'), - service: SERVICE, - agent: SERVICE, - step, - actor, - }), - ) + const turnSpan = userPromptSpan({ + traceId, + spanId: `msg:${step}:user`, + parentSpanId: rootId, + startTime: ts, + content: prompt, + contentSource: textSources(l.payload, 'content'), + service: SERVICE, + agent: SERVICE, + step, + actor, + }) + spans.push(turnSpan) + unpairedUserItems.push({ span: turnSpan, key, task: taskIndex }) step += 1 } } else if (l.type === 'response_item' && l.payload?.type === 'message') { @@ -1003,6 +1273,26 @@ export class CodexAdapter implements HarnessTraceAdapter { `Codex turn ${JSON.stringify(options.taskTurnId)} does not exist in ${ref.path}`, ) } + // Where Codex recorded user_message events for a task, a user-role message + // without one is harness context, even under a wrapper not listed in actor.ts. + for (const candidate of unpairedUserItems) { + if (candidate.span.attributes[ACTOR_ATTR] !== 'human' || !tasksWithUserEvents.has(candidate.task)) continue + candidate.span.attributes[ACTOR_ATTR] = 'injected' + candidate.span.attributes['traces.codex.actor_evidence'] = 'no_user_message_event' + } + const itemContext: ItemSpanContext = { traceId, rootId, windows: toolWindows } + for (const completed of completedItems) { + if (completed.type === 'CommandExecution') { + spans.push(commandSpan(itemContext, completed.item, completed.command, completed.recordTime)) + } else { + spans.push(...fileChangeSpans(itemContext, completed.item, completed.fileChange, completed.recordTime)) + } + } + if (skippedItemCounts.size > 0) { + root.attributes['traces.codex.skipped_item_counts'] = JSON.stringify( + Object.fromEntries([...skippedItemCounts].sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))), + ) + } if (selectedBoundary?.turnId) { for (const item of spans) item.attributes['traces.codex.turn_id'] ??= selectedBoundary.turnId } diff --git a/src/adapters/tool-io.ts b/src/adapters/tool-io.ts index 900ff46..42ca933 100644 --- a/src/adapters/tool-io.ts +++ b/src/adapters/tool-io.ts @@ -5,6 +5,19 @@ import { sourceAttributes, SOURCE_ATTRIBUTE_PREFIX, type SourceReferences } from export const TOOL_IO_VALUE_MAX_BYTES = 16 * 1024 export const TOOL_IO_VALUE_KEYS = ['input.value', 'output.value'] as const +/** + * Marks work recorded inside a model-issued tool call, such as each command a + * code-mode script ran. These spans are CHAIN, not TOOL, and carry no + * `tool.name`, so every tool-call counter here and in agent-eval stays at the + * model-issued level; readers that want the inner facts select this value. + */ +export const TOOL_CALL_LEVEL_ATTR = 'traces.tool_call.level' +export const INNER_TOOL_CALL_LEVEL = 'inner' + +export function isInnerToolCall(attributes: Readonly>): boolean { + return attributes[TOOL_CALL_LEVEL_ATTR] === INNER_TOOL_CALL_LEVEL +} + interface ToolIoInput { inputSource?: SourceReferences outputSource?: SourceReferences diff --git a/src/live.ts b/src/live.ts index 5025925..1cac7f3 100644 --- a/src/live.ts +++ b/src/live.ts @@ -1,4 +1,5 @@ import { createHash } from 'node:crypto' +import { isInnerToolCall } from './adapters/tool-io.js' import type { OtlpSpan } from './otlp.js' import type { PipelineReport } from './pipelines.js' import { runPipelines } from './pipelines.js' @@ -241,8 +242,9 @@ function isTool(span: OtlpSpan): boolean { return spanKind(span) === 'TOOL' || span.attributes['tool.name'] != null || span.name.startsWith('tool.') } +/** Command and file-change records inside a tool call carry tool I/O, not prose. */ function isTextSpan(span: OtlpSpan): boolean { - return !isTool(span) && spanContent(span).length > 0 + return !isTool(span) && !isInnerToolCall(span.attributes) && spanContent(span).length > 0 } function isVerification(span: OtlpSpan): boolean { diff --git a/src/session-relationship.ts b/src/session-relationship.ts index a71ce7c..49ac061 100644 --- a/src/session-relationship.ts +++ b/src/session-relationship.ts @@ -95,6 +95,8 @@ export function describeSessionRelationship( spans: readonly OtlpSpan[], ): SessionRelationship { const root = sessionRoot(ref, spans) + const sessionId = sessionIdFromAttributes(root?.attributes ?? {}) ?? root?.trace_id ?? ref.sessionId + const parentSessionId = stringAttribute(root, 'traces.parent_session_id') const childSessionIds = new Set() const spawnedChildSessionIds = new Set() const resumedChildSessionIds = new Set() @@ -125,8 +127,14 @@ export function describeSessionRelationship( } } + // A child that messages its parent (`send_message`, `followup_task`) names the + // parent as a target; the session and its parent are never its own children. + for (const ids of [childSessionIds, spawnedChildSessionIds, resumedChildSessionIds]) { + ids.delete(sessionId) + if (parentSessionId) ids.delete(parentSessionId) + } + const role = stringAttribute(root, 'traces.session.role') - const parentSessionId = stringAttribute(root, 'traces.parent_session_id') const depth = numberAttribute(root, 'traces.codex.agent_depth') const agentNickname = stringAttribute(root, 'traces.codex.agent_nickname') const agentRole = stringAttribute(root, 'traces.codex.agent_role') @@ -140,7 +148,7 @@ export function describeSessionRelationship( : undefined const turnId = stringAttribute(root, 'traces.codex.turn_id') return { - sessionId: sessionIdFromAttributes(root?.attributes ?? {}) ?? root?.trace_id ?? ref.sessionId, + sessionId, role: role === 'operator' || role === 'child' ? role : 'unknown', ...(parentSessionId ? { parentSessionId } : {}), childSessionIds: [...childSessionIds].sort(), diff --git a/tests/codex-command-facts.test.ts b/tests/codex-command-facts.test.ts new file mode 100644 index 0000000..b8a2418 --- /dev/null +++ b/tests/codex-command-facts.test.ts @@ -0,0 +1,301 @@ +/** + * Codex rollout facts that audit questions ask about: the commands a code-mode + * script ran, the files a patch changed, and which user turns a person typed. + * Every rollout comes from `tests/codex-facts-fixture.ts` and is synthetic. + */ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterAll, describe, expect, it } from 'vitest' +import { CodexAdapter } from '../src/adapters/codex.js' +import { buildPolicyEvidenceRecord } from '../src/evidence.js' +import type { OtlpSpan } from '../src/otlp.js' +import { runPipelines } from '../src/pipelines.js' +import { describeSessionRelationship } from '../src/session-relationship.js' +import { + at, + command, + commandItem, + ENVIRONMENT_BLOCK, + FIRST_REQUEST, + operatorRollout, + type RecordOrder, + type Row, + script, + scriptOutput, + task, + userEvent, + userItem, + userItemCompleted, + writeRollout as writeRolloutIn, +} from './codex-facts-fixture.js' + +const dir = mkdtempSync(join(tmpdir(), 'traces-codex-facts-')) +afterAll(() => rmSync(dir, { recursive: true, force: true })) + +const rollout = (name: string, order?: RecordOrder) => operatorRollout(dir, name, order) +const writeRollout = (name: string, rows: readonly Row[]) => writeRolloutIn(dir, name, rows) + +const kindOf = (item: OtlpSpan) => item.attributes['openinference.span.kind'] +const byName = (spans: readonly OtlpSpan[], name: string) => spans.filter((item) => item.name === name) +const inner = (spans: readonly OtlpSpan[]) => spans.filter((item) => item.attributes['traces.tool_call.level'] === 'inner') +const humanTurns = (spans: readonly OtlpSpan[]) => + byName(spans, 'user.prompt').filter((item) => item.attributes['tangle.actor'] === 'human') + +/** + * The same fixture parsed by the adapter this branch changes, at origin/main + * (7633ebc): no inner spans, and the two `` blocks plus the + * AGENTS.md block counted as human turns. Measured by running + * `tests/codex-facts-fixture.ts` against that checkout. + */ +const BASELINE = { total: 12, outerTools: 2, inner: 0, userPrompts: 6, humanTurns: 5 } as const + +function spanCounts(spans: readonly OtlpSpan[]) { + return { + total: spans.length, + outerTools: spans.filter((item) => kindOf(item) === 'TOOL').length, + inner: inner(spans).length, + userPrompts: byName(spans, 'user.prompt').length, + humanTurns: humanTurns(spans).length, + } +} + +describe('Codex command and file-change spans', () => { + it('emits each command inside a script with its own times, exit code, and process', async () => { + const spans = await new CodexAdapter().parse(rollout('commands')) + const scriptSpan = spans.find((item) => item.attributes['traces.codex.source_span_id'] === 'tool:call-script')! + const commands = byName(spans, 'command.execution') + + const joined = commands.filter((item) => item.parent_span_id === scriptSpan.span_id) + expect(joined.map((item) => ({ + input: JSON.parse(String(item.attributes['input.value'])).command.at(-1), + start: item.start_time, + end: item.end_time, + exit: item.attributes['process.exit_code'], + pid: item.attributes['traces.codex.process_id'], + status: item.status.code, + join: item.attributes['traces.codex.item_join'], + }))).toEqual([ + { input: 'gh pr create --fill', start: at(4.1), end: at(5), exit: 0, pid: '41001', status: 'OK', join: 'call' }, + { input: 'gh-drew pr merge 3 --squash', start: at(5.1), end: at(6), exit: 1, pid: '41002', status: 'ERROR', join: 'call' }, + { input: 'git status --short', start: at(6.1), end: at(6.5), exit: 0, pid: '41003', status: 'OK', join: 'call' }, + ]) + expect(joined[1]!.attributes['output.value']).toBe('X Pull request #3 is not mergeable\n') + expect(joined[1]!.status.message).toBe('command exited 1') + for (const item of commands) { + expect(kindOf(item)).toBe('CHAIN') + expect(item.attributes['span.type']).toBe('tool.execution') + expect(item.attributes['tool.name']).toBeUndefined() + expect(item.trace_id).toBe(scriptSpan.trace_id) + } + }) + + it('keeps a command that outlives its call under the session root', async () => { + const spans = await new CodexAdapter().parse(rollout('unmatched')) + const root = spans.find((item) => item.parent_span_id === null)! + const watch = byName(spans, 'command.execution') + .find((item) => String(item.attributes['input.value']).includes('pnpm test --watch=false'))! + expect(watch.parent_span_id).toBe(root.span_id) + expect(watch.attributes['traces.codex.item_join']).toBe('unmatched') + expect([watch.start_time, watch.end_time]).toEqual([at(6.8), at(9)]) + }) + + it('leaves a command inside two overlapping calls under the session root', async () => { + const spans = await new CodexAdapter().parse(writeRollout('ambiguous', [ + { t: 0, type: 'session_meta', payload: { id: 'facts-session', cwd: '/workspace/demo' } }, + task(1, 'task_started', 'turn-1'), + script(2, 'call-a', 'await tools.exec_command({ cmd: "pnpm test" })'), + script(3, 'call-b', 'await tools.exec_command({ cmd: "pnpm build" })'), + command(4, commandItem('item-shared', '42001', 'pnpm test', 0, 'ok\n'), { start: 3.5, end: 4 }), + scriptOutput(5, 'call-a', 'Script completed\nWall time 1.0 seconds\nOutput:\nok'), + scriptOutput(6, 'call-b', 'Script completed\nWall time 1.0 seconds\nOutput:\nok'), + task(7, 'task_complete', 'turn-1'), + ])) + const root = spans.find((item) => item.parent_span_id === null)! + const shared = byName(spans, 'command.execution') + expect(shared).toHaveLength(1) + expect(shared[0]!.attributes['traces.codex.item_join']).toBe('ambiguous') + expect(shared[0]!.parent_span_id).toBe(root.span_id) + }) + + it('emits the paths an apply_patch inside exec changed', async () => { + const spans = await new CodexAdapter().parse(rollout('patch')) + const patchCall = spans.find((item) => item.attributes['traces.codex.source_span_id'] === 'tool:call-patch')! + const files = byName(spans, 'file.change') + expect(files.map((item) => ({ + input: JSON.parse(String(item.attributes['input.value'])), + kind: item.attributes['traces.codex.file_change_kind'], + parent: item.parent_span_id, + status: item.status.code, + }))).toEqual([ + { input: { kind: 'update', path: '/workspace/demo/src/parser.ts' }, kind: 'update', parent: patchCall.span_id, status: 'OK' }, + { input: { kind: 'add', path: '/workspace/demo/tests/parser.test.ts' }, kind: 'add', parent: patchCall.span_id, status: 'OK' }, + ]) + expect(files.every((item) => item.start_time === at(10.2) && item.end_time === at(10.5))).toBe(true) + }) + + it('counts item shapes it cannot represent instead of guessing', async () => { + const spans = await new CodexAdapter().parse(rollout('skipped')) + const root = spans.find((item) => item.parent_span_id === null)! + expect(JSON.parse(String(root.attributes['traces.codex.skipped_item_counts']))).toEqual({ + 'CommandExecution:malformed': 1, + FixtureFutureItem: 1, + }) + expect(spans.some((item) => item.attributes['traces.codex.item_id'] === 'item-broken')).toBe(false) + }) + + it('marks a script whose receipt omits the Wall time colon as successful', async () => { + const spans = await new CodexAdapter().parse(rollout('receipt')) + const scriptSpan = spans.find((item) => item.attributes['traces.codex.source_span_id'] === 'tool:call-script')! + expect(scriptSpan.status.code).toBe('OK') + }) + + it('adds only inner spans: outer tool counts, evidence tool counts, and loop input stay unchanged', async () => { + const spans = await new CodexAdapter().parse(rollout('counts')) + expect(spanCounts(spans)).toEqual({ ...BASELINE, total: BASELINE.total + 6, inner: 6, humanTurns: 2 }) + + const ref = rollout('counts-evidence') + const outerOnly = spans.filter((item) => item.attributes['traces.tool_call.level'] !== 'inner') + const evidence = await buildPolicyEvidenceRecord(ref, spans, { generatedAt: at(0) }) + const outerEvidence = await buildPolicyEvidenceRecord(ref, outerOnly, { generatedAt: at(0) }) + expect(evidence.metrics).toMatchObject({ spanCount: BASELINE.total + 6, toolCallCount: 2, erroredToolCallCount: 0 }) + expect(evidence.metrics.tools).toEqual([ + { name: 'apply_patch', calls: 1, errors: 0 }, + { name: 'exec_command.verify', calls: 1, errors: 0 }, + ]) + const { spanCount: _all, ...toolMetrics } = evidence.metrics + const { spanCount: _outer, ...outerToolMetrics } = outerEvidence.metrics + expect(toolMetrics).toEqual(outerToolMetrics) + expect(evidence.signals).toEqual(outerEvidence.signals) + // agent-eval counts a failed inner command as one more execution error. + expect(evidence.execution.execution.executionErrors.events) + .toBe(outerEvidence.execution.execution.executionErrors.events + 1) + const [pipelines, outerPipelines] = await Promise.all([runPipelines(spans), runPipelines(outerOnly)]) + expect(pipelines.toolUse).toEqual(outerPipelines.toolUse) + expect(pipelines.stuckLoops.findings).toEqual(outerPipelines.stuckLoops.findings) + expect(pipelines.toolUse[0]).toMatchObject({ totalCalls: 2 }) + }) +}) + +describe('Codex human turns', () => { + it.each(['item-first', 'event-first'] as const)( + 'records each human turn once with its text and timestamp (%s)', + async (order) => { + const spans = await new CodexAdapter().parse(rollout(`turns-${order}`, order)) + const turns = humanTurns(spans) + expect(turns.map((item) => [item.attributes.content, item.start_time])).toEqual([ + [FIRST_REQUEST, at(2)], + ['ya?', at(21)], + ]) + expect(turns.every((item) => item.attributes['traces.codex.user_message_event'] === true)).toBe(true) + expect(byName(spans, 'user.prompt').filter((item) => item.attributes.content === FIRST_REQUEST)).toHaveLength(1) + }, + ) + + it('labels injected context blocks as non-human', async () => { + const spans = await new CodexAdapter().parse(rollout('injected')) + const injected = byName(spans, 'user.prompt') + .filter((item) => String(item.attributes.content).startsWith('')) + expect(injected).toHaveLength(3) + expect(injected.every((item) => item.attributes['tangle.actor'] === 'injected')).toBe(true) + expect(humanTurns(spans).at(-1)?.attributes.content).toBe('ya?') + }) + + it('treats a user-role message with no user_message event as injected once the event stream exists', async () => { + const spans = await new CodexAdapter().parse(writeRollout('unknown-wrapper', [ + { t: 0, type: 'session_meta', payload: { id: 'wrapper-session', cwd: '/workspace/demo' } }, + task(1, 'task_started', 'turn-1'), + userItem(2, 'Please add a changelog entry.'), + userEvent(2.001, 'Please add a changelog entry.'), + userItem(3, 'harness text'), + task(4, 'task_complete', 'turn-1'), + ])) + expect(byName(spans, 'user.prompt').map((item) => [item.attributes.content, item.attributes['tangle.actor']])).toEqual([ + ['Please add a changelog entry.', 'human'], + ['harness text', 'injected'], + ]) + }) + + it('records a turn reported as an item_completed UserMessage once', async () => { + const spans = await new CodexAdapter().parse(writeRollout('item-completed-turns', [ + { t: 0, type: 'session_meta', payload: { id: 'facts-session', cwd: '/workspace/demo' } }, + userItem(0.5, [{ type: 'input_text', text: ENVIRONMENT_BLOCK }]), + task(1, 'task_started', 'turn-1'), + userItem(2, [{ type: 'input_text', text: 'Rerun the parser tests.' }]), + userItemCompleted(2.001, 'item-turn-1', 'Rerun the parser tests.'), + task(3, 'task_complete', 'turn-1'), + ])) + expect(byName(spans, 'user.prompt').map((item) => [item.attributes.content, item.attributes['tangle.actor']])).toEqual([ + [ENVIRONMENT_BLOCK, 'injected'], + ['Rerun the parser tests.', 'human'], + ]) + expect(humanTurns(spans)[0]!.start_time).toBe(at(2)) + expect(humanTurns(spans)[0]!.attributes['traces.codex.user_message_event']).toBe(true) + const root = spans.find((item) => item.parent_span_id === null)! + expect(root.attributes['traces.codex.skipped_item_counts']).toBeUndefined() + }) + + it('pairs the two records of a turn whose message record carries a context prefix', async () => { + const typed = 'Fix the parser and rerun the tests.' + const prefixed = `${ENVIRONMENT_BLOCK}\n\n## My request for Codex:\n\n${typed}` + const spans = await new CodexAdapter().parse(writeRollout('prefixed-turn', [ + { t: 0, type: 'session_meta', payload: { id: 'facts-session', cwd: '/workspace/demo' } }, + task(1, 'task_started', 'turn-1'), + userItem(2, [{ type: 'input_text', text: prefixed }]), + userEvent(2.001, typed), + task(3, 'task_complete', 'turn-1'), + ])) + expect(byName(spans, 'user.prompt')).toHaveLength(1) + expect(humanTurns(spans).map((item) => [item.attributes.content, item.start_time])).toEqual([[prefixed, at(2)]]) + }) + + it('keeps text heuristics for rollouts that never recorded user_message events', async () => { + const spans = await new CodexAdapter().parse(writeRollout('legacy-turns', [ + { t: 0, type: 'session_meta', payload: { id: 'legacy-session', cwd: '/workspace/demo' } }, + task(1, 'task_started', 'turn-1'), + userItem(2, 'Please add a changelog entry.'), + task(3, 'task_complete', 'turn-1'), + ])) + expect(humanTurns(spans).map((item) => item.attributes.content)).toEqual(['Please add a changelog entry.']) + }) +}) + +describe('Codex child relationships', () => { + it('does not list the parent among the children a forked child messages', async () => { + const parentId = '019f0000-0000-7000-8000-00000000aaaa' + const childId = '019f0000-0000-7000-8000-00000000bbbb' + const siblingId = '019f0000-0000-7000-8000-00000000cccc' + const ref = writeRollout('forked-child', [ + { + t: 0, + type: 'session_meta', + payload: { + id: childId, + parent_thread_id: parentId, + thread_source: 'subagent', + cwd: '/workspace/demo', + source: { subagent: { thread_spawn: { parent_thread_id: parentId, depth: 1, agent_path: '/root/worker' } } }, + }, + }, + task(1, 'task_started', 'child-turn'), + userItem(2, 'Check the parser and report back.'), + { + t: 3, + type: 'response_item', + payload: { type: 'function_call', call_id: 'call-report', name: 'send_message', arguments: JSON.stringify({ target: parentId, message: 'done' }) }, + }, + { t: 4, type: 'response_item', payload: { type: 'function_call_output', call_id: 'call-report', output: '{"ok":true}' } }, + { + t: 5, + type: 'response_item', + payload: { type: 'function_call', call_id: 'call-sibling', name: 'send_message', arguments: JSON.stringify({ target: siblingId, message: 'fyi' }) }, + }, + { t: 6, type: 'response_item', payload: { type: 'function_call_output', call_id: 'call-sibling', output: '{"ok":true}' } }, + task(7, 'task_complete', 'child-turn'), + ]) + const relationship = describeSessionRelationship({ ...ref, sessionId: childId }, await new CodexAdapter().parse(ref)) + expect(relationship.parentSessionId).toBe(parentId) + expect(relationship.childSessionIds).toEqual([siblingId]) + expect(relationship.resumedChildSessionIds).toEqual([siblingId]) + }) +}) diff --git a/tests/codex-facts-fixture.ts b/tests/codex-facts-fixture.ts new file mode 100644 index 0000000..4b2155d --- /dev/null +++ b/tests/codex-facts-fixture.ts @@ -0,0 +1,204 @@ +/** + * Synthetic Codex rollouts for the facts an audit question asks about: the + * commands a code-mode script ran, the files a patch changed, and which user + * turns a person typed. No rollout here comes from a recorded session. + * + * `tests/codex-command-facts.test.ts` asserts against these rollouts. Keeping + * them in their own module also lets a reader parse the same fixture with an + * older checkout of the adapter to compare span counts. + */ +import { writeFileSync } from 'node:fs' +import { join } from 'node:path' +import type { SessionRef } from '../src/types.js' + +export const BASE_MS = Date.UTC(2026, 8, 8, 10, 0, 0) +/** The order in which a rollout recorded the two copies of one submitted turn. */ +export type RecordOrder = 'item-first' | 'event-first' + +export const at = (seconds: number): string => new Date(BASE_MS + seconds * 1000).toISOString() +export const ms = (seconds: number): number => BASE_MS + seconds * 1000 + +export type Row = { readonly t: number } & Record + +export function writeRollout(dir: string, name: string, rows: readonly Row[]): SessionRef { + const path = join(dir, `rollout-${name}.jsonl`) + writeFileSync(path, rows.map(({ t, ...row }) => JSON.stringify({ timestamp: at(t), ...row })).join('\n')) + return { harness: 'codex', sessionId: name, path, cwd: null, mtimeMs: 0 } +} + +export const userItem = (t: number, content: unknown): Row => ({ + t, + type: 'response_item', + payload: { type: 'message', role: 'user', content }, +}) +export const userEvent = (t: number, message: string): Row => ({ + t, + type: 'event_msg', + payload: { type: 'user_message', message, images: [], local_images: [] }, +}) +/** The current rollout shape for a submitted turn: an `item_completed` UserMessage item. */ +export const userItemCompleted = (t: number, id: string, text: string): Row => ({ + t, + type: 'event_msg', + payload: { + type: 'item_completed', + thread_id: 'facts-session', + turn_id: 'turn-1', + started_at_ms: ms(t), + completed_at_ms: ms(t), + item: { type: 'UserMessage', id, content: [{ type: 'text', text }] }, + }, +}) +export const task = (t: number, kind: 'task_started' | 'task_complete', turnId: string): Row => ({ + t, + type: 'event_msg', + payload: { type: kind, turn_id: turnId }, +}) +export const tokens = (t: number, input: number): Row => ({ + t, + type: 'event_msg', + payload: { + type: 'token_count', + info: { + last_token_usage: { input_tokens: input, output_tokens: 20 }, + total_token_usage: { input_tokens: input * 2, output_tokens: 40 }, + }, + }, +}) +export const script = (t: number, callId: string, input: string): Row => ({ + t, + type: 'response_item', + payload: { type: 'custom_tool_call', call_id: callId, name: 'exec', input }, +}) +export const scriptOutput = (t: number, callId: string, output: string): Row => ({ + t, + type: 'response_item', + payload: { type: 'custom_tool_call_output', call_id: callId, output }, +}) +export const command = ( + t: number, + item: Record, + window: { start?: number; end?: number } = {}, +): Row => ({ + t, + type: 'event_msg', + payload: { + type: 'item_completed', + thread_id: 'facts-session', + turn_id: 'turn-1', + ...(window.start === undefined ? {} : { started_at_ms: ms(window.start) }), + ...(window.end === undefined ? {} : { completed_at_ms: ms(window.end) }), + item: { type: 'CommandExecution', ...item }, + }, +}) +export const commandItem = ( + id: string, + processId: string, + script: string, + exitCode: number, + output: string, +): Record => ({ + id, + process_id: processId, + command: ['/bin/zsh', '-lc', script], + cwd: '/workspace/demo', + parsed_cmd: [{ type: 'unknown', cmd: script }], + source: 'agent', + status: exitCode === 0 ? 'completed' : 'failed', + stdout: output, + stderr: '', + aggregated_output: output, + exit_code: exitCode, + duration: { secs: 0, nanos: 800_000_000 }, + formatted_output: output, +}) + +export const AGENTS_BLOCK = '# AGENTS.md instructions for /workspace/demo\n\n\nUse pnpm.\n' +export const ENVIRONMENT_BLOCK = '\n /workspace/demo\n zsh\n' +export const FIRST_REQUEST = 'Open a PR for the parser fix, merge PR 3, then tell me what git status reports.' +export const SCRIPT_INPUT = [ + 'const created = await tools.exec_command({ cmd: "gh pr create --fill" })', + 'const merged = await tools.exec_command({ cmd: "gh-drew pr merge 3 --squash" })', + 'const status = await tools.exec_command({ cmd: "git status --short" })', + 'text([created.output, merged.output, status.output].join("\\n"))', +].join('\n') +export const PATCH_INPUT = [ + 'await tools.apply_patch(`*** Begin Patch', + '*** Update File: src/parser.ts', + '@@', + '-export const mode = "old"', + '+export const mode = "new"', + '*** Add File: tests/parser.test.ts', + '+test("mode", () => {})', + '*** End Patch`)', +].join('\n') + +/** + * One operator session: injected context, a substantive request logged twice, + * a code-mode script around three commands, a command that outlives its call, + * a patch applied inside `exec`, two item shapes the adapter cannot represent, + * and a short last typed turn followed by one more injected block. + */ +export function operatorRollout(dir: string, name: string, order: RecordOrder = 'item-first'): SessionRef { + const request = order === 'item-first' + ? [userItem(2, [{ type: 'input_text', text: FIRST_REQUEST }]), userEvent(2.001, FIRST_REQUEST)] + : [userEvent(2, FIRST_REQUEST), userItem(2.001, [{ type: 'input_text', text: FIRST_REQUEST }])] + const followUp = order === 'item-first' + ? [userItem(21, [{ type: 'input_text', text: 'ya?' }]), userEvent(21.001, 'ya?')] + : [userEvent(21, 'ya?'), userItem(21.001, [{ type: 'input_text', text: 'ya?' }])] + return writeRollout(dir, name, [ + { t: 0, type: 'session_meta', payload: { id: 'facts-session', cwd: '/workspace/demo' } }, + { t: 0.5, type: 'turn_context', payload: { cwd: '/workspace/demo', model: 'gpt-fixture' } }, + userItem(0.6, [{ type: 'input_text', text: AGENTS_BLOCK }]), + userItem(0.7, [{ type: 'input_text', text: ENVIRONMENT_BLOCK }]), + task(1, 'task_started', 'turn-1'), + ...request, + tokens(3, 1000), + script(4, 'call-script', SCRIPT_INPUT), + command(5, commandItem('item-create', '41001', 'gh pr create --fill', 0, 'https://example.test/demo/pull/7\n'), { start: 4.1, end: 5 }), + command(6, commandItem('item-merge', '41002', 'gh-drew pr merge 3 --squash', 1, 'X Pull request #3 is not mergeable\n'), { start: 5.1, end: 6 }), + command(6.5, commandItem('item-status', '41003', 'git status --short', 0, ' M src/parser.ts\n'), { start: 6.1, end: 6.5 }), + scriptOutput(7, 'call-script', 'Script completed\nWall time 3.0 seconds\nOutput:\nhttps://example.test/demo/pull/7'), + command(9, commandItem('item-watch', '41004', 'pnpm test --watch=false', 0, 'Tests 12 passed\n'), { start: 6.8, end: 9 }), + tokens(9.5, 1400), + script(10, 'call-patch', PATCH_INPUT), + { + t: 10.5, + type: 'event_msg', + payload: { + type: 'item_completed', + thread_id: 'facts-session', + turn_id: 'turn-1', + started_at_ms: ms(10.2), + completed_at_ms: ms(10.5), + item: { + type: 'FileChange', + id: 'item-patch', + changes: { + '/workspace/demo/src/parser.ts': { + type: 'update', + unified_diff: '@@ -1 +1 @@\n-export const mode = "old"\n+export const mode = "new"\n', + move_path: null, + }, + '/workspace/demo/tests/parser.test.ts': { type: 'add', content: 'test("mode", () => {})\n' }, + }, + status: 'completed', + stdout: 'Success. Updated the following files:\nM src/parser.ts\nA tests/parser.test.ts\n', + stderr: '', + }, + }, + }, + scriptOutput(11, 'call-patch', 'Script completed\nWall time 0.3 seconds\nOutput:\nSuccess.'), + command(11.5, { id: 'item-broken', status: 'completed', exit_code: 0 }, { start: 11.2, end: 11.5 }), + { t: 11.6, type: 'event_msg', payload: { type: 'item_completed', turn_id: 'turn-1', completed_at_ms: ms(11.6), item: { type: 'FixtureFutureItem', id: 'item-future' } } }, + task(12, 'task_complete', 'turn-1'), + userItem(19, [{ type: 'input_text', text: ENVIRONMENT_BLOCK.replace('zsh', 'bash') }]), + task(20, 'task_started', 'turn-2'), + ...followUp, + tokens(22, 1500), + // Injected after the last typed turn: a reader that trusts the user role + // reports this block as the session's last human turn. + userItem(22.5, [{ type: 'input_text', text: ENVIRONMENT_BLOCK.replace('zsh', 'fish') }]), + task(23, 'task_complete', 'turn-2'), + ]) +} diff --git a/tests/codex-tool-status.test.ts b/tests/codex-tool-status.test.ts index a059aa7..789f412 100644 --- a/tests/codex-tool-status.test.ts +++ b/tests/codex-tool-status.test.ts @@ -57,6 +57,9 @@ describe.each(['function', 'custom'] as const)('Codex %s tool outcomes', (varian { label: 'explicit failure with optimistic stdout', output: { exit_code: 1, output: 'success' }, code: 'ERROR' }, { label: 'initial process header', output: 'Process exited with code 0\nOutput:\nerror: command failed ENOENT', code: 'OK' }, { label: 'initial failed process header', output: 'Process exited with code 1\nOutput:\nsuccess', code: 'ERROR' }, + { label: 'script receipt with wall-time colon', output: 'Script completed\nWall time: 1.2 seconds\nOutput:\nerror: command failed ENOENT', code: 'OK' }, + { label: 'script receipt without wall-time colon', output: 'Script completed\nWall time 1.2 seconds\nOutput:', code: 'OK' }, + { label: 'failed script receipt without wall-time colon', output: 'Script failed\nWall time 1.2 seconds\nOutput:\nsuccess', code: 'ERROR' }, { label: 'initial script error receipt', output: 'Script error:\nExit code: 1\nOutput:\nsuccess', code: 'ERROR' }, { label: 'initial script zero exit receipt', output: 'Script error:\nExit code: 0\nOutput:\nerror: source code', code: 'OK' }, { label: 'script receipt with CRLF', output: 'Script error:\r\nExit code: -1\r\nOutput:\r\nsuccess', code: 'ERROR' }, From bbf1ba5dfbb797855ae03a8cca581310c7f0d47b Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Thu, 10 Sep 2026 01:49:30 -0700 Subject: [PATCH 2/2] fix(codex): bound adapter memory by span count, not command output Build each command and file-change span when its item_completed record is read, so `toolIoAttributes` caps the retained command text and output at 16 KiB there, and keep only {span, startMs, endMs} until the tool-call windows are complete. Retaining the parsed items instead made adapter heap grow with total raw command output: a 100 MB rollout of command items dies under a 64 MB heap, which `tests/adapters.test.ts` now enforces for Codex the way it already does for Claude. Output is unchanged; only the join and the parent span depend on records written after the item. Also from review: - Split `traces.codex.skipped_item_counts` into `traces.codex.unmodeled_item_counts` (item types the adapter models no span for, whose work the response_item stream records) and `traces.codex.dropped_item_counts` (a modeled item that produced no span: malformed, duplicate, or no text). Only the second reads as lost facts, so they no longer share one count. - Scope the item dedupe key by task index, so a turn-scoped id scheme cannot make a later turn's command a `:duplicate`. - Emit `process.exit_code` additively on codex-exec tool spans, so one fact has one spelling across both Codex adapters. - Assert `traces.codex.user_message_event` in the prefix-pairing test, so it fails without the submitted-turn record. Co-Authored-By: Claude Opus 5 (1M context) --- src/adapters/codex-exec.ts | 3 + src/adapters/codex-format.ts | 25 ++++--- src/adapters/codex.ts | 114 ++++++++++++++++++++---------- tests/adapters.test.ts | 96 +++++++++++++++++++++++++ tests/codex-command-facts.test.ts | 10 +-- tests/codex-exec.test.ts | 1 + 6 files changed, 199 insertions(+), 50 deletions(-) diff --git a/src/adapters/codex-exec.ts b/src/adapters/codex-exec.ts index 156746a..d0cb8c6 100644 --- a/src/adapters/codex-exec.ts +++ b/src/adapters/codex-exec.ts @@ -302,6 +302,9 @@ export class CodexExecAdapter implements HarnessTraceAdapter { pending.span.attributes['traces.codex.exec_item_status'] = item.status if (result.exitCode !== undefined) { pending.span.attributes['traces.codex.exec_exit_code'] = result.exitCode + // One spelling for one fact: the rollout adapter's command spans use + // this key, so a reader of exit codes needs no per-adapter branch. + pending.span.attributes['process.exit_code'] = result.exitCode } if (type === 'command_execution') { recordToolOutput(pending.span, typeof item.aggregated_output === 'string' ? item.aggregated_output : undefined, sourceOf(item, 'aggregated_output')) diff --git a/src/adapters/codex-format.ts b/src/adapters/codex-format.ts index ee7f328..2f58184 100644 --- a/src/adapters/codex-format.ts +++ b/src/adapters/codex-format.ts @@ -162,15 +162,24 @@ export interface CodexUserMessage { } /** - * One `item_completed` event, normalized. `skipped` carries the label the adapter - * counts when an item produces no span: the item type, or `:malformed` - * when a required field is missing. + * One `item_completed` event, normalized. + * + * `skipped` separates the two reasons an item produces no span, because they + * carry opposite meanings for a reader counting facts: + * + * - `unmodeled` — the adapter builds no span from this item type. The rollout + * records the same work as a `response_item` (reasoning, assistant messages, + * tool calls), and that record is what becomes a span, so the count is a + * census of item types, not missing facts. + * - `dropped` — an item of a modeled type produced no span: a required field + * was missing or mistyped, the item repeats one already recorded, or a user + * message carried no text. This count is the one that reads as lost facts. */ export type CodexCompletedItem = | { readonly type: 'CommandExecution'; readonly item: object; readonly command: CodexCommandExecution } | { readonly type: 'FileChange'; readonly item: object; readonly fileChange: CodexFileChange } | { readonly type: 'UserMessage'; readonly item: object; readonly userMessage: CodexUserMessage } - | { readonly type: 'skipped'; readonly label: string } + | { readonly type: 'skipped'; readonly reason: 'unmodeled' | 'dropped'; readonly label: string } type JsonRecord = Record @@ -259,13 +268,13 @@ export function codexCompletedItem(line: CodexLine): CodexCompletedItem | undefi const item = recordValue(payload.item) const type = nonEmptyString(item?.type) ?? 'unknown' if (!item || (type !== 'CommandExecution' && type !== 'FileChange' && type !== 'UserMessage')) { - return { type: 'skipped', label: type } + return { type: 'skipped', reason: 'unmodeled', label: type } } const itemId = nonEmptyString(item.id) if (type === 'UserMessage') { const text = userMessageText(item.content) // An image-only or audio-only turn carries no text to record as a turn span. - if (!itemId || !text) return { type: 'skipped', label: `${type}:${itemId ? 'no_text' : 'malformed'}` } + if (!itemId || !text) return { type: 'skipped', reason: 'dropped', label: `${type}:${itemId ? 'no_text' : 'malformed'}` } return { type, item, userMessage: { itemId, text } } } const startedAtMs = epochMs(payload.started_at_ms) ?? epochMs(item.started_at_ms) @@ -277,11 +286,11 @@ export function codexCompletedItem(line: CodexLine): CodexCompletedItem | undefi } if (type === 'FileChange') { const changes = fileChangeEntries(item.changes) - if (!itemId || !changes) return { type: 'skipped', label: `${type}:malformed` } + if (!itemId || !changes) return { type: 'skipped', reason: 'dropped', label: `${type}:malformed` } return { type, item, fileChange: { itemId, changes, ...(status ? { status } : {}), ...times } } } const command = commandValue(item.command) - if (!itemId || !command) return { type: 'skipped', label: `${type}:malformed` } + if (!itemId || !command) return { type: 'skipped', reason: 'dropped', label: `${type}:malformed` } const cwd = nonEmptyString(item.cwd) const processId = typeof item.process_id === 'number' && Number.isSafeInteger(item.process_id) ? String(item.process_id) diff --git a/src/adapters/codex.ts b/src/adapters/codex.ts index 6edd98d..d62d43f 100644 --- a/src/adapters/codex.ts +++ b/src/adapters/codex.ts @@ -37,7 +37,6 @@ import { codexActor } from './actor.js' import { ACTOR_ATTR, capText, userPromptSpan } from './conversation.js' import { type CodexCommandExecution, - type CodexCompletedItem, codexCompletedItem, type CodexFileChange, type CodexLine, @@ -386,10 +385,21 @@ function closeSpanAt(target: OtlpSpan, sourceEndTime: string): void { target.end_time = sourceEndTime } -/** A completed item the adapter turns into spans; user messages take the turn path instead. */ -type CompletedItem = - & Extract - & { readonly recordTime: string } +/** The item types the adapter turns into inner spans; user messages take the turn path instead. */ +type InnerItemType = 'CommandExecution' | 'FileChange' + +/** + * An inner span built at the moment its item was read, waiting only for the + * complete map of tool-call windows to place it. Retaining the span rather than + * the parsed item keeps adapter memory bounded by span count instead of by + * total raw command output: `toolIoAttributes` already capped the command text + * and the command output when the span was built. + */ +interface PendingInnerSpan { + readonly span: OtlpSpan + readonly startMs: number + readonly endMs: number +} /** The time window of a model-issued call: its call record to its output record. */ interface ToolWindow { @@ -447,9 +457,8 @@ function itemSources(item: object, fields: readonly string[]) { } function innerItemAttributes( - type: CompletedItem['type'], + type: InnerItemType, itemId: string, - join: ItemJoin, timeSource: string | undefined, status: string | undefined, ): Record { @@ -459,7 +468,9 @@ function innerItemAttributes( [TOOL_CALL_LEVEL_ATTR]: INNER_TOOL_CALL_LEVEL, 'traces.codex.item_type': type, 'traces.codex.item_id': itemId, - 'traces.codex.item_join': join, + // `placeInnerSpans` decides the join once every tool window is known. An + // item that matches no window keeps this value. + 'traces.codex.item_join': 'unmatched', ...(timeSource ? { 'traces.codex.item_time_source': timeSource } : {}), ...(status ? { 'traces.codex.item_status': status } : {}), } @@ -468,21 +479,42 @@ function innerItemAttributes( interface ItemSpanContext { readonly traceId: string readonly rootId: string - readonly windows: ReadonlyMap +} + +function itemCountsJson(counts: ReadonlyMap): string { + return JSON.stringify( + Object.fromEntries([...counts].sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))), + ) +} + +/** + * Attach each inner span to the model-issued call whose window contains it. + * This is the only part of an inner span that depends on records written after + * the item, which is why the span itself is built when the item is read. + */ +function placeInnerSpans( + pending: readonly PendingInnerSpan[], + windows: ReadonlyMap, +): OtlpSpan[] { + return pending.map((entry) => { + const { parent, join } = joinItem(windows, entry.startMs, entry.endMs) + if (parent) entry.span.parent_span_id = parent.span_id + entry.span.attributes['traces.codex.item_join'] = join + return entry.span + }) } /** * One CHAIN span per command. Command text, cwd, and output stay in the tool * I/O keys, which metadata-only upload strips and external redactors scrub. */ -function commandSpan(context: ItemSpanContext, item: object, command: CodexCommandExecution, recordTime: string): OtlpSpan { +function commandSpan(context: ItemSpanContext, item: object, command: CodexCommandExecution, recordTime: string): PendingInnerSpan { const { start, end, timeSource } = itemTimes(command, recordTime) - const { parent, join } = joinItem(context.windows, Date.parse(start), Date.parse(end)) const status = itemStatus(command.status, 'command', command.exitCode) const commandSpan = span({ traceId: context.traceId, spanId: `command:${command.itemId}`, - parentSpanId: parent?.span_id ?? context.rootId, + parentSpanId: context.rootId, name: 'command.execution', kind: 'CHAIN', startTime: start, @@ -497,26 +529,27 @@ function commandSpan(context: ItemSpanContext, item: object, command: CodexComma output: command.output, outputSource: itemSources(item, command.outputFields), }), - ...innerItemAttributes('CommandExecution', command.itemId, join, timeSource, command.status), + ...innerItemAttributes('CommandExecution', command.itemId, timeSource, command.status), ...(command.exitCode === undefined ? {} : { 'process.exit_code': command.exitCode }), ...(command.processId ? { 'traces.codex.process_id': command.processId } : {}), ...(command.source ? { 'traces.codex.command_source': command.source } : {}), }, }) closeSpanAt(commandSpan, end) - return commandSpan + return { span: commandSpan, startMs: Date.parse(start), endMs: Date.parse(end) } } /** One CHAIN span per changed path; the path stays in `input.value` for the same reason as commands. */ -function fileChangeSpans(context: ItemSpanContext, item: object, fileChange: CodexFileChange, recordTime: string): OtlpSpan[] { +function fileChangeSpans(context: ItemSpanContext, item: object, fileChange: CodexFileChange, recordTime: string): PendingInnerSpan[] { const { start, end, timeSource } = itemTimes(fileChange, recordTime) - const { parent, join } = joinItem(context.windows, Date.parse(start), Date.parse(end)) const status = itemStatus(fileChange.status, 'file change') + const startMs = Date.parse(start) + const endMs = Date.parse(end) return fileChange.changes.map((change, index) => { const changeSpan = span({ traceId: context.traceId, spanId: `file-change:${fileChange.itemId}:${index}`, - parentSpanId: parent?.span_id ?? context.rootId, + parentSpanId: context.rootId, name: 'file.change', kind: 'CHAIN', startTime: start, @@ -529,12 +562,12 @@ function fileChangeSpans(context: ItemSpanContext, item: object, fileChange: Cod input: { path: change.path, kind: change.kind, ...(change.movePath ? { move_path: change.movePath } : {}) }, inputSource: sourceOf(item, 'changes'), }), - ...innerItemAttributes('FileChange', fileChange.itemId, join, timeSource, fileChange.status), + ...innerItemAttributes('FileChange', fileChange.itemId, timeSource, fileChange.status), 'traces.codex.file_change_kind': change.kind, }, }) closeSpanAt(changeSpan, end) - return changeSpan + return { span: changeSpan, startMs, endMs } }) } @@ -856,11 +889,16 @@ export class CodexAdapter implements HarnessTraceAdapter { let lastTimestamp: string | undefined const awaitingModel = model ? [] : [root] const toolWindows = new Map() - const completedItems: CompletedItem[] = [] + const itemContext: ItemSpanContext = { traceId, rootId } + const pendingInnerSpans: PendingInnerSpan[] = [] const completedItemKeys = new Set() - const skippedItemCounts = new Map() - const countSkippedItem = (label: string): void => { - skippedItemCounts.set(label, (skippedItemCounts.get(label) ?? 0) + 1) + // Two separate censuses: item types this adapter models no span for, and + // items of a modeled type that produced none. Only the second reads as + // lost facts, so they never share one count. + const unmodeledItemCounts = new Map() + const droppedItemCounts = new Map() + const countItem = (counts: Map, label: string): void => { + counts.set(label, (counts.get(label) ?? 0) + 1) } // Pairs the two records of one typed turn. A task index scopes the pairing, // so the same short reply in two turns stays two turns. @@ -1119,16 +1157,22 @@ export class CodexAdapter implements HarnessTraceAdapter { if (!activity) { const completed = codexCompletedItem(l) if (completed?.type === 'skipped') { - countSkippedItem(completed.label) + countItem(completed.reason === 'unmodeled' ? unmodeledItemCounts : droppedItemCounts, completed.label) } else if (completed?.type === 'UserMessage') { recordSubmittedTurn(completed.userMessage.text, ts, textSources(completed.item, 'content')) } else if (completed) { const itemId = completed.type === 'CommandExecution' ? completed.command.itemId : completed.fileChange.itemId - const itemKey = `${completed.type}:${itemId}` - if (completedItemKeys.has(itemKey)) countSkippedItem(`${completed.type}:duplicate`) + // Scoped by task: Codex item ids are UUIDs today, but a turn-scoped + // id scheme must not make a later turn's command a `:duplicate`. + const itemKey = `${completed.type}:${taskIndex}:${itemId}` + if (completedItemKeys.has(itemKey)) countItem(droppedItemCounts, `${completed.type}:duplicate`) else { completedItemKeys.add(itemKey) - completedItems.push({ ...completed, recordTime: ts }) + if (completed.type === 'CommandExecution') { + pendingInnerSpans.push(commandSpan(itemContext, completed.item, completed.command, ts)) + } else { + pendingInnerSpans.push(...fileChangeSpans(itemContext, completed.item, completed.fileChange, ts)) + } } } continue @@ -1280,18 +1324,12 @@ export class CodexAdapter implements HarnessTraceAdapter { candidate.span.attributes[ACTOR_ATTR] = 'injected' candidate.span.attributes['traces.codex.actor_evidence'] = 'no_user_message_event' } - const itemContext: ItemSpanContext = { traceId, rootId, windows: toolWindows } - for (const completed of completedItems) { - if (completed.type === 'CommandExecution') { - spans.push(commandSpan(itemContext, completed.item, completed.command, completed.recordTime)) - } else { - spans.push(...fileChangeSpans(itemContext, completed.item, completed.fileChange, completed.recordTime)) - } + spans.push(...placeInnerSpans(pendingInnerSpans, toolWindows)) + if (unmodeledItemCounts.size > 0) { + root.attributes['traces.codex.unmodeled_item_counts'] = itemCountsJson(unmodeledItemCounts) } - if (skippedItemCounts.size > 0) { - root.attributes['traces.codex.skipped_item_counts'] = JSON.stringify( - Object.fromEntries([...skippedItemCounts].sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))), - ) + if (droppedItemCounts.size > 0) { + root.attributes['traces.codex.dropped_item_counts'] = itemCountsJson(droppedItemCounts) } if (selectedBoundary?.turnId) { for (const item of spans) item.attributes['traces.codex.turn_id'] ??= selectedBoundary.turnId diff --git a/tests/adapters.test.ts b/tests/adapters.test.ts index 3634b6b..6498ed3 100644 --- a/tests/adapters.test.ts +++ b/tests/adapters.test.ts @@ -307,6 +307,102 @@ describe('JSONL adapter streaming', () => { expect(result.maxRssKb).toBeLessThan(maxRssMb * 1024) }) + it('parses a 100 MB Codex rollout of command items within a bounded heap', () => { + const path = join(dir, 'large-codex-items.jsonl') + const output = 'y'.repeat(1024 * 1024) + const itemCount = 101 + const at = (second: number) => new Date(Date.UTC(2026, 0, 1, 0, 0, second)).toISOString() + const file = openSync(path, 'w') + try { + writeSync(file, `${JSON.stringify({ + type: 'session_meta', + timestamp: at(0), + payload: { id: 'large-codex', cwd: '/workspace/demo' }, + })}\n`) + writeSync(file, `${JSON.stringify({ + type: 'response_item', + timestamp: at(1), + payload: { type: 'custom_tool_call', call_id: 'call-0', name: 'exec', input: 'await tools.exec_command({ cmd: "pnpm test" })' }, + })}\n`) + for (let index = 0; index < itemCount; index += 1) { + writeSync(file, `${JSON.stringify({ + type: 'event_msg', + timestamp: at(2 + index), + payload: { + type: 'item_completed', + item: { + id: `item-${index}`, + type: 'CommandExecution', + command: ['pnpm', 'test', `--shard=${index}`], + cwd: '/workspace/demo', + status: 'completed', + exit_code: 0, + aggregated_output: `${index}:${output}`, + }, + }, + })}\n`) + } + writeSync(file, `${JSON.stringify({ + type: 'response_item', + timestamp: at(2 + itemCount), + payload: { type: 'custom_tool_call_output', call_id: 'call-0', output: 'Process exited with code 0' }, + })}\n`) + } finally { + closeSync(file) + } + expect(statSync(path).size).toBeGreaterThan(100 * 1024 * 1024) + + const adapterUrl = pathToFileURL(join(process.cwd(), 'src/adapters/codex.ts')).href + const childSource = ` + import { CodexAdapter } from ${JSON.stringify(adapterUrl)} + const ref = { + harness: 'codex', + sessionId: 'large-codex', + path: ${JSON.stringify(path)}, + cwd: null, + mtimeMs: 0, + } + const spans = await new CodexAdapter().parse(ref) + const commands = spans.filter((span) => span.attributes['traces.codex.item_type'] === 'CommandExecution') + process.stdout.write(JSON.stringify({ + spanCount: spans.length, + commandCount: commands.length, + joinedToCall: commands.filter((span) => span.attributes['traces.codex.item_join'] === 'call').length, + maxRssKb: process.resourceUsage().maxRSS, + bounded: commands.every((span) => + span.attributes['traces.output.truncated'] === true && + Buffer.byteLength(String(span.attributes['input.value'])) <= 16 * 1024 && + Buffer.byteLength(String(span.attributes['output.value'])) <= 16 * 1024 + ), + })) + ` + const env: NodeJS.ProcessEnv = { ...process.env, FORCE_COLOR: '0' } + delete env.NODE_OPTIONS + // Retaining the parsed items instead of the built spans exhausts this cap. + const childHeapLimitMb = 64 + const maxRssMb = 224 + const child = spawnSync( + process.execPath, + [`--max-old-space-size=${childHeapLimitMb}`, '--max-semi-space-size=1', '--import', 'tsx', '--input-type=module', '--eval', childSource], + { cwd: process.cwd(), encoding: 'utf8', env, timeout: 60_000 }, + ) + + expect(child.status, child.stderr || child.error?.message).toBe(0) + const result = JSON.parse(child.stdout) as { + spanCount: number + commandCount: number + joinedToCall: number + maxRssKb: number + bounded: boolean + } + expect(result).toMatchObject({ + commandCount: itemCount, + joinedToCall: itemCount, + bounded: true, + }) + expect(result.maxRssKb).toBeLessThan(maxRssMb * 1024) + }) + it('customer session parsing retains valid Codex records and stamps a degraded receipt', async () => { const path = join(dir, 'codex-recovered-session.jsonl') const rawSecret = 'secret-corrupt-codex-record' diff --git a/tests/codex-command-facts.test.ts b/tests/codex-command-facts.test.ts index b8a2418..ebc4238 100644 --- a/tests/codex-command-facts.test.ts +++ b/tests/codex-command-facts.test.ts @@ -134,12 +134,12 @@ describe('Codex command and file-change spans', () => { expect(files.every((item) => item.start_time === at(10.2) && item.end_time === at(10.5))).toBe(true) }) - it('counts item shapes it cannot represent instead of guessing', async () => { + it('separates an item type it models no span for from an item it dropped', async () => { const spans = await new CodexAdapter().parse(rollout('skipped')) const root = spans.find((item) => item.parent_span_id === null)! - expect(JSON.parse(String(root.attributes['traces.codex.skipped_item_counts']))).toEqual({ + expect(JSON.parse(String(root.attributes['traces.codex.unmodeled_item_counts']))).toEqual({ FixtureFutureItem: 1 }) + expect(JSON.parse(String(root.attributes['traces.codex.dropped_item_counts']))).toEqual({ 'CommandExecution:malformed': 1, - FixtureFutureItem: 1, }) expect(spans.some((item) => item.attributes['traces.codex.item_id'] === 'item-broken')).toBe(false) }) @@ -232,7 +232,7 @@ describe('Codex human turns', () => { expect(humanTurns(spans)[0]!.start_time).toBe(at(2)) expect(humanTurns(spans)[0]!.attributes['traces.codex.user_message_event']).toBe(true) const root = spans.find((item) => item.parent_span_id === null)! - expect(root.attributes['traces.codex.skipped_item_counts']).toBeUndefined() + expect(root.attributes['traces.codex.dropped_item_counts']).toBeUndefined() }) it('pairs the two records of a turn whose message record carries a context prefix', async () => { @@ -247,6 +247,8 @@ describe('Codex human turns', () => { ])) expect(byName(spans, 'user.prompt')).toHaveLength(1) expect(humanTurns(spans).map((item) => [item.attributes.content, item.start_time])).toEqual([[prefixed, at(2)]]) + // Without the pairing, this span exists but carries no record of its own. + expect(humanTurns(spans)[0]!.attributes['traces.codex.user_message_event']).toBe(true) }) it('keeps text heuristics for rollouts that never recorded user_message events', async () => { diff --git a/tests/codex-exec.test.ts b/tests/codex-exec.test.ts index 9ef810b..1663605 100644 --- a/tests/codex-exec.test.ts +++ b/tests/codex-exec.test.ts @@ -168,6 +168,7 @@ describe('Codex exec JSONL adapter', () => { 'input.value': '{"cmd":"printf ok"}', 'output.value': 'ok', 'traces.codex.exec_exit_code': 0, + 'process.exit_code': 0, 'traces.codex.exec_lifecycle': 'paired', }) expect(file).toMatchObject({