From d9904a5c308908efc163e3000349bcfae001438d Mon Sep 17 00:00:00 2001 From: Ben Bachem <10088265+bezbac@users.noreply.github.com> Date: Fri, 9 Oct 2026 18:15:47 +0200 Subject: [PATCH] Emit structured tool observation outputs --- src/langfuse.ts | 284 +++++++++-- test/integration/v1.test.ts | 761 ++++++++++++++++++++++++++-- test/integration/v2-exports.test.ts | 510 +++++++++++++++++++ 3 files changed, 1473 insertions(+), 82 deletions(-) create mode 100644 test/integration/v2-exports.test.ts diff --git a/src/langfuse.ts b/src/langfuse.ts index b909fd2..80b0c55 100644 --- a/src/langfuse.ts +++ b/src/langfuse.ts @@ -18,6 +18,10 @@ export class LangfuseClient { readonly baseUrl: string; readonly forceFlush: Effect.Effect; private readonly traceState: LangfuseTraceState; + private readonly assistantParents = new Map< + string, + { sessionID: string; parentID: string } + >(); constructor(input: { baseUrl: string; @@ -30,6 +34,7 @@ export class LangfuseClient { } clearTraceState() { + this.assistantParents.clear(); this.traceState.assistantParts.clear(); this.traceState.abortedSessions.clear(); this.traceState.tracedEventIds.clear(); @@ -40,7 +45,6 @@ export class LangfuseClient { this.traceState.generationParentSpans.clear(); this.traceState.generationInputsBySession.clear(); this.traceState.generationInputSnapshotsBySession.clear(); - this.traceState.toolResultSourceMessageIdsBySession.clear(); this.traceState.turnObservationsByMessageId.clear(); this.traceState.latestTurnObservationsBySession.clear(); this.traceState.finalizedToolCallIds.clear(); @@ -50,6 +54,11 @@ export class LangfuseClient { } clearSessionTraceState(sessionID: string) { + for (const [messageID, parent] of this.assistantParents) { + if (parent.sessionID === sessionID) { + this.assistantParents.delete(messageID); + } + } const sessionMessageIds = new Set(); for (const [messageID, parts] of this.traceState.assistantParts) { @@ -100,7 +109,6 @@ export class LangfuseClient { this.traceState.generationParentSpans.delete(sessionID); this.traceState.generationInputsBySession.delete(sessionID); this.traceState.generationInputSnapshotsBySession.delete(sessionID); - this.traceState.toolResultSourceMessageIdsBySession.delete(sessionID); this.traceState.latestTurnObservationsBySession.delete(sessionID); this.traceState.sessionHistories.delete(sessionID); this.traceState.pendingUserMessageIdsBySession.delete(sessionID); @@ -707,6 +715,10 @@ export class LangfuseClient { } this.traceState.tracedGenerationIds.add(input.messageID); + this.assistantParents.set(input.messageID, { + sessionID: input.sessionID, + parentID: input.parentID, + }); const output = input.output ?? this.getAssistantMessage(input.messageID); const turn = this.getTurnObservation(input.sessionID, input.parentID); @@ -1041,7 +1053,12 @@ export class LangfuseClient { span.setAttribute( "langfuse.observation.output", - JSON.stringify({ title: input.title, output: input.output }), + JSON.stringify({ + role: "tool", + name: input.tool, + tool_call_id: input.callID, + content: input.output, + } satisfies ChatMlMessage), ); span.setAttribute( "langfuse.observation.metadata", @@ -1063,7 +1080,6 @@ export class LangfuseClient { }); this.traceState.activeToolObservations.delete(input.callID); this.traceState.finalizedToolCallIds.add(input.callID); - this.traceState.toolMessageIdsByCallId.delete(input.callID); } traceToolError(input: { @@ -1106,7 +1122,12 @@ export class LangfuseClient { span.setAttribute( "langfuse.observation.output", - JSON.stringify({ error: input.error }), + JSON.stringify({ + role: "tool", + name: input.tool ?? observation.tool, + tool_call_id: input.callID, + content: input.error, + } satisfies ChatMlMessage), ); span.setStatus({ code: SpanStatusCode.ERROR, @@ -1123,7 +1144,6 @@ export class LangfuseClient { }); this.traceState.activeToolObservations.delete(input.callID); this.traceState.finalizedToolCallIds.add(input.callID); - this.traceState.toolMessageIdsByCallId.delete(input.callID); } private ensureGenerationParent(sessionID: string) { @@ -1238,7 +1258,7 @@ export class LangfuseClient { } const startIndex = - history.startIndexByAssistantMessageId.get(assistantMessageID); + history.assistantMessages.get(assistantMessageID)?.startIndex; return startIndex == null ? history.messages @@ -1266,31 +1286,181 @@ export class LangfuseClient { const snapshot = this.traceState.generationInputSnapshotsBySession.get(sessionID); this.traceState.generationInputSnapshotsBySession.delete(sessionID); + const pending = this.traceState.generationInputsBySession.get(sessionID); + this.traceState.generationInputsBySession.delete(sessionID); + const sourceMessageIds = new Set(); + for (const message of pending ?? []) { + if (message.role !== "tool") { + continue; + } + const messageID = this.traceState.toolMessageIdsByCallId.get( + message.tool_call_id, + ); + if (messageID != null) { + sourceMessageIds.add(messageID); + } + this.traceState.toolMessageIdsByCallId.delete(message.tool_call_id); + } if (snapshot !== undefined) { - this.traceState.generationInputsBySession.delete(sessionID); - this.traceState.toolResultSourceMessageIdsBySession.delete(sessionID); return snapshot; } - const pending = this.traceState.generationInputsBySession.get(sessionID); - const sourceMessageID = - this.traceState.toolResultSourceMessageIdsBySession.get(sessionID); const pendingUserMessageID = this.traceState.pendingUserMessageIdsBySession.get(sessionID); - this.traceState.generationInputsBySession.delete(sessionID); - this.traceState.toolResultSourceMessageIdsBySession.delete(sessionID); - const assistantOfToolResults = - sourceMessageID == null - ? [] - : (this.getAssistantMessage(sourceMessageID) ?? []); - const prefix = this.getHistoryPrefix(sessionID, assistantMessageID); + const unplacedToolCallIds = new Set(); + const prefix = (() => { + const storedPrefix = this.getHistoryPrefix(sessionID, assistantMessageID); + if (sourceMessageIds.size === 0) { + return storedPrefix; + } + + const history = this.traceState.sessionHistories.get(sessionID); + const rebuiltMessages = new Map(); + // Live message order is independent of the order tools finish in. + const orderedSourceMessageIds = new Set([ + ...Array.from(this.traceState.assistantParts.keys()).filter( + (messageID) => sourceMessageIds.has(messageID), + ), + ...sourceMessageIds, + ]); + for (const sourceMessageID of orderedSourceMessageIds) { + const source = history?.assistantMessages.get(sourceMessageID); + const liveParts = Array.from( + this.traceState.assistantParts.get(sourceMessageID)?.values() ?? [], + ); + const parts = [...(source?.parts ?? [])]; + let previousIndex: number | undefined; + + // Tool hooks and message events use different part IDs for the same call. + // Step markers anchor live deltas without collapsing sequential calls + // into one assistant message or losing snapshot-only parts after a restart. + const partKey = (part: MessagePart) => + part.type === "tool" ? `tool:${part.callID}` : part.id; + for (const [liveIndex, part] of liveParts.entries()) { + const storedIndex = parts.findIndex( + (stored) => partKey(stored) === partKey(part), + ); + if (storedIndex >= 0) { + const stored = parts[storedIndex]; + previousIndex = storedIndex; + if ( + stored.type === "tool" && + part.type === "tool" && + (stored.state.status === "completed" || + stored.state.status === "error") && + (part.state.status === "pending" || + part.state.status === "running") + ) { + continue; + } + if ( + (stored.type === "text" || stored.type === "reasoning") && + (part.type === "text" || part.type === "reasoning") && + stored.text.startsWith(part.text) + ) { + continue; + } + parts[storedIndex] = part; + continue; + } + + const insertionIndex = (() => { + if (previousIndex !== undefined) { + if (part.type === "step-start") { + // Snapshot-only parts still belong to the preceding step. + const searchStartIndex = previousIndex + 1; + const nextStepIndex = parts.findIndex( + (stored, index) => + index >= searchStartIndex && stored.type === "step-start", + ); + return nextStepIndex < 0 ? parts.length : nextStepIndex; + } + return previousIndex + 1; + } + const nextKeys = new Set( + liveParts.slice(liveIndex + 1).map(partKey), + ); + const nextStoredIndex = parts.findIndex((stored) => + nextKeys.has(partKey(stored)), + ); + if (nextStoredIndex >= 0) { + return nextStoredIndex; + } + return parts.length; + })(); + parts.splice(insertionIndex, 0, part); + previousIndex = insertionIndex; + } + + const rebuilt = buildSessionHistory([ + { info: { id: sourceMessageID, role: "assistant" }, parts }, + ]).messages; + rebuiltMessages.set(sourceMessageID, rebuilt); + } + + const replacements: (Pick< + NonNullable>, + "startIndex" | "endIndex" + > & { messages: ChatMlMessage[] })[] = []; + for (const [messageID, source] of history?.assistantMessages ?? []) { + const rebuilt = rebuiltMessages.get(messageID); + if (!rebuilt || source.endIndex > storedPrefix.length) { + continue; + } + replacements.push({ ...source, messages: rebuilt }); + rebuiltMessages.delete(messageID); + } + for (const [messageID, rebuilt] of rebuiltMessages) { + const parentID = this.assistantParents.get(messageID)?.parentID; + const parentIndex = + parentID == null ? undefined : history?.userMessages.get(parentID); + const userCount = storedPrefix.filter( + (message) => message.role === "user", + ).length; + if (parentIndex === undefined && userCount > 1) { + // Without a parent anchor, assigning this group to a turn would be a guess. + for (const message of rebuilt) { + if (message.role === "assistant") { + for (const call of message.tool_calls ?? []) { + unplacedToolCallIds.add(call.id); + } + } + } + continue; + } + const nextUserIndex = + parentIndex === undefined + ? -1 + : storedPrefix.findIndex( + (message, index) => + index > parentIndex && message.role === "user", + ); + const insertionIndex = + nextUserIndex < 0 ? storedPrefix.length : nextUserIndex; + replacements.push({ + startIndex: insertionIndex, + endIndex: insertionIndex, + messages: rebuilt, + }); + } + const merged: ChatMlMessage[] = []; + let cursor = 0; + for (const replacement of replacements.sort( + (a, b) => a.startIndex - b.startIndex || a.endIndex - b.endIndex, + )) { + merged.push( + ...storedPrefix.slice(cursor, replacement.startIndex), + ...replacement.messages, + ); + cursor = replacement.endIndex; + } + merged.push(...storedPrefix.slice(cursor)); + return merged; + })(); - if (prefix.length === 0) { - // No snapshot from OpenCode yet, so the live delta is all there is. - return sourceMessageID == null - ? pending - : [...assistantOfToolResults, ...(pending ?? [])]; + if (prefix.length === 0 && pending === undefined) { + return undefined; } // Taking the user message from both sources would list it twice, so it @@ -1303,15 +1473,39 @@ export class LangfuseClient { ) ? [] : (pending ?? []).filter((message) => message.role === "user"); - const toolResults = (pending ?? []).filter( - (message) => message.role === "tool", + const existingToolResultIds = new Set( + prefix + .filter((message) => message.role === "tool") + .map((message) => message.tool_call_id), ); - const combined = [ - ...prefix, - ...assistantOfToolResults, - ...toolResults, - ...pendingUserMessages, - ]; + const combined = [...prefix]; + for (const message of pending ?? []) { + if ( + message.role !== "tool" || + unplacedToolCallIds.has(message.tool_call_id) || + existingToolResultIds.has(message.tool_call_id) + ) { + continue; + } + existingToolResultIds.add(message.tool_call_id); + const assistantIndex = combined.findIndex( + (assistant) => + assistant.role === "assistant" && + assistant.tool_calls?.some( + (call) => call.id === message.tool_call_id, + ) === true, + ); + if (assistantIndex < 0) { + combined.push(message); + continue; + } + let resultIndex = assistantIndex + 1; + while (combined[resultIndex]?.role === "tool") { + resultIndex++; + } + combined.splice(resultIndex, 0, message); + } + combined.push(...pendingUserMessages); if (pendingUserMessages.length > 0) { return combined; @@ -1350,10 +1544,7 @@ export class LangfuseClient { this.traceState.generationInputsBySession.set(input.sessionID, toolResults); if (messageID != null) { - this.traceState.toolResultSourceMessageIdsBySession.set( - input.sessionID, - messageID, - ); + this.traceState.toolMessageIdsByCallId.set(input.callID, messageID); } } @@ -1395,7 +1586,6 @@ export type LangfuseTraceState = { generationParentSpans: Map; generationInputsBySession: Map; generationInputSnapshotsBySession: Map; - toolResultSourceMessageIdsBySession: Map; sessionParentIds: Map; sessionHistories: Map; pendingUserMessageIdsBySession: Map; @@ -1408,7 +1598,11 @@ export type MessagePart = Extract< export type SessionHistory = { messages: ChatMlMessage[]; - startIndexByAssistantMessageId: Map; + userMessages: Map; + assistantMessages: Map< + string, + { startIndex: number; endIndex: number; parts: MessagePart[] } + >; // Which messages the snapshot actually holds. A stale snapshot - the last // refresh failed, or it ran before the message existed - does not contain // the request that triggered the generation, and the live buffer has to @@ -1471,18 +1665,20 @@ export function buildSessionHistory( }[], ): SessionHistory { const history: ChatMlMessage[] = []; - const startIndexByAssistantMessageId = new Map(); + const assistantMessages: SessionHistory["assistantMessages"] = new Map(); + const userMessages: SessionHistory["userMessages"] = new Map(); const messageIds = new Set(); for (const message of messages) { messageIds.add(message.info.id); if (message.info.role === "user") { + userMessages.set(message.info.id, history.length); history.push(formatUserMessage(message.parts)); continue; } - startIndexByAssistantMessageId.set(message.info.id, history.length); + const startIndex = history.length; for (const step of splitAssistantSteps(message.parts)) { const assistant = buildAssistantMessage(step); @@ -1493,9 +1689,14 @@ export function buildSessionHistory( history.push(...toolResultsOfStep(step)); } + assistantMessages.set(message.info.id, { + startIndex, + endIndex: history.length, + parts: message.parts, + }); } - return { messages: history, startIndexByAssistantMessageId, messageIds }; + return { messages: history, assistantMessages, userMessages, messageIds }; } function withToolDefinitions( @@ -1756,7 +1957,6 @@ export const createLangfuseClient = (input: { generationParentSpans: new Map(), generationInputsBySession: new Map(), generationInputSnapshotsBySession: new Map(), - toolResultSourceMessageIdsBySession: new Map(), sessionParentIds: new Map(), sessionHistories: new Map(), pendingUserMessageIdsBySession: new Map(), diff --git a/test/integration/v1.test.ts b/test/integration/v1.test.ts index d5010d8..5749c50 100644 --- a/test/integration/v1.test.ts +++ b/test/integration/v1.test.ts @@ -8,7 +8,6 @@ import { Schema } from "effect"; import { afterAll, afterEach, - beforeAll, beforeEach, describe, expect, @@ -147,6 +146,7 @@ let plugin: Plugin; let collectorStatus = 200; let hooksDisposed = false; let collectorBaseUrl: string; +let collectorPort: number | undefined; let toolListCalls = 0; let toolListShouldFail = false; @@ -519,7 +519,15 @@ const disposeHooks = async () => { } }; -beforeAll(async () => { +beforeEach(async () => { + requests.length = 0; + sessionStore.clear(); + collectorErrors.length = 0; + collectorStatus = 200; + hooksDisposed = false; + toolListCalls = 0; + toolListShouldFail = false; + sessionMessagesShouldFail = false; server = createServer((request, response) => { const chunks: Buffer[] = []; @@ -553,7 +561,9 @@ beforeAll(async () => { await new Promise((resolve, reject) => { server.once("error", reject); - server.listen(0, "127.0.0.1", () => { + // The process-wide tracer retains its endpoint across plugin instances. + // Rebind the same port, but give each test a new collector and connections. + server.listen(collectorPort ?? 0, "127.0.0.1", () => { resolve(); }); }); @@ -562,6 +572,7 @@ beforeAll(async () => { if (address === null || typeof address === "string") { throw new Error("Expected the test collector to listen on a TCP port"); } + collectorPort = address.port; process.env.LANGFUSE_PUBLIC_KEY = "pk-test"; process.env.LANGFUSE_SECRET_KEY = "sk-test"; @@ -574,26 +585,13 @@ beforeAll(async () => { const builtPluginUrl = new URL("../../dist/v1/index.js", import.meta.url); const builtPlugin: unknown = await import(builtPluginUrl.href); plugin = Schema.decodeUnknownSync(PluginModuleSchema)(builtPlugin).default; -}); - -beforeEach(async () => { - requests.length = 0; - sessionStore.clear(); - collectorErrors.length = 0; - collectorStatus = 200; - hooksDisposed = false; - toolListCalls = 0; - toolListShouldFail = false; - sessionMessagesShouldFail = false; hooks = await createHooks(collectorBaseUrl); }); afterEach(async () => { - await disposeHooks(); -}); - -afterAll(async () => { try { + await disposeHooks(); + } finally { await new Promise((resolve, reject) => { server.close((error) => { if (error) { @@ -603,20 +601,22 @@ afterAll(async () => { } }); }); - } finally { - for (const [name, value] of Object.entries({ - LANGFUSE_PUBLIC_KEY: originalEnvironment.publicKey, - LANGFUSE_SECRET_KEY: originalEnvironment.secretKey, - LANGFUSE_BASE_URL: originalEnvironment.baseUrl, - LANGFUSE_BASEURL: originalEnvironment.legacyBaseUrl, - LANGFUSE_ENVIRONMENT: originalEnvironment.environment, - LANGFUSE_USER_ID: originalEnvironment.userId, - })) { - if (value === undefined) { - Reflect.deleteProperty(process.env, name); - } else { - process.env[name] = value; - } + } +}); + +afterAll(() => { + for (const [name, value] of Object.entries({ + LANGFUSE_PUBLIC_KEY: originalEnvironment.publicKey, + LANGFUSE_SECRET_KEY: originalEnvironment.secretKey, + LANGFUSE_BASE_URL: originalEnvironment.baseUrl, + LANGFUSE_BASEURL: originalEnvironment.legacyBaseUrl, + LANGFUSE_ENVIRONMENT: originalEnvironment.environment, + LANGFUSE_USER_ID: originalEnvironment.userId, + })) { + if (value === undefined) { + Reflect.deleteProperty(process.env, name); + } else { + process.env[name] = value; } } }); @@ -1052,13 +1052,17 @@ describe("built plugin", { concurrent: false }, () => { path: "README.md", }); expect(getJsonAttribute(tool, "langfuse.observation.output")).toEqual({ - title: "README.md", - output: "# Project", + role: "tool", + name: "read", + tool_call_id: "nested-observations-call", + content: "# Project", }); const mcpTool = getSpan(spans, "mcp_test_tool"); expect(getJsonAttribute(mcpTool, "langfuse.observation.output")).toEqual({ - title: "mcp_test_tool", - output: "MCP tool result\n\nMCP resource contents", + role: "tool", + name: "mcp_test_tool", + tool_call_id: mcpCallID, + content: "MCP tool result\n\nMCP resource contents", }); const failedMcpTool = getSpan(spans, "mcp_failing_tool"); expect(failedMcpTool.status).toEqual({ @@ -1067,7 +1071,12 @@ describe("built plugin", { concurrent: false }, () => { }); expect( getJsonAttribute(failedMcpTool, "langfuse.observation.output"), - ).toEqual({ error: "MCP tool failed" }); + ).toEqual({ + role: "tool", + name: "mcp_failing_tool", + tool_call_id: failedMcpCallID, + content: "MCP tool failed", + }); expect(tool.traceId).toBe(generation.traceId); expect(tool.parentSpanId).toBe(generation.spanId); @@ -1080,7 +1089,12 @@ describe("built plugin", { concurrent: false }, () => { message: "Tool execution timed out after 5 seconds", }); expect(getJsonAttribute(failedTool, "langfuse.observation.output")).toEqual( - { error: "Tool execution timed out after 5 seconds" }, + { + role: "tool", + name: "webfetch", + tool_call_id: "timed-out-webfetch", + content: "Tool execution timed out after 5 seconds", + }, ); const retry = getSpan(spans, "opencode.generation.retry"); @@ -1511,6 +1525,671 @@ describe("built plugin", { concurrent: false }, () => { ]); }); + test.each([ + { snapshot: "complete", storedCallCount: 3 }, + { snapshot: "partial", storedCallCount: 1 }, + { snapshot: "text-only", storedCallCount: 0 }, + ])( + "matches parallel same-name tool responses without duplicating $snapshot history", + async ({ storedCallCount }) => { + const sessionID = "parallel-tool-session"; + const userMessageID = "parallel-tool-user"; + const assistantMessageID = "parallel-tool-assistant"; + const calls = [ + { + callID: "call-first", + path: "first.txt", + content: "First\nresponse", + failed: false, + }, + { + callID: "call-second", + path: "second.txt", + content: "Second response", + failed: false, + }, + { + callID: "call-missing", + path: "missing.txt", + content: "ENOENT: missing.txt", + failed: true, + }, + ]; + + await sendUserMessage({ + sessionID, + messageID: userMessageID, + text: "Read three files", + started: startedAt, + }); + await startAssistantMessage({ + sessionID, + userMessageID, + assistantMessageID, + started: startedAt + 100, + }); + await startGeneration({ + id: "parallel-tool-step", + sessionID, + assistantMessageID, + started: startedAt + 100, + }); + + const textPart = { + id: "parallel-tool-text", + sessionID, + messageID: assistantMessageID, + type: "text" as const, + text: "Reading files", + }; + await emitEvent({ + type: "message.part.updated", + properties: { part: textPart }, + }); + + const parts = calls.map((call) => ({ + id: `${call.callID}-part`, + sessionID, + messageID: assistantMessageID, + type: "tool" as const, + callID: call.callID, + tool: "read", + state: call.failed + ? { + status: "error" as const, + input: { path: call.path }, + error: call.content, + time: { start: startedAt + 200, end: startedAt + 300 }, + } + : { + status: "completed" as const, + input: { path: call.path }, + title: call.path, + output: call.content, + metadata: {}, + time: { start: startedAt + 200, end: startedAt + 300 }, + }, + })); + for (const part of parts) { + await emitEvent({ + type: "message.part.updated", + properties: { + part: { + ...part, + state: { + status: "running", + input: part.state.input, + time: { start: startedAt + 200 }, + }, + }, + }, + }); + } + // Complete in reverse order so names and arrival order cannot match results. + for (const part of [...parts].reverse()) { + await emitEvent({ type: "message.part.updated", properties: { part } }); + await emitEvent({ type: "message.part.updated", properties: { part } }); + } + await completeGeneration({ + sessionID, + userMessageID, + assistantMessageID, + started: startedAt + 100, + completed: startedAt + 400, + }); + storeMessage(sessionID, { + info: { id: assistantMessageID, role: "assistant", sessionID }, + parts: [textPart, ...parts.slice(0, storedCallCount)], + }); + + await startGeneration({ + id: "parallel-tool-followup-step", + sessionID, + assistantMessageID: "parallel-tool-followup", + started: startedAt + 500, + }); + await completeGeneration({ + sessionID, + userMessageID, + assistantMessageID: "parallel-tool-followup", + started: startedAt + 500, + completed: startedAt + 700, + text: "Done", + }); + + const { spans } = await flushSession(sessionID); + const toolSpans = spans.filter((span) => span.name === "read"); + expect(toolSpans).toHaveLength(calls.length); + const generations = spans.filter( + (span) => span.name === "opencode.generation", + ); + const generation = generations.find( + (span) => span.spanId === toolSpans[0].parentSpanId, + ); + const followup = generations.find((span) => span !== generation); + if (!generation || !followup) { + throw new Error("Expected both generations"); + } + const toolCalls = calls.map((call) => ({ + id: call.callID, + name: "read", + arguments: JSON.stringify({ path: call.path }), + })); + expect( + getJsonAttribute(generation, "langfuse.observation.output"), + ).toEqual([ + { role: "assistant", content: "Reading files", tool_calls: toolCalls }, + ]); + const results = calls.map((call) => ({ + role: "tool", + name: "read", + tool_call_id: call.callID, + content: call.content, + })); + const outputs = toolSpans.map((span) => + getJsonAttribute(span, "langfuse.observation.output"), + ); + expect(outputs).toEqual(expect.arrayContaining(results)); + for (const [index, call] of calls.entries()) { + const span = toolSpans.find((candidate) => { + const output = getJsonAttribute( + candidate, + "langfuse.observation.output", + ); + return ( + typeof output === "object" && + output !== null && + "tool_call_id" in output && + output.tool_call_id === call.callID + ); + }); + if (!span) { + throw new Error(`Expected tool response for ${call.callID}`); + } + expect(getJsonAttribute(span, "langfuse.observation.output")).toEqual( + results[index], + ); + expect(span.parentSpanId).toBe(generation.spanId); + expect(getJsonAttribute(span, "langfuse.observation.metadata")).toEqual( + { + callID: call.callID, + tool: "read", + }, + ); + if (call.failed) { + expect(span.status).toEqual({ code: 2, message: call.content }); + } + } + expect(getJsonAttribute(followup, "langfuse.observation.input")).toEqual([ + { role: "user", content: [{ type: "text", text: "Read three files" }] }, + { role: "assistant", content: "Reading files", tool_calls: toolCalls }, + ...results, + ]); + }, + ); + + test.each(["known", "unknown"] as const)( + "places a missing tool source only with a %s parent turn", + async (parent) => { + const sessionID = "missing-source-turn-session"; + const userMessageID = "missing-source-first-user"; + const assistantMessageID = "missing-source-assistant"; + const callID = "missing-source-call"; + await sendUserMessage({ + sessionID, + messageID: userMessageID, + text: "Read a file", + started: startedAt, + }); + await startAssistantMessage({ + sessionID, + userMessageID, + assistantMessageID, + started: startedAt + 100, + }); + await startGeneration({ + id: "missing-source-step", + sessionID, + assistantMessageID, + started: startedAt + 100, + }); + await emitEvent({ + type: "message.part.updated", + properties: { + part: { + id: "missing-source-tool", + sessionID, + messageID: assistantMessageID, + type: "tool", + callID, + tool: "read", + state: { + status: "running", + input: {}, + time: { start: startedAt + 100 }, + }, + }, + }, + }); + await completeGeneration({ + sessionID, + userMessageID: parent === "known" ? userMessageID : "unknown-parent", + assistantMessageID, + started: startedAt + 100, + completed: startedAt + 200, + text: "Reading", + }); + sessionStore.set( + sessionID, + (sessionStore.get(sessionID) ?? []).filter( + (message) => message.info.id !== assistantMessageID, + ), + ); + const nextUserMessageID = "missing-source-next-user"; + await sendUserMessage({ + sessionID, + messageID: nextUserMessageID, + text: "Now summarize", + started: startedAt + 300, + }); + // The old tool finishes after the new user message reaches the store. + await hooks["tool.execute.after"]?.( + { sessionID, callID, tool: "read", args: {} }, + { title: "read", output: "File contents", metadata: {} }, + ); + await startGeneration({ + id: "missing-source-followup-step", + sessionID, + assistantMessageID: "missing-source-followup", + started: startedAt + 400, + }); + await completeGeneration({ + sessionID, + userMessageID: nextUserMessageID, + assistantMessageID: "missing-source-followup", + started: startedAt + 400, + completed: startedAt + 500, + text: "Done", + }); + const { spans } = await flushSession(sessionID); + const followup = spans + .filter((span) => span.name === "opencode.generation") + .sort( + (a, b) => Number(a.startTimeUnixNano) - Number(b.startTimeUnixNano), + ) + .at(-1); + if (!followup) { + throw new Error("Expected followup generation"); + } + expect(getJsonAttribute(followup, "langfuse.observation.input")).toEqual([ + { role: "user", content: [{ type: "text", text: "Read a file" }] }, + ...(parent === "known" + ? [ + { + role: "assistant", + content: "Reading", + tool_calls: [{ id: callID, name: "read", arguments: "{}" }], + }, + { + role: "tool", + name: "read", + tool_call_id: callID, + content: "File contents", + }, + ] + : []), + { + role: "user", + content: [{ type: "text", text: "Now summarize" }], + tools: expectedToolDefinitions, + }, + ]); + }, + ); + + test.each(["stale", "partial"] as const)( + "rebuilds every tool source with reverse completions and %s history", + async (history) => { + const sessionID = "multiple-tool-sources-session"; + const userMessageID = "multiple-tool-sources-user"; + const calls = [1, 2].map((index) => ({ + assistantMessageID: `multiple-tool-sources-assistant-${index.toString()}`, + callID: `multiple-tool-sources-call-${index.toString()}`, + output: `response ${index.toString()}`, + })); + await sendUserMessage({ + sessionID, + messageID: userMessageID, + text: "Read two files", + started: startedAt, + }); + for (const [index, call] of calls.entries()) { + await startAssistantMessage({ + sessionID, + userMessageID, + assistantMessageID: call.assistantMessageID, + started: startedAt + index * 100, + }); + await startGeneration({ + id: `${call.callID}-step`, + sessionID, + assistantMessageID: call.assistantMessageID, + started: startedAt + index * 100, + }); + if (history === "stale" || index === calls.length - 1) { + sessionMessagesShouldFail = true; + } + await emitEvent({ + type: "message.part.updated", + properties: { + part: { + id: `${call.callID}-part`, + sessionID, + messageID: call.assistantMessageID, + type: "tool", + callID: call.callID, + tool: "read", + state: { + status: "running", + input: {}, + time: { start: startedAt + index * 100 }, + }, + }, + }, + }); + await completeGeneration({ + sessionID, + userMessageID, + assistantMessageID: call.assistantMessageID, + started: startedAt + index * 100, + completed: startedAt + index * 100 + 50, + text: `Reading ${call.callID}`, + }); + } + for (const call of calls.toReversed()) { + await hooks["tool.execute.after"]?.( + { sessionID, callID: call.callID, tool: "read", args: {} }, + { title: "read", output: call.output, metadata: {} }, + ); + } + await startGeneration({ + id: "multiple-tool-sources-followup-step", + sessionID, + assistantMessageID: "multiple-tool-sources-followup", + started: startedAt + 500, + }); + await completeGeneration({ + sessionID, + userMessageID, + assistantMessageID: "multiple-tool-sources-followup", + started: startedAt + 500, + completed: startedAt + 600, + text: "Done", + }); + const { spans } = await flushSession(sessionID); + const followup = spans + .filter((span) => span.name === "opencode.generation") + .sort( + (a, b) => Number(a.startTimeUnixNano) - Number(b.startTimeUnixNano), + ) + .at(-1); + if (!followup) { + throw new Error("Expected followup generation"); + } + expect(getJsonAttribute(followup, "langfuse.observation.input")).toEqual([ + { role: "user", content: [{ type: "text", text: "Read two files" }] }, + ...calls.flatMap((call) => [ + { + role: "assistant", + content: `Reading ${call.callID}`, + tool_calls: [{ id: call.callID, name: "read", arguments: "{}" }], + }, + { + role: "tool", + name: "read", + tool_call_id: call.callID, + content: call.output, + }, + ]), + ]); + }, + ); + + test.each([ + "stale-content", + "missing-later-call", + "missing-later-step", + "snapshot-only-first-step", + "live-first-marker-only", + "snapshot-only-content", + "snapshot-ahead-of-live", + "empty-assistant", + "missing-earlier-call", + "hook-only-results", + ] as const)("preserves assistant steps with %s history", async (snapshot) => { + const sessionID = "assistant-step-history-session"; + const userMessageID = "assistant-step-history-user"; + const assistantMessageID = "assistant-step-history-assistant"; + const steps = ["first", "second"].map((name) => ({ + name, + marker: { + id: `${name}-step`, + sessionID, + messageID: assistantMessageID, + type: "step-start" as const, + }, + text: { + id: `${name}-text`, + sessionID, + messageID: assistantMessageID, + type: "text" as const, + text: `Reading ${name} file`, + }, + reasoning: { + id: `${name}-reasoning`, + sessionID, + messageID: assistantMessageID, + type: "reasoning" as const, + text: `Need ${name} file`, + time: { start: startedAt + 100, end: startedAt + 200 }, + }, + tool: { + id: `${name}-tool`, + sessionID, + messageID: assistantMessageID, + type: "tool" as const, + callID: `${name}-call`, + tool: "read", + state: { + status: "completed" as const, + input: { path: `${name}.txt` }, + title: `${name}.txt`, + output: `${name} response`, + metadata: {}, + time: { start: startedAt + 200, end: startedAt + 300 }, + }, + }, + })); + await sendUserMessage({ + sessionID, + messageID: userMessageID, + text: "Read two files", + started: startedAt, + }); + await startAssistantMessage({ + sessionID, + userMessageID, + assistantMessageID, + started: startedAt + 100, + }); + await startGeneration({ + id: "assistant-step-history-generation", + sessionID, + assistantMessageID, + started: startedAt + 100, + }); + for (const step of steps) { + if (snapshot === "snapshot-only-first-step" && step.name === "first") { + continue; + } + for (const part of [step.marker, step.text, step.reasoning, step.tool]) { + if ( + snapshot === "live-first-marker-only" && + step.name === "first" && + part.type !== "step-start" + ) { + continue; + } + if ( + snapshot === "snapshot-only-content" && + step.name === "second" && + (part.type === "text" || part.type === "reasoning") + ) { + continue; + } + const livePart = (() => { + if ( + snapshot === "snapshot-ahead-of-live" && + step.name === "second" && + (part.type === "text" || part.type === "reasoning") + ) { + return { ...part, text: part.text.split(" ")[0] }; + } + if ( + part.type === "tool" && + (snapshot === "hook-only-results" || + (snapshot === "snapshot-ahead-of-live" && step.name === "second")) + ) { + return { + ...part, + state: { + status: "running" as const, + input: part.state.input, + time: { start: startedAt + 200 }, + }, + }; + } + return part; + })(); + await emitEvent({ + type: "message.part.updated", + properties: { part: livePart }, + }); + } + if (snapshot === "hook-only-results") { + await hooks["tool.execute.after"]?.( + { + sessionID, + callID: step.tool.callID, + tool: step.tool.tool, + args: step.tool.state.input, + }, + { + title: step.tool.state.title, + output: step.tool.state.output, + metadata: {}, + }, + ); + } + } + await completeGeneration({ + sessionID, + userMessageID, + assistantMessageID, + started: startedAt + 100, + completed: startedAt + 400, + }); + storeMessage(sessionID, { + info: { id: assistantMessageID, role: "assistant", sessionID }, + parts: steps.flatMap((step) => { + if (snapshot === "empty-assistant") { + return []; + } + if (snapshot === "missing-earlier-call" && step.name === "first") { + return [step.marker, step.text, step.reasoning]; + } + if (step.name === "first") { + return [step.marker, step.text, step.reasoning, step.tool]; + } + if ( + snapshot === "snapshot-only-content" || + snapshot === "snapshot-ahead-of-live" || + snapshot === "missing-earlier-call" + ) { + return [step.marker, step.text, step.reasoning, step.tool]; + } + if ( + snapshot === "missing-later-step" || + snapshot === "live-first-marker-only" + ) { + return []; + } + const content = [ + step.marker, + { ...step.text, text: "Reading" }, + { ...step.reasoning, text: "Need" }, + ]; + return snapshot === "missing-later-call" || + snapshot === "hook-only-results" + ? content + : [...content, step.tool]; + }), + }); + await startGeneration({ + id: "assistant-step-history-followup", + sessionID, + assistantMessageID: "assistant-step-history-followup-message", + started: startedAt + 500, + }); + await completeGeneration({ + sessionID, + userMessageID, + assistantMessageID: "assistant-step-history-followup-message", + started: startedAt + 500, + completed: startedAt + 600, + text: "Done", + }); + const { spans } = await flushSession(sessionID); + const followup = spans.find((span) => { + if (span.name !== "opencode.generation") { + return false; + } + const metadata = getJsonAttribute(span, "langfuse.observation.metadata"); + return ( + typeof metadata === "object" && + metadata !== null && + "messageID" in metadata && + metadata.messageID === "assistant-step-history-followup-message" + ); + }); + if (!followup) { + throw new Error("Expected followup generation"); + } + expect(getJsonAttribute(followup, "langfuse.observation.input")).toEqual([ + { role: "user", content: [{ type: "text", text: "Read two files" }] }, + ...steps.flatMap((step) => [ + { + role: "assistant", + content: step.text.text, + thinking: [{ type: "thinking", content: step.reasoning.text }], + tool_calls: [ + { + id: step.tool.callID, + name: "read", + arguments: JSON.stringify(step.tool.state.input), + }, + ], + }, + { + role: "tool", + name: "read", + tool_call_id: step.tool.callID, + content: step.tool.state.output, + }, + ]), + ]); + }); + test("creates a nested tool observation from message parts without execution hooks", async () => { const sessionID = "message-part-tool-session"; const userMessageID = "message-part-tool-user"; @@ -1649,8 +2328,10 @@ describe("built plugin", { concurrent: false }, () => { expect( getJsonAttribute(toolSpans[0], "langfuse.observation.output"), ).toEqual({ - title: "Recent commits", - output: "07f9a68 Fix tool observation parenting", + role: "tool", + name: "bash", + tool_call_id: callID, + content: "07f9a68 Fix tool observation parenting", }); expect( getJsonAttribute(secondGeneration, "langfuse.observation.input"), diff --git a/test/integration/v2-exports.test.ts b/test/integration/v2-exports.test.ts new file mode 100644 index 0000000..6789e85 --- /dev/null +++ b/test/integration/v2-exports.test.ts @@ -0,0 +1,510 @@ +import { createServer } from "node:http"; + +import LangfusePlugin from "@langfuse/opencode-observability-plugin/v2"; +import { Agent, Model, Provider } from "@opencode/plugin"; +import type { Hooks } from "@opencode/plugin/promise/registration"; +import type { SessionHooks as SessionHookInputs } from "@opencode/plugin/promise/session"; +import { CallID, Error as ToolError } from "@opencode/plugin/promise/tool"; +import { Schema } from "effect"; +import { expect, onTestFinished, test, vi } from "vitest"; + +type PluginContext = Parameters[0]; +type ToolHook = Parameters[1]; +type ToolHookInputs = + PluginContext["tool"]["hook"] extends Hooks ? Inputs : never; + +// These SDK IDs are string brands with no additional runtime constraints. +const SessionIDSchema = Schema.declare( + (value): value is Parameters[0]["sessionID"] => + typeof value === "string", +); +const MessageIDSchema = Schema.declare( + (value): value is Parameters[0]["messageID"] => + typeof value === "string", +); + +const SpanSchema = Schema.Struct({ + name: Schema.String, + spanId: Schema.String, + parentSpanId: Schema.optional(Schema.String), + attributes: Schema.Array( + Schema.Struct({ + key: Schema.String, + value: Schema.Struct({ stringValue: Schema.optional(Schema.String) }), + }), + ), + status: Schema.optional( + Schema.Struct({ + code: Schema.optional(Schema.Number), + message: Schema.optional(Schema.String), + }), + ), +}); +const PayloadSchema = Schema.parseJson( + Schema.Struct({ + resourceSpans: Schema.Array( + Schema.Struct({ + scopeSpans: Schema.Array( + Schema.Struct({ spans: Schema.Array(SpanSchema) }), + ), + }), + ), + }), +); + +const getJsonAttribute = (span: typeof SpanSchema.Type, key: string) => { + const value = span.attributes.find((attribute) => attribute.key === key) + ?.value.stringValue; + if (value === undefined) { + throw new Error(`Expected ${span.name} attribute ${key}`); + } + return Schema.decodeUnknownSync(Schema.parseJson(Schema.Unknown))(value); +}; + +test("exports V2 parallel tool responses with explicit IDs and no duplicate context results", async () => { + const spans: (typeof SpanSchema.Type)[] = []; + const collectorErrors: unknown[] = []; + const server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("error", (error) => { + collectorErrors.push(error); + }); + request.on("data", (chunk: Buffer) => { + chunks.push(chunk); + }); + request.on("end", () => { + try { + const payload = Schema.decodeUnknownSync(PayloadSchema)( + Buffer.concat(chunks).toString("utf8"), + ); + spans.push( + ...payload.resourceSpans.flatMap((resource) => + resource.scopeSpans.flatMap((scope) => scope.spans), + ), + ); + response.writeHead(200, { "content-type": "application/json" }); + response.end("{}"); + } catch (error) { + collectorErrors.push(error); + response.writeHead(400); + response.end("{}"); + } + }); + }); + onTestFinished(async () => { + try { + await new Promise((resolve, reject) => { + server.close((error) => { + if (error) { + reject(error); + return; + } + resolve(); + }); + }); + } finally { + vi.unstubAllEnvs(); + } + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (address === null || typeof address === "string") { + throw new Error("Expected a TCP collector address"); + } + vi.stubEnv("LANGFUSE_PUBLIC_KEY", "pk-test"); + vi.stubEnv("LANGFUSE_SECRET_KEY", "sk-test"); + vi.stubEnv( + "LANGFUSE_BASE_URL", + `http://127.0.0.1:${address.port.toString()}`, + ); + vi.stubEnv("LANGFUSE_ENVIRONMENT", "integration-test"); + + const createHookRegistry = () => { + const hooks: { + [Name in keyof Inputs]?: (input: Inputs[Name]) => Promise | void; + } = {}; + return { + register: ( + name: Name, + handler: (input: Inputs[Name]) => Promise | void, + ) => { + hooks[name] = handler; + return Promise.resolve({ + dispose: () => { + hooks[name] = undefined; + return Promise.resolve(); + }, + }); + }, + run: async ( + name: Name, + input: Inputs[Name], + ) => { + const handler = hooks[name]; + if (!handler) { + throw new Error(`Expected hook ${String(name)}`); + } + await handler(input); + }, + }; + }; + const sessionHooks = createHookRegistry(); + const toolHooks = createHookRegistry(); + const eventsFinished = Promise.withResolvers(); + + const sessionID = + Schema.decodeUnknownSync(SessionIDSchema)("v2-parallel-tools"); + const assistantMessageID = + Schema.decodeUnknownSync(MessageIDSchema)("v2-tool-generation"); + const agent = Agent.ID.make("build"); + const model = { + id: Model.ID.make("test-model"), + providerID: Provider.ID.make("test-provider"), + }; + const tokens = { + input: 10, + output: 5, + reasoning: 0, + cache: { read: 0, write: 0 }, + }; + const calls = [ + { + id: CallID.make("v2-call-first"), + path: "first.txt", + content: "First\nresponse", + status: "completed" as const, + }, + { + id: CallID.make("v2-call-second"), + path: "second.txt", + content: "Second\n\nresponse", + status: "completed" as const, + }, + { + id: CallID.make("v2-call-missing"), + path: "missing.txt", + content: "ENOENT: missing.txt", + status: "error" as const, + }, + ]; + const toolCalls = calls.map((call) => ({ + id: call.id, + name: "read", + arguments: JSON.stringify({ path: call.path }), + })); + const results = calls.map((call) => ({ + role: "tool", + name: "read", + tool_call_id: call.id, + content: call.content, + })); + const messages = [ + { + role: "user" as const, + content: [{ type: "text" as const, text: "Read three files" }], + }, + { + role: "assistant" as const, + content: calls.map((call) => ({ + type: "tool-call" as const, + id: call.id, + name: "read", + input: { path: call.path }, + })), + }, + ...calls.map((call) => ({ + role: "tool" as const, + content: [ + { + type: "tool-result" as const, + id: call.id, + name: "read", + result: { + type: ( + { + completed: "text", + error: "error", + } as const satisfies Record< + (typeof calls)[number]["status"], + "text" | "error" + > + )[call.status], + value: call.content, + }, + }, + ], + })), + ] satisfies SessionHookInputs["context"]["messages"]; + const unexpectedApi = () => { + throw new Error("The plugin used an unsupported mock host capability"); + }; + const contextInput = { + app: { name: "opencode", version: "2.0.4", channel: "stable" }, + session: { + hook: sessionHooks.register, + create: unexpectedApi, + get: unexpectedApi, + switchAgent: unexpectedApi, + switchModel: unexpectedApi, + prompt: unexpectedApi, + generate: unexpectedApi, + command: unexpectedApi, + synthetic: unexpectedApi, + interrupt: unexpectedApi, + update: unexpectedApi, + move: unexpectedApi, + wait: unexpectedApi, + context: unexpectedApi, + }, + tool: { + hook: toolHooks.register, + transform: unexpectedApi, + reload: unexpectedApi, + }, + event: { + subscribe: () => ({ + async *[Symbol.asyncIterator]() { + try { + await sessionHooks.run("prompt", { + sessionID, + messageID: Schema.decodeUnknownSync(MessageIDSchema)("v2-user"), + prompt: { text: "Read three files" }, + delivery: "steer", + }); + yield { + id: "v2-tools-started", + durable: { aggregateID: sessionID, seq: 1, version: 1 as const }, + type: "session.step.started", + created: Date.now(), + data: { sessionID, assistantMessageID, agent, model }, + }; + for (const call of calls) { + await toolHooks.run("execute.before", { + sessionID, + messageID: assistantMessageID, + agent, + tool: "read", + id: call.id, + input: { path: call.path }, + }); + } + // Same names and reversed completions must still match by call ID. + for (const call of [...calls].reverse()) { + const input = { + sessionID, + messageID: assistantMessageID, + agent, + tool: "read", + id: call.id, + input: { path: call.path }, + }; + const resultsByStatus = { + error: { + ...input, + status: "error" as const, + error: new ToolError({ message: call.content }), + }, + completed: { + ...input, + status: "completed" as const, + result: { + content: + call.id === "v2-call-second" + ? [ + { type: "text" as const, text: "Second" }, + { type: "text" as const, text: "response" }, + ] + : call.content, + }, + }, + } satisfies Record< + (typeof calls)[number]["status"], + Parameters[0] + >; + const result = resultsByStatus[call.status]; + await toolHooks.run("execute.after", result); + await toolHooks.run("execute.after", result); + } + yield { + id: "v2-tools-ended", + durable: { aggregateID: sessionID, seq: 2, version: 1 as const }, + type: "session.step.ended", + created: Date.now(), + data: { + sessionID, + assistantMessageID, + finish: "tool-calls", + cost: 0, + tokens, + }, + }; + await sessionHooks.run("context", { + sessionID, + agent, + model, + system: [], + messages, + options: {}, + tools: {}, + }); + yield { + id: "v2-followup-started", + durable: { aggregateID: sessionID, seq: 3, version: 1 as const }, + type: "session.step.started", + created: Date.now(), + data: { + sessionID, + assistantMessageID: "v2-followup", + agent, + model, + }, + }; + yield { + id: "v2-followup-ended", + durable: { aggregateID: sessionID, seq: 4, version: 1 as const }, + type: "session.step.ended", + created: Date.now(), + data: { + sessionID, + assistantMessageID: "v2-followup", + finish: "stop", + cost: 0, + tokens, + }, + }; + yield { + id: "v2-execution-succeeded", + durable: { aggregateID: sessionID, seq: 5, version: 1 as const }, + type: "session.execution.succeeded", + created: Date.now(), + data: { sessionID }, + }; + } finally { + eventsFinished.resolve(undefined); + } + }, + }), + }, + get location() { + return unexpectedApi(); + }, + get options() { + return unexpectedApi(); + }, + get agent() { + return unexpectedApi(); + }, + get aisdk() { + return unexpectedApi(); + }, + get command() { + return unexpectedApi(); + }, + get experimental() { + return unexpectedApi(); + }, + get integration() { + return unexpectedApi(); + }, + get mcp() { + return unexpectedApi(); + }, + get model() { + return unexpectedApi(); + }, + get generate() { + return unexpectedApi(); + }, + get permission() { + return unexpectedApi(); + }, + get plugin() { + return unexpectedApi(); + }, + get provider() { + return unexpectedApi(); + }, + get reference() { + return unexpectedApi(); + }, + get rpc() { + return unexpectedApi(); + }, + get shell() { + return unexpectedApi(); + }, + get skill() { + return unexpectedApi(); + }, + get storage() { + return unexpectedApi(); + }, + get vcs() { + return unexpectedApi(); + }, + get websearch() { + return unexpectedApi(); + }, + get worktree() { + return unexpectedApi(); + }, + } satisfies PluginContext; + // This mock host does not prove compatibility with a running OpenCode server. + const cleanup = await LangfusePlugin.setup(contextInput); + if (!cleanup) { + throw new Error("Expected plugin cleanup"); + } + onTestFinished(cleanup); + await eventsFinished.promise; + + expect(collectorErrors).toEqual([]); + const tools = spans.filter((span) => span.name === "read"); + expect(tools).toHaveLength(calls.length); + const generations = spans.filter( + (span) => span.name === "opencode.generation", + ); + expect(generations).toHaveLength(2); + const generation = generations.find( + (span) => span.spanId === tools[0].parentSpanId, + ); + const followup = generations.find((span) => span !== generation); + if (!generation || !followup) { + throw new Error("Expected both generations"); + } + expect(getJsonAttribute(generation, "langfuse.observation.output")).toEqual([ + { role: "assistant", content: "", tool_calls: toolCalls }, + ]); + for (const [index, call] of calls.entries()) { + const tool = tools.find((span) => { + const output = getJsonAttribute(span, "langfuse.observation.output"); + return ( + typeof output === "object" && + output !== null && + "tool_call_id" in output && + output.tool_call_id === call.id + ); + }); + if (!tool) { + throw new Error(`Expected response for ${call.id}`); + } + expect(getJsonAttribute(tool, "langfuse.observation.output")).toEqual( + results[index], + ); + expect(tool.parentSpanId).toBe(generation.spanId); + expect(getJsonAttribute(tool, "langfuse.observation.metadata")).toEqual({ + callID: call.id, + tool: "read", + }); + if (call.status === "error") { + expect(tool.status).toEqual({ code: 2, message: call.content }); + } + } + expect(getJsonAttribute(followup, "langfuse.observation.input")).toEqual({ + system: [], + messages, + tools: [], + }); +});