diff --git a/apps/cloud/src/mcp/agent-handler.ts b/apps/cloud/src/mcp/agent-handler.ts index 914c7b024..33b5f88e2 100644 --- a/apps/cloud/src/mcp/agent-handler.ts +++ b/apps/cloud/src/mcp/agent-handler.ts @@ -1,3 +1,4 @@ +import * as OtelTracer from "@effect/opentelemetry/Tracer"; import { Effect, Predicate } from "effect"; import { @@ -17,8 +18,10 @@ import type { McpSessionProps } from "@executor-js/cloudflare/mcp/agent-durable- import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { wrapMcpSseResponse } from "../observability/memory-metrics"; +import { WorkerTelemetryLive } from "../observability/telemetry"; import { cloudMcpAuth } from "./auth-provider"; import { McpSessionDOSqlite } from "./session-durable-object"; +import { parseTraceparent } from "./traceparent"; const corsPreflightResponse = (): Response => new Response(null, { @@ -82,6 +85,28 @@ const authenticate = (request: Request) => return { auth, outcome }; }).pipe(Effect.provide(cloudMcpAuth)); +// The pre-Agents envelope ran the MCP auth path inside the Effect app, whose +// HttpMiddleware provided the OTEL tracer — that is where the `mcp.request` +// span (client fingerprint, rpc method, auth outcome) exported from. This +// handler dispatches from the raw worker entry instead, so a bare +// `Effect.runPromise` leaves every span in Effect's no-op default tracer and +// they silently never export. Run each program under the worker telemetry +// layer, parented to the edge `http.server` span server.ts stamps onto the +// forwarded request's traceparent — which also makes `currentPropagationHeaders` +// (via Effect.currentParentSpan) ferry that same trace into the session DO +// instead of letting the DO start a fresh root per request. +const runTraced = (request: Request, program: Effect.Effect): Promise => { + const parsed = parseTraceparent( + request.headers.get("traceparent"), + request.headers.get("tracestate"), + ); + return Effect.runPromise( + (parsed ? OtelTracer.withSpanContext(program, parsed) : program).pipe( + Effect.provide(WorkerTelemetryLive), + ), + ); +}; + // The MCP resource the request targets. `server.ts` routes both the bare `/mcp` // and `/mcp/toolkits/` to this handler (`prepareMcpOrgScope` strips the org // selector but keeps the toolkit segment), so a session minted on a toolkit path @@ -142,7 +167,7 @@ export const makeCloudMcpAgentHandler = () => { } const sessionId = request.headers.get("mcp-session-id"); - const { auth, outcome } = await Effect.runPromise(authenticate(request)); + const { auth, outcome } = await runTraced(request, authenticate(request)); if (!Predicate.isTagged(outcome, "Authenticated")) { // Destroying a live session on auth grounds requires a POSITIVE // determination that access is genuinely gone — only `Forbidden` carries @@ -191,7 +216,7 @@ export const makeCloudMcpAgentHandler = () => { } const resource = resourceFromPath(request); - const props = await Effect.runPromise(propsForPrincipal(request, outcome.principal, resource)); + const props = await runTraced(request, propsForPrincipal(request, outcome.principal, resource)); (ctx as ExecutionContext & { props?: McpSessionProps }).props = props; const forwarded = withVerifiedIdentityHeaders( request, diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index b6b9ccb52..c2d249aef 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -14,7 +14,6 @@ // --------------------------------------------------------------------------- import { env } from "cloudflare:workers"; -import { createTraceState } from "@opentelemetry/api"; import { Data, Effect, Layer } from "effect"; import type { Cause } from "effect"; import * as OtelTracer from "@effect/opentelemetry/Tracer"; @@ -73,6 +72,7 @@ import { captureCauseEffect as reportCauseEffect, tagCurrentSentryScopeWithCurrentOtelSpan, } from "../observability"; +import { parseTraceparent } from "./traceparent"; // Re-export the shared types so existing cloud importers // (`auth/handlers.ts`, etc.) keep their `../mcp/session-durable-object` path. @@ -113,34 +113,6 @@ class McpModelResumeForwardError extends Data.TaggedError("McpModelResumeForward readonly cause: unknown; }> {} -// W3C propagation across the worker→DO boundary. The worker injects its -// `traceparent` and forwards incoming `tracestate` / `baggage`; we parse the -// context and use `OtelTracer.withSpanContext` to stitch the DO's root span -// under the worker span so the entire logical request lives in one trace. -const TRACEPARENT_PATTERN = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/; - -type IncomingSpanContext = { - readonly traceId: string; - readonly spanId: string; - readonly traceFlags: number; - readonly traceState?: ReturnType; -}; - -const parseTraceparent = ( - traceparent: string | null | undefined, - tracestate: string | null | undefined, -): IncomingSpanContext | null => { - if (!traceparent) return null; - const match = TRACEPARENT_PATTERN.exec(traceparent); - if (!match) return null; - return { - traceId: match[2]!, - spanId: match[3]!, - traceFlags: parseInt(match[4]!, 16), - ...(tracestate ? { traceState: createTraceState(tracestate) } : {}), - }; -}; - /** * The DO keeps one postgres.js client for the MCP session runtime. postgres.js * closes idle sockets quickly, while the runtime object stays alive so the MCP diff --git a/apps/cloud/src/mcp/traceparent.ts b/apps/cloud/src/mcp/traceparent.ts new file mode 100644 index 000000000..0be604025 --- /dev/null +++ b/apps/cloud/src/mcp/traceparent.ts @@ -0,0 +1,32 @@ +// --------------------------------------------------------------------------- +// W3C traceparent parsing shared by the worker edge (server.ts), the MCP agent +// handler, and the session DO. Single-sourced so the producer (the worker span +// stamping traceparent onto the forwarded request) and the consumers (the +// Effect programs joining that span) cannot drift on the header grammar. +// --------------------------------------------------------------------------- + +import { createTraceState } from "@opentelemetry/api"; + +const TRACEPARENT_PATTERN = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/; + +export type IncomingSpanContext = { + readonly traceId: string; + readonly spanId: string; + readonly traceFlags: number; + readonly traceState?: ReturnType; +}; + +export const parseTraceparent = ( + traceparent: string | null | undefined, + tracestate: string | null | undefined, +): IncomingSpanContext | null => { + if (!traceparent) return null; + const match = TRACEPARENT_PATTERN.exec(traceparent); + if (!match) return null; + return { + traceId: match[2]!, + spanId: match[3]!, + traceFlags: parseInt(match[4]!, 16), + ...(tracestate ? { traceState: createTraceState(tracestate) } : {}), + }; +}; diff --git a/apps/cloud/src/server.ts b/apps/cloud/src/server.ts index b814c2d65..12aac6773 100644 --- a/apps/cloud/src/server.ts +++ b/apps/cloud/src/server.ts @@ -14,6 +14,7 @@ import handler from "@tanstack/react-start/server-entry"; import { isAppOwnedPath } from "./app-paths"; import { makeCloudMcpAgentHandler } from "./mcp/agent-handler"; import { classifyMcpPath, prepareMcpOrgScope } from "./mcp/mount"; +import { parseTraceparent } from "./mcp/traceparent"; import { McpSessionDOSqlite as McpSessionDOBase } from "./mcp/session-durable-object"; import { beforeSendWithOtelCorrelation, @@ -124,16 +125,66 @@ const cloudflareHandler: ExportedHandler = { const url = new URL(request.url); const mcpRoute = classifyMcpPath(url.pathname); const tracingInstalled = installTracerProvider(); + // Join the caller's W3C trace when the request carries one — the web UI + // sends traceparent on every API fetch, so the browser's spans and this + // request share one trace id end to end. Same parsing the DO path does + // in session-durable-object.ts. + const inbound = parseTraceparent(request.headers.get("traceparent"), null); + const parentContext = inbound + ? trace.setSpanContext(context.active(), { + traceId: inbound.traceId, + spanId: inbound.spanId, + traceFlags: inbound.traceFlags, + isRemote: true, + }) + : context.active(); if (mcpRoute?.kind === "mcp") { // The Cloudflare Agents MCP bridge needs the platform ExecutionContext // to pass authenticated session props into the hibernatable DO. // Discovery docs still flow through the app-level MCP envelope. - // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary; keep trace export alive after the Agents bridge resolves or rejects - try { - return await mcpAgentHandler(prepareMcpOrgScope(request), env, ctx); - } finally { - if (tracingInstalled) ctx.waitUntil(flushTracerProvider()); + const forwarded = prepareMcpOrgScope(request); + if (!tracingInstalled) { + return mcpAgentHandler(forwarded, env, ctx); } + // /mcp left the Effect app in the Agents-bridge migration, so no + // downstream HttpMiddleware.tracer opens the request envelope anymore — + // this worker span is now THE `http.server` span for MCP traffic. Its + // context is stamped onto the forwarded request's traceparent so the + // agent handler's Effect programs (mcp.request and children) and the + // session DO parent under it instead of exporting orphaned roots. + return tracer.startActiveSpan( + `http.server ${request.method}`, + { kind: SpanKind.SERVER }, + parentContext, + async (span) => { + span.setAttribute(ATTR_HTTP_REQUEST_METHOD, request.method); + span.setAttribute(ATTR_URL_FULL, request.url); + span.setAttribute(ATTR_URL_PATH, url.pathname); + span.setAttribute(ATTR_URL_SCHEME, url.protocol.replace(/:$/, "")); + const spanContext = span.spanContext(); + const headers = new Headers(forwarded.headers); + headers.set( + "traceparent", + `00-${spanContext.traceId}-${spanContext.spanId}-${(spanContext.traceFlags & 0xff).toString(16).padStart(2, "0")}`, + ); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary; observe response/error for span status, keep trace export alive after the Agents bridge resolves or rejects + try { + const response = await mcpAgentHandler(new Request(forwarded, { headers }), env, ctx); + span.setAttribute(ATTR_HTTP_RESPONSE_STATUS_CODE, response.status); + if (response.status >= 500) { + span.setStatus({ code: SpanStatusCode.ERROR }); + } + return response; + } catch (err) { + span.setStatus({ code: SpanStatusCode.ERROR }); + // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary; preserve original error to Cloudflare runtime + throw err; + } finally { + span.end(); + ctx.waitUntil(flushTracerProvider()); + } + }, + ); } if (!tracingInstalled) { return fetchHandler(request, env, ctx); @@ -150,21 +201,6 @@ const cloudflareHandler: ExportedHandler = { ctx.waitUntil(flushTracerProvider()); } } - // Join the caller's W3C trace when the request carries one — the web UI - // sends traceparent on every API fetch, so the browser's spans and this - // request share one trace id end to end. Same parsing the DO path does - // in session-durable-object.ts. - const traceparentMatch = /^[0-9a-f]{2}-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec( - request.headers.get("traceparent") ?? "", - ); - const parentContext = traceparentMatch - ? trace.setSpanContext(context.active(), { - traceId: traceparentMatch[1]!, - spanId: traceparentMatch[2]!, - traceFlags: parseInt(traceparentMatch[3]!, 16), - isRemote: true, - }) - : context.active(); return tracer.startActiveSpan( `http.server ${request.method}`, { kind: SpanKind.SERVER }, diff --git a/e2e/cloud/mcp-telemetry-contract.test.ts b/e2e/cloud/mcp-telemetry-contract.test.ts new file mode 100644 index 000000000..91a7220bf --- /dev/null +++ b/e2e/cloud/mcp-telemetry-contract.test.ts @@ -0,0 +1,106 @@ +// Cloud: the /mcp observability contract. Every MCP request must land in the +// EXPORTED trace store as an `mcp.request` span carrying the client +// fingerprint (clientInfo name/version from initialize) and the auth verdict. +// This is the per-client visibility production debugging depends on: which +// client is failing auth, which client sees slow initializes. +// +// Pins the regression shipped with the hibernatable-DO migration (2026-06-28): +// /mcp moved off the Effect app onto the raw worker Agents bridge, whose +// `Effect.runPromise` ran the auth path under Effect's no-op default tracer — +// spans were still created, but never exported. Absence looks exactly like +// health, so the contract is asserted where the data is read (the OTLP store +// the dev stack exports to, the same exporter layer that ships prod spans to +// Axiom). +import { randomUUID } from "node:crypto"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; + +import { scenario } from "../src/scenario"; +import { Mcp, Target, Telemetry } from "../src/services"; +import type { Identity } from "../src/target"; + +const JSON_AND_SSE = "application/json, text/event-stream"; +const CLIENT_NAME = "executor-e2e-telemetry-probe"; +const CLIENT_VERSION = "9.9.9"; + +const initializeRequest = { + jsonrpc: "2.0" as const, + id: 1, + method: "initialize", + params: { + protocolVersion: "2025-03-26", + capabilities: {}, + clientInfo: { name: CLIENT_NAME, version: CLIENT_VERSION }, + }, +}; + +const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; + +const mcpPost = (url: string | URL, bearer: string | null, body: unknown): Promise => + fetch(url, { + method: "POST", + headers: { + accept: JSON_AND_SSE, + "content-type": "application/json", + ...(bearer ? { authorization: `Bearer ${bearer}` } : {}), + }, + body: JSON.stringify(body), + }); + +scenario( + "MCP telemetry · an authenticated initialize exports mcp.request with the client fingerprint", + {}, + Effect.gen(function* () { + const target = yield* Target; + const mcp = yield* Mcp; + const telemetry = yield* Telemetry; + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + + const response = yield* Effect.promise(() => mcpPost(target.mcpUrl, bearer, initializeRequest)); + yield* Effect.promise(() => response.text()); + expect(response.status, "initialize opens a session").toBe(200); + + const span = yield* telemetry.expectSpan({ + operation: "mcp.request", + attributes: { "mcp.client.name": CLIENT_NAME }, + }); + expect(span.span.tags["mcp.client.version"], "the client version is fingerprinted").toBe( + CLIENT_VERSION, + ); + expect(span.span.tags["mcp.rpc.method"], "the rpc method is visible").toBe("initialize"); + expect(span.span.tags["mcp.auth.verified"], "the auth verdict is on the span").toBe("true"); + }), +); + +scenario( + "MCP telemetry · a rejected bearer exports mcp.request with the failure fingerprint", + {}, + Effect.gen(function* () { + const target = yield* Target; + const telemetry = yield* Telemetry; + + // A garbage bearer: the 401 is the designed first step of the OAuth flow, + // and per-client 401 attribution is exactly what production debugging of + // "client X cannot log in" runs on. + const marker = `rejected-${randomUUID().slice(0, 8)}`; + const response = yield* Effect.promise(() => + mcpPost(new URL(`/mcp?probe=${marker}`, target.baseUrl), "bogus.bogus.bogus", { + ...initializeRequest, + params: { + ...initializeRequest.params, + clientInfo: { name: marker, version: CLIENT_VERSION }, + }, + }), + ); + yield* Effect.promise(() => response.text()); + expect(response.status, "the bearer is rejected").toBe(401); + + const span = yield* telemetry.expectSpan({ + operation: "mcp.request", + attributes: { "mcp.auth.has_bearer": "true", "mcp.auth.verified": "false" }, + }); + expect(span.span.tags["mcp.request.method"], "the request method is on the span").toBe("POST"); + }), +); diff --git a/packages/hosts/cloudflare/src/mcp/do-headers.test.ts b/packages/hosts/cloudflare/src/mcp/do-headers.test.ts new file mode 100644 index 000000000..ce3be391d --- /dev/null +++ b/packages/hosts/cloudflare/src/mcp/do-headers.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Tracer } from "effect"; + +import { currentPropagationHeaders } from "./do-headers"; + +const TRACE_ID = "0af7651916cd43dd8448eb211c80319c"; +const SPAN_ID = "b7ad6b7169203331"; + +describe("currentPropagationHeaders", () => { + it("emits a traceparent from an external parent span (the worker edge span)", async () => { + // The worker joins the edge http.server span via OtelTracer.withSpanContext, + // which provides an ExternalSpan (this is its effect-level shape) — not an + // Effect-local Span. Propagation must still see it, or the session DO + // starts a fresh root per request. + const request = new Request("https://executor.sh/mcp", { method: "POST" }); + const headers = await Effect.runPromise( + currentPropagationHeaders(request).pipe( + Effect.withParentSpan( + Tracer.externalSpan({ traceId: TRACE_ID, spanId: SPAN_ID, sampled: true }), + ), + ), + ); + expect(headers.traceparent).toBe(`00-${TRACE_ID}-${SPAN_ID}-01`); + }); + + it("emits a traceparent from an Effect-local span", async () => { + const request = new Request("https://executor.sh/mcp", { method: "POST" }); + const headers = await Effect.runPromise( + currentPropagationHeaders(request).pipe(Effect.withSpan("test.span")), + ); + expect(headers.traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/); + }); + + it("omits the traceparent when no span is active", async () => { + const request = new Request("https://executor.sh/mcp", { method: "POST" }); + const headers = await Effect.runPromise(currentPropagationHeaders(request)); + expect(headers.traceparent).toBeUndefined(); + }); + + it("passes through tracestate and baggage from the inbound request", async () => { + const request = new Request("https://executor.sh/mcp", { + method: "POST", + headers: { tracestate: "vendor=state", baggage: "k=v" }, + }); + const headers = await Effect.runPromise(currentPropagationHeaders(request)); + expect(headers.tracestate).toBe("vendor=state"); + expect(headers.baggage).toBe("k=v"); + }); +}); diff --git a/packages/hosts/cloudflare/src/mcp/do-headers.ts b/packages/hosts/cloudflare/src/mcp/do-headers.ts index 67d788306..c90b27393 100644 --- a/packages/hosts/cloudflare/src/mcp/do-headers.ts +++ b/packages/hosts/cloudflare/src/mcp/do-headers.ts @@ -37,8 +37,13 @@ export type IncomingPropagationHeaders = { readonly baggage?: string; }; -const currentTraceparent = Effect.map(Effect.currentSpan, (span) => { - if (!span || !span.traceId || !span.spanId) return undefined; +// `currentParentSpan`, not `currentSpan`: the worker's MCP handler joins the +// edge `http.server` span via `OtelTracer.withSpanContext`, which provides an +// ExternalSpan as the fiber's ParentSpan without opening an Effect-local span. +// `currentSpan` filters to real Spans and would come up empty there, silently +// dropping propagation and letting the DO start a fresh root per request. +const currentTraceparent = Effect.map(Effect.currentParentSpan, (span) => { + if (!span.traceId || !span.spanId) return undefined; const flags = span.sampled ? "01" : "00"; return `00-${span.traceId}-${span.spanId}-${flags}`; }).pipe(Effect.orElseSucceed(() => undefined));