From 4b934dfffe51dbcade330f6353331c77d73e6047 Mon Sep 17 00:00:00 2001 From: phamngocquy Date: Fri, 31 Jul 2026 21:21:10 +0800 Subject: [PATCH] mcp(telemetry): put the tool call's outcome on the OTel span instead of only in the log line Fixes #10042 --- src/mcp/dispatch-span-registry.ts | 8 +- src/mcp/dispatch-telemetry-sink.ts | 11 ++- src/mcp/dispatch-telemetry.ts | 62 ++++++++------ src/server.ts | 4 +- test/unit/mcp-dispatch-telemetry-sink.test.ts | 32 ++++++-- test/unit/mcp-dispatch-telemetry.test.ts | 82 ++++++++++++++++++- test/unit/selfhost-otel.test.ts | 30 +++++++ 7 files changed, 193 insertions(+), 36 deletions(-) diff --git a/src/mcp/dispatch-span-registry.ts b/src/mcp/dispatch-span-registry.ts index 6dd37efa28..4a401605a9 100644 --- a/src/mcp/dispatch-span-registry.ts +++ b/src/mcp/dispatch-span-registry.ts @@ -8,7 +8,13 @@ // means every tool call runs unwrapped, at zero cost. That asymmetry is deliberate and is why this // is a registry rather than a direct import: Workers has no OTel collector to export to, and pulling // the tracer into that bundle would cost real bytes for a capability it cannot use. -export type McpDispatchSpanRunner = (name: string, attributes: Record, fn: () => Promise) => Promise; +import type { SetMcpSpanOutcomeAttributes } from "./dispatch-telemetry"; + +export type McpDispatchSpanRunner = ( + name: string, + attributes: Record, + fn: (setOutcomeAttributes: SetMcpSpanOutcomeAttributes) => Promise, +) => Promise; let runner: McpDispatchSpanRunner | null = null; diff --git a/src/mcp/dispatch-telemetry-sink.ts b/src/mcp/dispatch-telemetry-sink.ts index 863ab8faf3..f096edae28 100644 --- a/src/mcp/dispatch-telemetry-sink.ts +++ b/src/mcp/dispatch-telemetry-sink.ts @@ -26,7 +26,7 @@ import { type McpAnalyticsContext, type McpInitializeTelemetry, } from "@loopover/contract"; -import type { DispatchTelemetrySink } from "./dispatch-telemetry"; +import type { DispatchTelemetrySink, SetMcpSpanOutcomeAttributes } from "./dispatch-telemetry"; import { getMcpDispatchSpanRunner } from "./dispatch-span-registry"; const DEFAULT_POSTHOG_HOST = "https://us.i.posthog.com"; @@ -84,7 +84,11 @@ async function captureEvents( export function createDispatchTelemetrySink( env: DispatchTelemetryEnv, defer: DeferWork, - withSpan?: (name: string, attributes: Record, fn: () => Promise) => Promise, + withSpan?: ( + name: string, + attributes: Record, + fn: (setOutcomeAttributes: SetMcpSpanOutcomeAttributes) => Promise, + ) => Promise, context: McpAnalyticsContext = {}, ): DispatchTelemetrySink { return { @@ -102,7 +106,8 @@ export function createDispatchTelemetrySink( }, // The registry is consulted per call rather than captured at construction so a self-host boot // that fills the slot after the first request still traces. - withSpan: (name, attributes, fn) => (withSpan ?? getMcpDispatchSpanRunner() ?? ((_n, _a, run) => run()))(name, attributes, fn), + withSpan: (name, attributes, fn) => + (withSpan ?? getMcpDispatchSpanRunner() ?? ((_n, _a, run) => run(() => {})))(name, attributes, fn), }; } diff --git a/src/mcp/dispatch-telemetry.ts b/src/mcp/dispatch-telemetry.ts index 70960bcce9..a2ec629136 100644 --- a/src/mcp/dispatch-telemetry.ts +++ b/src/mcp/dispatch-telemetry.ts @@ -43,13 +43,20 @@ function log(level: "warn" | "error", event: string, fields: Record) => void; + export type DispatchTelemetrySink = { /** Both usage events. Never throws. */ recordToolCall: (call: McpToolCallTelemetry, properties: { usage: Record; mcpToolCall: Record }) => void; /** A genuine throw. Never throws. */ captureException: (error: unknown, call: McpToolCallTelemetry) => void; /** Wrap the call in a span when tracing is on; a no-op passthrough when it is not. */ - withSpan: (name: string, attributes: Record, fn: () => Promise) => Promise; + withSpan: ( + name: string, + attributes: Record, + fn: (setOutcomeAttributes: SetMcpSpanOutcomeAttributes) => Promise, + ) => Promise; /** * Session/server/client identity for the canonical `$mcp_*` events (#10175). * @@ -65,7 +72,7 @@ export type DispatchTelemetrySink = { export const NOOP_DISPATCH_SINK: DispatchTelemetrySink = { recordToolCall: () => undefined, captureException: () => undefined, - withSpan: async (_name, _attributes, fn) => fn(), + withSpan: async (_name, _attributes, fn) => fn(() => {}), }; function describe(toolName: string): { category: string; excluded: boolean } { @@ -91,7 +98,6 @@ export function instrumentToolDispatch => { const { category, excluded } = describe(toolName); const startedAt = Date.now(); - const attributes = { tool: toolName, category, surface: "remote" as const }; const emit = (call: McpToolCallTelemetry, payloads: { arguments?: unknown; result?: unknown }): void => { try { @@ -104,8 +110,16 @@ export function instrumentToolDispatch { + const publishSpanOutcome = (setOutcomeAttributes: SetMcpSpanOutcomeAttributes, call: McpToolCallTelemetry): void => { + try { + setOutcomeAttributes(buildMcpToolSpanAttributes(call)); + } catch { + // Telemetry must never surface into the tool caller. + } + }; + + return await sink.withSpan(mcpToolSpanName(toolName), {}, async (setOutcomeAttributes) => { + try { const result = await handler(...args); const ok = result?.isError !== true; const call: McpToolCallTelemetry = { @@ -121,29 +135,31 @@ export function instrumentToolDispatch { // #9525: hand the MCP dispatch chokepoint a real span runner. Only this entry does -- the cloud // Worker has no collector to export to, so its slot stays null and every tool call runs // unwrapped. Registry rather than a direct import so ./selfhost/otel never enters that bundle. - setMcpDispatchSpanRunner((name, attributes, fn) => withOtelSpan(name, attributes, fn)); + setMcpDispatchSpanRunner((name, attributes, fn) => + withOtelSpan(name, attributes, () => fn((outcome) => setCurrentOtelSpanAttributes(outcome))), + ); } /* v8 ignore stop */ const startedAt = Date.now(); diff --git a/test/unit/mcp-dispatch-telemetry-sink.test.ts b/test/unit/mcp-dispatch-telemetry-sink.test.ts index 4b6a1c30a6..50cff48dfd 100644 --- a/test/unit/mcp-dispatch-telemetry-sink.test.ts +++ b/test/unit/mcp-dispatch-telemetry-sink.test.ts @@ -26,7 +26,11 @@ afterEach(() => { describe("MCP dispatch span registry (#9525)", () => { it("is empty until a self-host boot fills it, and clears again", () => { expect(getMcpDispatchSpanRunner()).toBeUndefined(); - const runner = async (_name: string, _attributes: Record, fn: () => Promise): Promise => fn(); + const runner = async ( + _name: string, + _attributes: Record, + fn: (setOutcomeAttributes: (attributes: Record) => void) => Promise, + ): Promise => fn(() => {}); setMcpDispatchSpanRunner(runner); expect(getMcpDispatchSpanRunner()).toBe(runner); setMcpDispatchSpanRunner(null); @@ -98,21 +102,35 @@ describe("MCP dispatch telemetry sink (#9525)", () => { const seen: Array<{ name: string; attributes: Record }> = []; setMcpDispatchSpanRunner(async (name, attributes, fn) => { seen.push({ name, attributes }); - return fn(); + return fn(() => {}); }); const sink = createDispatchTelemetrySink(env(), () => undefined); await expect(sink.withSpan("mcp.tool/x", { tool: "x" }, async () => "wrapped")).resolves.toBe("wrapped"); expect(seen).toEqual([{ name: "mcp.tool/x", attributes: { tool: "x" } }]); }); + it("forwards setOutcomeAttributes through the registry runner (#10042)", async () => { + const outcomes: Record[] = []; + setMcpDispatchSpanRunner(async (_name, _attributes, fn) => fn((attrs) => outcomes.push(attrs))); + const sink = createDispatchTelemetrySink(env(), () => undefined); + await sink.withSpan("mcp.tool/x", {}, async (setOutcome) => { + setOutcome({ ok: false, error_code: "timeout" }); + }); + expect(outcomes).toEqual([{ ok: false, error_code: "timeout" }]); + }); + it("prefers an explicitly injected runner over the registry", async () => { setMcpDispatchSpanRunner(async () => { throw new Error("registry runner should not have been used"); }); let injectedCalls = 0; - const injected = async (_name: string, _attributes: Record, fn: () => Promise): Promise => { + const injected = async ( + _name: string, + _attributes: Record, + fn: (setOutcomeAttributes: (attributes: Record) => void) => Promise, + ): Promise => { injectedCalls += 1; - return fn(); + return fn(() => {}); }; const sink = createDispatchTelemetrySink(env(), () => undefined, injected); await expect(sink.withSpan("mcp.tool/x", {}, async () => "injected")).resolves.toBe("injected"); @@ -131,7 +149,11 @@ describe("LoopoverMcp telemetry-sink injection (#9525)", () => { const sink = { recordToolCall: (entry: McpToolCallTelemetry) => recorded.push(entry), captureException: () => undefined, - withSpan: async (_name: string, _attributes: Record, fn: () => Promise) => fn(), + withSpan: async ( + _name: string, + _attributes: Record, + fn: (setOutcomeAttributes: (attributes: Record) => void) => Promise, + ) => fn(() => {}), }; const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); diff --git a/test/unit/mcp-dispatch-telemetry.test.ts b/test/unit/mcp-dispatch-telemetry.test.ts index 90563cf5d9..7d48dfe265 100644 --- a/test/unit/mcp-dispatch-telemetry.test.ts +++ b/test/unit/mcp-dispatch-telemetry.test.ts @@ -240,7 +240,7 @@ describe("MCP dispatch chokepoint (#9525)", () => { sink: { recordToolCall: (recorded) => calls.push(recorded), captureException: (error) => exceptions.push(error), - withSpan: async (_name, _attributes, fn) => fn(), + withSpan: async (_name, _attributes, fn) => fn(() => {}), }, }; }; @@ -316,7 +316,7 @@ describe("MCP dispatch chokepoint (#9525)", () => { captureException: () => { throw new Error("sink down"); }, - withSpan: async (_name, _attributes, fn) => fn(), + withSpan: async (_name, _attributes, fn) => fn(() => {}), }; const ok = instrumentToolDispatch("loopover_get_repo_context", hostile, async (_args: unknown) => ({ structuredContent: { fine: true } })); await expect(ok({})).resolves.toMatchObject({ structuredContent: { fine: true } }); @@ -338,7 +338,10 @@ describe("MCP dispatch chokepoint (#9525)", () => { // every single call, so "does nothing and returns nothing" is worth asserting outright. expect(NOOP_DISPATCH_SINK.recordToolCall(call, { usage: {}, mcpToolCall: {} })).toBeUndefined(); expect(NOOP_DISPATCH_SINK.captureException(new Error("x"), call)).toBeUndefined(); - await expect(NOOP_DISPATCH_SINK.withSpan("n", {}, async () => "through")).resolves.toBe("through"); + await expect(NOOP_DISPATCH_SINK.withSpan("n", {}, async (setOutcome) => { + setOutcome({ ok: false }); + return "through"; + })).resolves.toBe("through"); }); it("falls back to the unknown category for a tool with no contract entry", async () => { @@ -349,6 +352,79 @@ describe("MCP dispatch chokepoint (#9525)", () => { }); }); +describe("MCP dispatch span outcome attributes (#10042)", () => { + it("publishes buildMcpToolSpanAttributes onto the span on success and on throw", async () => { + const error = vi.spyOn(console, "error").mockImplementation(() => undefined); + const spans: Array<{ name: string; open: Record; outcome?: Record }> = []; + const recordingSink: DispatchTelemetrySink = { + recordToolCall: () => undefined, + captureException: () => undefined, + withSpan: async (name, attributes, fn) => { + let outcome: Record | undefined; + try { + const result = await fn((attrs) => { + outcome = attrs; + }); + spans.push({ name, open: attributes, ...(outcome !== undefined ? { outcome } : {}) }); + return result; + } catch (error) { + spans.push({ name, open: attributes, ...(outcome !== undefined ? { outcome } : {}) }); + throw error; + } + }, + }; + + const okWrapped = instrumentToolDispatch("loopover_get_repo_context", recordingSink, async (_args: unknown) => ({ + structuredContent: { ok: 1 }, + })); + await okWrapped({}); + expect(spans[0]).toMatchObject({ name: "mcp.tool/loopover_get_repo_context", open: {} }); + expect(spans[0]!.outcome).toMatchObject({ + tool: "loopover_get_repo_context", + category: "maintainer", + surface: "remote", + transport: "local", + ok: true, + }); + expect("error_code" in spans[0]!.outcome!).toBe(false); + + const throwWrapped = instrumentToolDispatch("loopover_get_repo_context", recordingSink, async (_args: unknown) => { + throw new Error("request timed out"); + }); + await expect(throwWrapped({})).rejects.toThrow("request timed out"); + expect(spans[1]!.outcome).toMatchObject({ ok: false, error_code: "timeout" }); + expect(MCP_TELEMETRY_ERROR_CODES).toContain(spans[1]!.outcome!.error_code); + + error.mockRestore(); + }); + + it("keeps NOOP_DISPATCH_SINK.withSpan a pure passthrough that records nothing", async () => { + let setterCalled = false; + await expect( + NOOP_DISPATCH_SINK.withSpan("mcp.tool/x", { tool: "x" }, async (setOutcome) => { + setOutcome({ ok: false, error_code: "timeout" }); + setterCalled = true; + return "through"; + }), + ).resolves.toBe("through"); + expect(setterCalled).toBe(true); + }); + + it("never lets a failing setOutcomeAttributes reach the caller", async () => { + const recordingSink: DispatchTelemetrySink = { + recordToolCall: () => undefined, + captureException: () => undefined, + withSpan: async (_name, _attributes, fn) => fn(() => { + throw new Error("span attrs down"); + }), + }; + const wrapped = instrumentToolDispatch("loopover_get_repo_context", recordingSink, async (_args: unknown) => ({ + structuredContent: { ok: 1 }, + })); + await expect(wrapped({})).resolves.toMatchObject({ structuredContent: { ok: 1 } }); + }); +}); + describe("PostHog canonical MCP analytics contract (#10175)", () => { const ctx = { sessionId: "ses_abc123", serverName: "loopover", serverVersion: "3.18.4", clientName: "claude-code", clientVersion: "1.2.3" }; diff --git a/test/unit/selfhost-otel.test.ts b/test/unit/selfhost-otel.test.ts index 25824c4338..58d057fa8c 100644 --- a/test/unit/selfhost-otel.test.ts +++ b/test/unit/selfhost-otel.test.ts @@ -1,4 +1,5 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; +import { buildMcpToolSpanAttributes } from "@loopover/contract"; const otelMocks = vi.hoisted(() => { const exportedSpans: any[] = []; @@ -484,6 +485,35 @@ describe("self-host OpenTelemetry", () => { await expect(reviewTraceAttributes({})).resolves.toEqual({}); }); + it("applies MCP dispatch span outcome attributes through otelSafeAttributes (#10042)", async () => { + await initOpenTelemetry(env({ + OTEL_TRACES_EXPORTER: "otlp", + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "http://collector/v1/traces", + })); + const call = { + tool: "loopover_get_repo_context", + category: "maintainer", + surface: "remote" as const, + ok: false, + durationMs: 7, + errorCode: "timeout" as const, + }; + await withOtelSpan("mcp.tool/loopover_get_repo_context", {}, () => + Promise.resolve().then(() => setCurrentOtelSpanAttributes(buildMcpToolSpanAttributes(call))), + ); + await flushOpenTelemetry(); + const span = otelMocks.exportedSpans.find((entry) => entry.name === "mcp.tool/loopover_get_repo_context"); + expect(span.attributes).toMatchObject({ + tool: "loopover_get_repo_context", + category: "maintainer", + surface: "remote", + transport: "local", + ok: false, + duration_ms: 7, + error_code: "timeout", + }); + }); + it("swallows exporter flush and shutdown failures", async () => { await initOpenTelemetry(env({ OTEL_TRACES_EXPORTER: "otlp",