diff --git a/plugins/tracing/src/parse.ts b/plugins/tracing/src/parse.ts index e4804f6..5c70645 100644 --- a/plugins/tracing/src/parse.ts +++ b/plugins/tracing/src/parse.ts @@ -103,6 +103,7 @@ function newTurn(startTime: number): MutableTurn { endTime: startTime, steps: [], subagentThreadIds: [], + subagentSpawnCallIds: {}, promptSkills: [], userImages: [], toolDefinitions: [], @@ -175,10 +176,11 @@ export function parseSession(lines: RolloutLine[]): { const ensureTurn = (ts: number): MutableTurn => (turn ??= newTurn(ts)); const ensureStep = (ts: number) => (step ??= newStep(ts)); - const recordSubagentThread = (threadId: string) => { + const recordSubagentThread = (threadId: string, spawnCallId?: string) => { if (!turn!.subagentThreadIds.includes(threadId)) { turn!.subagentThreadIds.push(threadId); } + if (spawnCallId) turn!.subagentSpawnCallIds[threadId] = spawnCallId; }; const closeStep = (ts: number, usage?: TokenUsage) => { @@ -323,7 +325,7 @@ export function parseSession(lines: RolloutLine[]): { spawned !== null && typeof spawned === "object" ? (spawned as { agent_id?: unknown }).agent_id : undefined; - if (typeof agentId === "string" && agentId) recordSubagentThread(agentId); + if (typeof agentId === "string" && agentId) recordSubagentThread(agentId, tc.callId); } } } else if (p.type === "tool_search_output") { @@ -375,6 +377,10 @@ export function parseSession(lines: RolloutLine[]): { } else if (et === "item_completed" && p.item?.type === "UserMessage") { const text = extractMessageText(p.item.content); if (text && !turn!.userInput) turn!.userInput = text; + } else if (et === "item_completed" && p.item?.type === "SubAgentActivity") { + if (p.item.kind === "started" && typeof p.item.agent_thread_id === "string") { + recordSubagentThread(p.item.agent_thread_id, p.item.id); + } } else if (et === "agent_message" && typeof p.message === "string") { turn!.lastAgentMessage = p.message; } else if (et === "token_count") { @@ -386,14 +392,14 @@ export function parseSession(lines: RolloutLine[]): { finishTurn(ts, { completed: true, aborted: true }); } else { if (et === "collab_agent_spawn_end" && typeof p.new_thread_id === "string") { - recordSubagentThread(p.new_thread_id); + recordSubagentThread(p.new_thread_id, p.call_id); } if ( et === "sub_agent_activity" && p.kind === "started" && typeof p.agent_thread_id === "string" ) { - recordSubagentThread(p.agent_thread_id); + recordSubagentThread(p.agent_thread_id, p.event_id); } if ( (et === "mcp_tool_call_begin" || et === "mcp_tool_call_end") && diff --git a/plugins/tracing/src/trace.ts b/plugins/tracing/src/trace.ts index ab5c43d..b67396a 100644 --- a/plugins/tracing/src/trace.ts +++ b/plugins/tracing/src/trace.ts @@ -486,6 +486,7 @@ async function emitTurn( try { const systemMessage = systemPromptText(turn.systemPrompt); const historyPrefix = ctx.historyPrefix ?? []; + const spawnObservations = new Map(); for (let i = 0; i < turn.steps.length; i++) { const step = turn.steps[i]; @@ -515,7 +516,8 @@ async function emitTurn( ); for (const tc of step.toolCalls) { - emitToolCall(tc, root, step.endTime); + const observation = emitToolCall(tc, root, step.endTime); + if (tc.name === "spawn_agent") spawnObservations.set(tc.callId, observation); } generation.end(new Date(generationEnd(step))); @@ -536,7 +538,7 @@ async function emitTurn( const seedBeforeChild = currentIdSeed(); await convertRollout(sub.file, { config: ctx.config, - parentObservation: root, + parentObservation: spawnObservations.get(turn.subagentSpawnCallIds[sub.threadId]) ?? root, subagentIndex: ctx.subagentIndex, seenThreadIds: ctx.seenThreadIds, ancestorTurnIds: ctx.inheritableTurnIds, @@ -559,7 +561,11 @@ async function emitTurn( if (failure && ctx.config.fail_on_error) throw failure; } -function emitToolCall(tc: ToolCall, parent: LangfuseObservation, fallbackEnd: number): void { +function emitToolCall( + tc: ToolCall, + parent: LangfuseObservation, + fallbackEnd: number, +): LangfuseObservation { const tool = startObservation( toolObservationName(tc), { @@ -576,6 +582,7 @@ function emitToolCall(tc: ToolCall, parent: LangfuseObservation, fallbackEnd: nu }, ); tool.end(new Date(tc.endTime ?? fallbackEnd)); + return tool; } function isFinal( diff --git a/plugins/tracing/src/types.ts b/plugins/tracing/src/types.ts index fbe10e8..298bcc8 100644 --- a/plugins/tracing/src/types.ts +++ b/plugins/tracing/src/types.ts @@ -127,8 +127,15 @@ export type EventMsgPayload = { last_token_usage?: TokenUsage; model_context_window?: number; } | null; - item?: { type?: string; content?: MessageContentPart[] } | null; + item?: { + type?: string; + id?: string; + kind?: string; + agent_thread_id?: string; + content?: MessageContentPart[]; + } | null; new_thread_id?: string | null; + event_id?: string; kind?: string; agent_thread_id?: string | null; invocation?: { server?: string; tool?: string; arguments?: unknown } | null; @@ -211,6 +218,7 @@ export type Turn = { finalOutput?: string; steps: ModelStep[]; subagentThreadIds: string[]; + subagentSpawnCallIds: Record; promptSkills: string[]; completed: boolean; aborted: boolean; diff --git a/plugins/tracing/test/parse.test.ts b/plugins/tracing/test/parse.test.ts index 66e11df..38b50ea 100644 --- a/plugins/tracing/test/parse.test.ts +++ b/plugins/tracing/test/parse.test.ts @@ -392,6 +392,38 @@ describe("parseSession", () => { const { turns } = parseSession(lines); expect(turns).toHaveLength(1); expect(turns[0].subagentThreadIds).toEqual(["thread-a"]); + expect(turns[0].subagentSpawnCallIds).toEqual({ "thread-a": "c1" }); + }); + + it("links item_completed subagent activity to its spawn call", () => { + const { turns } = parseSession([ + { timestamp: "2026-06-03T13:00:00.000Z", type: "session_meta", payload: { id: "s" } }, + { + timestamp: "2026-06-03T13:00:01.000Z", + type: "event_msg", + payload: { type: "task_started", turn_id: "t" }, + }, + { + timestamp: "2026-06-03T13:00:02.000Z", + type: "event_msg", + payload: { + type: "item_completed", + item: { + type: "SubAgentActivity", + id: "spawn-call", + kind: "started", + agent_thread_id: "thread-a", + }, + }, + }, + { + timestamp: "2026-06-03T13:00:03.000Z", + type: "event_msg", + payload: { type: "task_complete", turn_id: "t" }, + }, + ]); + expect(turns[0].subagentThreadIds).toEqual(["thread-a"]); + expect(turns[0].subagentSpawnCallIds).toEqual({ "thread-a": "spawn-call" }); }); it("records subagent threads from the spawn tool output when no spawn event is emitted", () => { @@ -441,6 +473,11 @@ describe("parseSession", () => { const { turns } = parseSession(lines); expect(turns).toHaveLength(1); expect(turns[0].subagentThreadIds).toEqual(["thread-a", "thread-b", "thread-c"]); + expect(turns[0].subagentSpawnCallIds).toEqual({ + "thread-a": "c1", + "thread-b": "c2", + "thread-c": "c3", + }); }); it("treats a trailing, never-completed turn as not completed", () => { diff --git a/plugins/tracing/test/skills.test.ts b/plugins/tracing/test/skills.test.ts index 9826f41..e4719e7 100644 --- a/plugins/tracing/test/skills.test.ts +++ b/plugins/tracing/test/skills.test.ts @@ -19,6 +19,7 @@ function turnWith(toolCalls: ToolCall[], promptSkills: string[] = []): Turn { endTime: 1, steps: [{ startTime: 0, endTime: 1, toolCalls }], subagentThreadIds: [], + subagentSpawnCallIds: {}, promptSkills, userImages: [], toolDefinitions: [], diff --git a/plugins/tracing/test/trace.test.ts b/plugins/tracing/test/trace.test.ts index aa28b51..36e015e 100644 --- a/plugins/tracing/test/trace.test.ts +++ b/plugins/tracing/test/trace.test.ts @@ -465,6 +465,9 @@ describe("convertRollout", () => { expect(childTurns).toHaveLength(1); const child = childTurns[0]; expect(child.spanContext().traceId).toBe(parent!.spanContext().traceId); + const spawn = spans.find((s) => s.name === "spawn_agent"); + expect(spawn).toBeDefined(); + expect(parentId(child)).toBe(spawn!.spanContext().spanId); expect(attr(child, "langfuse.observation.input")).toContain("hottest chilli"); const childGeneration = spans.find( @@ -1136,7 +1139,9 @@ describe("attached mode (external parent span)", () => { const subagentTurn = roots.find((s) => s.name === "Codex Subagent Turn")!; expect(parentId(parentTurn)).toBe(EXTERNAL_SPAN_ID); - expect(parentId(subagentTurn)).toBe(parentTurn.spanContext().spanId); + const spawn = exporter.getFinishedSpans().find((s) => s.name === "spawn_agent")!; + expect(parentId(spawn)).toBe(parentTurn.spanContext().spanId); + expect(parentId(subagentTurn)).toBe(spawn.spanContext().spanId); expect(subagentTurn.attributes["langfuse.internal.is_app_root"]).toBeUndefined(); });