diff --git a/README.md b/README.md index 3d367b2..a27a88f 100644 --- a/README.md +++ b/README.md @@ -9,9 +9,9 @@ Once enabled, every Codex turn shows up in Langfuse as a trace you can inspect, After each Codex turn, the plugin reads the session's rollout transcript and uploads it to Langfuse as a [trace](https://langfuse.com/docs/observability/data-model). The structure mirrors how Codex actually works: - **Turn** (`Codex Turn`, an [agent observation](https://langfuse.com/docs/observability/features/observation-types)) — one trace per turn, from your prompt to the final answer. -- **Generations** — one per model response within the turn, with the model name, reasoning, assistant text, the tool calls it requested, and token usage. -- **Tool calls** — `exec_command`, `apply_patch`, `spawn_agent`, MCP tools, web search, etc., each with its input, output, and error status. Failed commands are flagged as errors. -- **Subagents** — subagent threads are resolved from their own rollout files and nested under the spawning turn. +- **Generations** — one per model response within the turn, named `LLM` (or `LLM Subagent` inside subagent threads), with the model recorded on the observation plus reasoning, assistant text, the tool calls it requested, and token usage. +- **Tool calls** — shell commands, `apply_patch`, `spawn_agent`, MCP tools, web searches, etc., each with its input, output, and error status. Observation names include what was called — the shell command (`exec_command: git status`), the search query (`web_search: …`), or the MCP server and tool (`server.tool`) — and failed commands are flagged as errors. +- **Subagents** — subagent threads are resolved from their own rollout files and nested under the spawning turn as `Codex Subagent Turn`. - **Sessions** — all turns from one Codex session are grouped via the Codex thread id, so you can replay the whole session in Langfuse's [Sessions](https://langfuse.com/docs/observability/features/sessions) view. Interrupted turns (where you cancel mid-response) are still uploaded and flagged as interrupted. diff --git a/plugins/tracing/dist/index.mjs b/plugins/tracing/dist/index.mjs index 83fc783..b894005 100644 --- a/plugins/tracing/dist/index.mjs +++ b/plugins/tracing/dist/index.mjs @@ -46762,6 +46762,35 @@ function parseSession(lines) { }; s.toolCalls.push(tc); toolCallsById.set(tc.callId, tc); + } else if (p.type === "local_shell_call") { + const call = p; + const s = ensureStep(ts); + const tc = { + callId: call.call_id ?? call.id ?? `local_shell_${ts}_${s.toolCalls.length}`, + name: "local_shell", + args: call.action ?? void 0, + startTime: ts + }; + s.toolCalls.push(tc); + toolCallsById.set(tc.callId, tc); + } else if (p.type === "web_search_call") { + const call = p; + const callId = call.id ?? `web_search_${ts}`; + const existing = toolCallsById.get(callId); + if (existing) { + existing.args = existing.args ?? call.action ?? void 0; + existing.endTime = Math.max(existing.endTime ?? ts, ts); + } else { + const s = ensureStep(ts); + const tc = { + callId, + name: "web_search", + args: call.action ?? void 0, + startTime: ts + }; + s.toolCalls.push(tc); + toolCallsById.set(callId, tc); + } } else if (p.type === "function_call_output" || p.type === "custom_tool_call_output") { const out = p; const tc = toolCallsById.get(out.call_id); @@ -46807,6 +46836,28 @@ function parseSession(lines) { }); else { if (et === "collab_agent_spawn_end" && typeof p.new_thread_id === "string") turn.subagentThreadIds.push(p.new_thread_id); + if ((et === "mcp_tool_call_begin" || et === "mcp_tool_call_end") && typeof p.call_id === "string") { + const tc = toolCallsById.get(p.call_id); + const inv = p.invocation; + if (tc && typeof inv?.server === "string" && typeof inv?.tool === "string") tc.mcp = { + server: inv.server, + tool: inv.tool + }; + } + if (et === "web_search_end" && typeof p.call_id === "string") { + let tc = toolCallsById.get(p.call_id); + if (!tc) { + tc = { + callId: p.call_id, + name: "web_search", + args: void 0, + startTime: ts + }; + ensureStep(ts).toolCalls.push(tc); + toolCallsById.set(tc.callId, tc); + } + tc.args = tc.args ?? p.action ?? (typeof p.query === "string" ? { query: p.query } : void 0); + } if (typeof p.call_id === "string" && et.endsWith("_end")) { const tc = toolCallsById.get(p.call_id); if (tc) { @@ -46960,10 +47011,56 @@ function buildGenerationOutput(step, clip) { })); return Object.keys(output).length > 0 ? output : void 0; } +/** Tools that execute a shell command; the command makes a better span name. */ +const SHELL_TOOL_NAMES = new Set([ + "shell", + "exec_command", + "local_shell", + "container.exec" +]); +/** Collapse a value to a short single line usable inside a span name. */ +function previewText(value, max = 60) { + const oneLine = value.trim().replace(/\s+/g, " "); + return oneLine.length > max ? `${oneLine.slice(0, max - 1)}…` : oneLine; +} +function extractCommandText(args) { + if (args == null || typeof args !== "object") return void 0; + const cmd = args.command ?? args.cmd; + if (typeof cmd === "string" && cmd.trim()) return cmd; + if (Array.isArray(cmd) && cmd.length > 0 && cmd.every((c) => typeof c === "string")) { + const parts = cmd; + if (parts.length === 3 && /^(ba|z|fi)?sh$/.test(parts[0]) && /^-l?c$/.test(parts[1])) return parts[2]; + return parts.join(" "); + } +} +/** +* Observation name for a tool call: the tool name enriched with what it was +* called with (shell command, search query, MCP server/tool), so the trace +* tree shows *what* ran, not just which tool. +*/ +function toolObservationName(tc) { + if (tc.mcp) return `${tc.mcp.server}.${tc.mcp.tool}`; + const name = tc.name || "tool"; + if (SHELL_TOOL_NAMES.has(name)) { + const command = extractCommandText(tc.args); + if (command) return `${name}: ${previewText(command)}`; + } + if (name === "web_search" && tc.args != null && typeof tc.args === "object") { + const a = tc.args; + const detail = [ + a.query, + a.url, + a.pattern + ].find((v) => typeof v === "string" && v); + if (detail) return `${name}: ${previewText(detail)}`; + } + return name; +} /** Emit a single turn (and its subagents) as a Langfuse observation tree. */ async function emitTurn(turn, sessionMeta, ctx) { const clip = makeClip(ctx.config.max_chars); - const root = startObservation("Codex Turn", { + const isSubagent = sessionMeta.isSubagentThread === true || ctx.parentObservation != null; + const root = startObservation(isSubagent ? "Codex Subagent Turn" : "Codex Turn", { input: turn.userInput != null ? clip(turn.userInput) : void 0, output: turn.finalOutput != null ? clip(turn.finalOutput) : void 0, level: turn.aborted ? "WARNING" : void 0, @@ -46985,7 +47082,7 @@ async function emitTurn(turn, sessionMeta, ctx) { let previousToolResults = void 0; for (let i = 0; i < turn.steps.length; i++) { const step = turn.steps[i]; - const generation = startObservation(turn.model ?? "codex.generation", { + const generation = startObservation(isSubagent ? "LLM Subagent" : "LLM", { input: i === 0 ? turn.userInput != null ? clip(turn.userInput) : void 0 : previousToolResults, output: buildGenerationOutput(step, clip), model: turn.model, @@ -47018,12 +47115,15 @@ async function emitTurn(turn, sessionMeta, ctx) { root.end(new Date(turn.endTime)); } function emitToolCall(tc, parent, clip, fallbackEnd) { - startObservation(tc.name || "tool", { + startObservation(toolObservationName(tc), { input: tc.args, output: tc.output != null ? clip(toText(tc.output)) : void 0, level: tc.error ? "ERROR" : void 0, statusMessage: tc.error ? clip(tc.error) : void 0, - metadata: { "codex.call_id": tc.callId } + metadata: { + "codex.call_id": tc.callId, + "codex.tool_name": tc.name || "tool" + } }, { asType: "tool", startTime: new Date(tc.startTime), @@ -47055,7 +47155,7 @@ async function convertRollout(rolloutFile, options) { const seededParent = await seededTraceParent(options.config, sessionMeta, turnIndex + 1); await propagateAttributes({ sessionId: sessionMeta.sessionId, - traceName: "Codex Turn", + traceName: sessionMeta.isSubagentThread ? "Codex Subagent Turn" : "Codex Turn", ...options.config.user_id ? { userId: options.config.user_id } : {}, ...options.config.tags ? { tags: options.config.tags } : {}, ...options.config.metadata ? { metadata: options.config.metadata } : {} diff --git a/plugins/tracing/src/parse.ts b/plugins/tracing/src/parse.ts index 9dbdda7..877ad96 100644 --- a/plugins/tracing/src/parse.ts +++ b/plugins/tracing/src/parse.ts @@ -5,7 +5,9 @@ import type { ResponseItemFunctionCall, ResponseItemFunctionCallOutput, ResponseItemCustomToolCall, + ResponseItemLocalShellCall, ResponseItemMessage, + ResponseItemWebSearchCall, RolloutLine, SessionMeta, TokenUsage, @@ -224,6 +226,41 @@ export function parseSession(lines: RolloutLine[]): { }; s.toolCalls.push(tc); toolCallsById.set(tc.callId, tc); + } else if (p.type === "local_shell_call") { + // Built-in local shell tool: the command lives in `action`, and + // exec_command_end / function_call_output enrich it like any function + // call. + const call = p as unknown as ResponseItemLocalShellCall; + const s = ensureStep(ts); + const tc: ToolCall = { + callId: call.call_id ?? call.id ?? `local_shell_${ts}_${s.toolCalls.length}`, + name: "local_shell", + args: call.action ?? undefined, + startTime: ts, + }; + s.toolCalls.push(tc); + toolCallsById.set(tc.callId, tc); + } else if (p.type === "web_search_call") { + // Server-side web search: there is no output item, and the + // web_search_end event may be recorded before or after this item, so + // merge with an existing call when one was already registered. + const call = p as unknown as ResponseItemWebSearchCall; + const callId = call.id ?? `web_search_${ts}`; + const existing = toolCallsById.get(callId); + if (existing) { + existing.args = existing.args ?? call.action ?? undefined; + existing.endTime = Math.max(existing.endTime ?? ts, ts); + } else { + const s = ensureStep(ts); + const tc: ToolCall = { + callId, + name: "web_search", + args: call.action ?? undefined, + startTime: ts, + }; + s.toolCalls.push(tc); + toolCallsById.set(callId, tc); + } } else if (p.type === "function_call_output" || p.type === "custom_tool_call_output") { const out = p as unknown as ResponseItemFunctionCallOutput; const tc = toolCallsById.get(out.call_id); @@ -273,6 +310,32 @@ export function parseSession(lines: RolloutLine[]): { if (et === "collab_agent_spawn_end" && typeof p.new_thread_id === "string") { turn!.subagentThreadIds.push(p.new_thread_id); } + // MCP tool calls are function calls with a mangled name + // (`server__tool`); the begin/end events carry the clean server/tool + // split, which makes a much better observation name. + if ( + (et === "mcp_tool_call_begin" || et === "mcp_tool_call_end") && + typeof p.call_id === "string" + ) { + const tc = toolCallsById.get(p.call_id); + const inv = p.invocation; + if (tc && typeof inv?.server === "string" && typeof inv?.tool === "string") { + tc.mcp = { server: inv.server, tool: inv.tool }; + } + } + // Web searches run server-side inside the model response. The end + // event can be recorded before the web_search_call response item, so + // register the call here if it is not known yet. + if (et === "web_search_end" && typeof p.call_id === "string") { + let tc = toolCallsById.get(p.call_id); + if (!tc) { + tc = { callId: p.call_id, name: "web_search", args: undefined, startTime: ts }; + ensureStep(ts).toolCalls.push(tc); + toolCallsById.set(tc.callId, tc); + } + tc.args = + tc.args ?? p.action ?? (typeof p.query === "string" ? { query: p.query } : undefined); + } // Tool execution lifecycle events (exec_command_end, patch_apply_end, // mcp_tool_call_end, collab_*_end, …) match a call by id and add // timing, status, and output. diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index 89e04cd..6f965aa 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -155,6 +155,51 @@ function buildGenerationOutput(step: ModelStep, clip: Clip): Record 0 ? output : undefined; } +/** Tools that execute a shell command; the command makes a better span name. */ +const SHELL_TOOL_NAMES = new Set(["shell", "exec_command", "local_shell", "container.exec"]); + +/** Collapse a value to a short single line usable inside a span name. */ +function previewText(value: string, max = 60): string { + const oneLine = value.trim().replace(/\s+/g, " "); + return oneLine.length > max ? `${oneLine.slice(0, max - 1)}…` : oneLine; +} + +function extractCommandText(args: unknown): string | undefined { + if (args == null || typeof args !== "object") return undefined; + const cmd = + (args as { command?: unknown; cmd?: unknown }).command ?? (args as { cmd?: unknown }).cmd; + if (typeof cmd === "string" && cmd.trim()) return cmd; + if (Array.isArray(cmd) && cmd.length > 0 && cmd.every((c) => typeof c === "string")) { + const parts = cmd as string[]; + // Codex wraps most commands as ["bash", "-lc", "