History

Restore MCP request span export on the cloud worker (#1410)

The Agents-bridge migration moved /mcp off the Effect app, so the auth path's Effect programs ran under the default no-op tracer and the mcp.request span (client fingerprint, rpc method, auth verdict) was created but never exported. Run those programs under the worker telemetry layer, open the http.server envelope span for MCP traffic at the worker edge, and propagate its context through the handler into the session DO so the whole request is one trace. currentPropagationHeaders now reads currentParentSpan instead of currentSpan so an external (edge) span still propagates into the DO. The e2e scenario asserts the contract against the exported store.

Rhys Sullivan committed 4 months ago

Commit 13732fe1 parent 0a50c797 files · +278 −53Browse files

Unified · Split

7 files
  1. apps/cloud/src/mcp/agent-handler.ts +27 −2
  2. apps/cloud/src/mcp/session-durable-object.ts +1 −29
  3. apps/cloud/src/mcp/traceparent.ts +32 −0
  4. apps/cloud/src/server.ts +56 −20
  5. e2e/cloud/mcp-telemetry-contract.test.ts +106 −0
  6. packages/hosts/cloudflare/src/mcp/do-headers.test.ts +49 −0
  7. packages/hosts/cloudflare/src/mcp/do-headers.ts +7 −2
apps/cloud/src/mcp/agent-handler.ts+27−2
@@ -1,3 +1,4 @@
1+import * as OtelTracer from "@effect/opentelemetry/Tracer";
12 import { Effect, Predicate } from "effect";
23
34 import {
@@ -17,8 +18,10 @@ import type { McpSessionProps } from "@executor-js/cloudflare/mcp/agent-durable- · 13 unmodified lines
1718 import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub";
1819
1920 import { wrapMcpSseResponse } from "../observability/memory-metrics";
21+import { WorkerTelemetryLive } from "../observability/telemetry";
2022 import { cloudMcpAuth } from "./auth-provider";
2123 import { McpSessionDOSqlite } from "./session-durable-object";
24+import { parseTraceparent } from "./traceparent";
2225
2326 const corsPreflightResponse = (): Response =>
2427 new Response(null, {
@@ -82,6 +85,28 @@ const authenticate = (request: Request) => · 57 unmodified lines
8285 return { auth, outcome };
8386 }).pipe(Effect.provide(cloudMcpAuth));
8487
88+// The pre-Agents envelope ran the MCP auth path inside the Effect app, whose
89+// HttpMiddleware provided the OTEL tracer — that is where the `mcp.request`
90+// span (client fingerprint, rpc method, auth outcome) exported from. This
91+// handler dispatches from the raw worker entry instead, so a bare
92+// `Effect.runPromise` leaves every span in Effect's no-op default tracer and
93+// they silently never export. Run each program under the worker telemetry
94+// layer, parented to the edge `http.server` span server.ts stamps onto the
95+// forwarded request's traceparent — which also makes `currentPropagationHeaders`
96+// (via Effect.currentParentSpan) ferry that same trace into the session DO
97+// instead of letting the DO start a fresh root per request.
98+const runTraced = <A>(request: Request, program: Effect.Effect<A>): Promise<A> => {
99+ const parsed = parseTraceparent(
100+ request.headers.get("traceparent"),
101+ request.headers.get("tracestate"),
102+ );
103+ return Effect.runPromise(
104+ (parsed ? OtelTracer.withSpanContext(program, parsed) : program).pipe(
105+ Effect.provide(WorkerTelemetryLive),
106+ ),
107+ );
108+};
109+
85110 // The MCP resource the request targets. `server.ts` routes both the bare `/mcp`
86111 // and `/mcp/toolkits/<slug>` to this handler (`prepareMcpOrgScope` strips the org
87112 // selector but keeps the toolkit segment), so a session minted on a toolkit path
@@ -142,7 +167,7 @@ export const makeCloudMcpAgentHandler = () => { · 54 unmodified lines
142167 }
143168 const sessionId = request.headers.get("mcp-session-id");
144169
145− const { auth, outcome } = await Effect.runPromise(authenticate(request));
170+ const { auth, outcome } = await runTraced(request, authenticate(request));
146171 if (!Predicate.isTagged(outcome, "Authenticated")) {
147172 // Destroying a live session on auth grounds requires a POSITIVE
148173 // determination that access is genuinely gone — only `Forbidden` carries
@@ -191,7 +216,7 @@ export const makeCloudMcpAgentHandler = () => { · 42 unmodified lines
191216 }
192217
193218 const resource = resourceFromPath(request);
194− const props = await Effect.runPromise(propsForPrincipal(request, outcome.principal, resource));
219+ const props = await runTraced(request, propsForPrincipal(request, outcome.principal, resource));
195220 (ctx as ExecutionContext & { props?: McpSessionProps }).props = props;
196221 const forwarded = withVerifiedIdentityHeaders(
197222 request,
apps/cloud/src/mcp/session-durable-object.ts+1−29
@@ -14,7 +14,6 @@ · 13 unmodified lines
1414 // ---------------------------------------------------------------------------
1515
1616 import { env } from "cloudflare:workers";
17−import { createTraceState } from "@opentelemetry/api";
1817 import { Data, Effect, Layer } from "effect";
1918 import type { Cause } from "effect";
2019 import * as OtelTracer from "@effect/opentelemetry/Tracer";
@@ -73,6 +72,7 @@ import { · 52 unmodified lines
7372 captureCauseEffect as reportCauseEffect,
7473 tagCurrentSentryScopeWithCurrentOtelSpan,
7574 } from "../observability";
75+import { parseTraceparent } from "./traceparent";
7676
7777 // Re-export the shared types so existing cloud importers
7878 // (`auth/handlers.ts`, etc.) keep their `../mcp/session-durable-object` path.
@@ -113,34 +113,6 @@ class McpModelResumeForwardError extends Data.TaggedError("McpModelResumeForward · 34 unmodified lines
113113 readonly cause: unknown;
114114 }> {}
115115
116−// W3C propagation across the worker→DO boundary. The worker injects its
117−// `traceparent` and forwards incoming `tracestate` / `baggage`; we parse the
118−// context and use `OtelTracer.withSpanContext` to stitch the DO's root span
119−// under the worker span so the entire logical request lives in one trace.
120−const TRACEPARENT_PATTERN = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/;
121−
122−type IncomingSpanContext = {
123− readonly traceId: string;
124− readonly spanId: string;
125− readonly traceFlags: number;
126− readonly traceState?: ReturnType<typeof createTraceState>;
127−};
128−
129−const parseTraceparent = (
130− traceparent: string | null | undefined,
131− tracestate: string | null | undefined,
132−): IncomingSpanContext | null => {
133− if (!traceparent) return null;
134− const match = TRACEPARENT_PATTERN.exec(traceparent);
135− if (!match) return null;
136− return {
137− traceId: match[2]!,
138− spanId: match[3]!,
139− traceFlags: parseInt(match[4]!, 16),
140− ...(tracestate ? { traceState: createTraceState(tracestate) } : {}),
141− };
142−};
143−
144116 /**
145117 * The DO keeps one postgres.js client for the MCP session runtime. postgres.js
146118 * closes idle sockets quickly, while the runtime object stays alive so the MCP
apps/cloud/src/mcp/traceparent.ts+32−0
@@ -0,0 +1,32 @@
1+// ---------------------------------------------------------------------------
2+// W3C traceparent parsing shared by the worker edge (server.ts), the MCP agent
3+// handler, and the session DO. Single-sourced so the producer (the worker span
4+// stamping traceparent onto the forwarded request) and the consumers (the
5+// Effect programs joining that span) cannot drift on the header grammar.
6+// ---------------------------------------------------------------------------
7+
8+import { createTraceState } from "@opentelemetry/api";
9+
10+const TRACEPARENT_PATTERN = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/;
11+
12+export type IncomingSpanContext = {
13+ readonly traceId: string;
14+ readonly spanId: string;
15+ readonly traceFlags: number;
16+ readonly traceState?: ReturnType<typeof createTraceState>;
17+};
18+
19+export const parseTraceparent = (
20+ traceparent: string | null | undefined,
21+ tracestate: string | null | undefined,
22+): IncomingSpanContext | null => {
23+ if (!traceparent) return null;
24+ const match = TRACEPARENT_PATTERN.exec(traceparent);
25+ if (!match) return null;
26+ return {
27+ traceId: match[2]!,
28+ spanId: match[3]!,
29+ traceFlags: parseInt(match[4]!, 16),
30+ ...(tracestate ? { traceState: createTraceState(tracestate) } : {}),
31+ };
32+};
apps/cloud/src/server.ts+56−20
@@ -14,6 +14,7 @@ import handler from "@tanstack/react-start/server-entry"; · 13 unmodified lines
1414 import { isAppOwnedPath } from "./app-paths";
1515 import { makeCloudMcpAgentHandler } from "./mcp/agent-handler";
1616 import { classifyMcpPath, prepareMcpOrgScope } from "./mcp/mount";
17+import { parseTraceparent } from "./mcp/traceparent";
1718 import { McpSessionDOSqlite as McpSessionDOBase } from "./mcp/session-durable-object";
1819 import {
1920 beforeSendWithOtelCorrelation,
@@ -124,16 +125,66 @@ const cloudflareHandler: ExportedHandler<Env> = { · 104 unmodified lines
124125 const url = new URL(request.url);
125126 const mcpRoute = classifyMcpPath(url.pathname);
126127 const tracingInstalled = installTracerProvider();
128+ // Join the caller's W3C trace when the request carries one — the web UI
129+ // sends traceparent on every API fetch, so the browser's spans and this
130+ // request share one trace id end to end. Same parsing the DO path does
131+ // in session-durable-object.ts.
132+ const inbound = parseTraceparent(request.headers.get("traceparent"), null);
133+ const parentContext = inbound
134+ ? trace.setSpanContext(context.active(), {
135+ traceId: inbound.traceId,
136+ spanId: inbound.spanId,
137+ traceFlags: inbound.traceFlags,
138+ isRemote: true,
139+ })
140+ : context.active();
127141 if (mcpRoute?.kind === "mcp") {
128142 // The Cloudflare Agents MCP bridge needs the platform ExecutionContext
129143 // to pass authenticated session props into the hibernatable DO.
130144 // Discovery docs still flow through the app-level MCP envelope.
131− // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary; keep trace export alive after the Agents bridge resolves or rejects
132− try {
133− return await mcpAgentHandler(prepareMcpOrgScope(request), env, ctx);
134− } finally {
135− if (tracingInstalled) ctx.waitUntil(flushTracerProvider());
145+ const forwarded = prepareMcpOrgScope(request);
146+ if (!tracingInstalled) {
147+ return mcpAgentHandler(forwarded, env, ctx);
136148 }
149+ // /mcp left the Effect app in the Agents-bridge migration, so no
150+ // downstream HttpMiddleware.tracer opens the request envelope anymore —
151+ // this worker span is now THE `http.server` span for MCP traffic. Its
152+ // context is stamped onto the forwarded request's traceparent so the
153+ // agent handler's Effect programs (mcp.request and children) and the
154+ // session DO parent under it instead of exporting orphaned roots.
155+ return tracer.startActiveSpan(
156+ `http.server ${request.method}`,
157+ { kind: SpanKind.SERVER },
158+ parentContext,
159+ async (span) => {
160+ span.setAttribute(ATTR_HTTP_REQUEST_METHOD, request.method);
161+ span.setAttribute(ATTR_URL_FULL, request.url);
162+ span.setAttribute(ATTR_URL_PATH, url.pathname);
163+ span.setAttribute(ATTR_URL_SCHEME, url.protocol.replace(/:$/, ""));
164+ const spanContext = span.spanContext();
165+ const headers = new Headers(forwarded.headers);
166+ headers.set(
167+ "traceparent",
168+ `00-${spanContext.traceId}-${spanContext.spanId}-${(spanContext.traceFlags & 0xff).toString(16).padStart(2, "0")}`,
169+ );
170+ // 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
171+ try {
172+ const response = await mcpAgentHandler(new Request(forwarded, { headers }), env, ctx);
173+ span.setAttribute(ATTR_HTTP_RESPONSE_STATUS_CODE, response.status);
174+ if (response.status >= 500) {
175+ span.setStatus({ code: SpanStatusCode.ERROR });
176+ }
177+ return response;
178+ } catch (err) {
179+ span.setStatus({ code: SpanStatusCode.ERROR });
180+ // oxlint-disable-next-line executor/no-try-catch-or-throw -- adapter boundary; preserve original error to Cloudflare runtime
181+ throw err;
182+ } finally {
183+ span.end();
184+ ctx.waitUntil(flushTracerProvider());
185+ }
186+ },
187+ );
137188 }
138189 if (!tracingInstalled) {
139190 return fetchHandler(request, env, ctx);
@@ -150,21 +201,6 @@ const cloudflareHandler: ExportedHandler<Env> = { · 10 unmodified lines
150201 ctx.waitUntil(flushTracerProvider());
151202 }
152203 }
153− // Join the caller's W3C trace when the request carries one — the web UI
154− // sends traceparent on every API fetch, so the browser's spans and this
155− // request share one trace id end to end. Same parsing the DO path does
156− // in session-durable-object.ts.
157− const traceparentMatch = /^[0-9a-f]{2}-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec(
158− request.headers.get("traceparent") ?? "",
159− );
160− const parentContext = traceparentMatch
161− ? trace.setSpanContext(context.active(), {
162− traceId: traceparentMatch[1]!,
163− spanId: traceparentMatch[2]!,
164− traceFlags: parseInt(traceparentMatch[3]!, 16),
165− isRemote: true,
166− })
167− : context.active();
168204 return tracer.startActiveSpan(
169205 `http.server ${request.method}`,
170206 { kind: SpanKind.SERVER },
e2e/cloud/mcp-telemetry-contract.test.ts+106−0
@@ -0,0 +1,106 @@
1+// Cloud: the /mcp observability contract. Every MCP request must land in the
2+// EXPORTED trace store as an `mcp.request` span carrying the client
3+// fingerprint (clientInfo name/version from initialize) and the auth verdict.
4+// This is the per-client visibility production debugging depends on: which
5+// client is failing auth, which client sees slow initializes.
6+//
7+// Pins the regression shipped with the hibernatable-DO migration (2026-06-28):
8+// /mcp moved off the Effect app onto the raw worker Agents bridge, whose
9+// `Effect.runPromise` ran the auth path under Effect's no-op default tracer —
10+// spans were still created, but never exported. Absence looks exactly like
11+// health, so the contract is asserted where the data is read (the OTLP store
12+// the dev stack exports to, the same exporter layer that ships prod spans to
13+// Axiom).
14+import { randomUUID } from "node:crypto";
15+
16+import { expect } from "@effect/vitest";
17+import { Effect } from "effect";
18+
19+import { scenario } from "../src/scenario";
20+import { Mcp, Target, Telemetry } from "../src/services";
21+import type { Identity } from "../src/target";
22+
23+const JSON_AND_SSE = "application/json, text/event-stream";
24+const CLIENT_NAME = "executor-e2e-telemetry-probe";
25+const CLIENT_VERSION = "9.9.9";
26+
27+const initializeRequest = {
28+ jsonrpc: "2.0" as const,
29+ id: 1,
30+ method: "initialize",
31+ params: {
32+ protocolVersion: "2025-03-26",
33+ capabilities: {},
34+ clientInfo: { name: CLIENT_NAME, version: CLIENT_VERSION },
35+ },
36+};
37+
38+const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label;
39+
40+const mcpPost = (url: string | URL, bearer: string | null, body: unknown): Promise<Response> =>
41+ fetch(url, {
42+ method: "POST",
43+ headers: {
44+ accept: JSON_AND_SSE,
45+ "content-type": "application/json",
46+ ...(bearer ? { authorization: `Bearer ${bearer}` } : {}),
47+ },
48+ body: JSON.stringify(body),
49+ });
50+
51+scenario(
52+ "MCP telemetry · an authenticated initialize exports mcp.request with the client fingerprint",
53+ {},
54+ Effect.gen(function* () {
55+ const target = yield* Target;
56+ const mcp = yield* Mcp;
57+ const telemetry = yield* Telemetry;
58+ const identity = yield* target.newIdentity();
59+ const bearer = yield* mcp.mintBearer(emailOf(identity));
60+
61+ const response = yield* Effect.promise(() => mcpPost(target.mcpUrl, bearer, initializeRequest));
62+ yield* Effect.promise(() => response.text());
63+ expect(response.status, "initialize opens a session").toBe(200);
64+
65+ const span = yield* telemetry.expectSpan({
66+ operation: "mcp.request",
67+ attributes: { "mcp.client.name": CLIENT_NAME },
68+ });
69+ expect(span.span.tags["mcp.client.version"], "the client version is fingerprinted").toBe(
70+ CLIENT_VERSION,
71+ );
72+ expect(span.span.tags["mcp.rpc.method"], "the rpc method is visible").toBe("initialize");
73+ expect(span.span.tags["mcp.auth.verified"], "the auth verdict is on the span").toBe("true");
74+ }),
75+);
76+
77+scenario(
78+ "MCP telemetry · a rejected bearer exports mcp.request with the failure fingerprint",
79+ {},
80+ Effect.gen(function* () {
81+ const target = yield* Target;
82+ const telemetry = yield* Telemetry;
83+
84+ // A garbage bearer: the 401 is the designed first step of the OAuth flow,
85+ // and per-client 401 attribution is exactly what production debugging of
86+ // "client X cannot log in" runs on.
87+ const marker = `rejected-${randomUUID().slice(0, 8)}`;
88+ const response = yield* Effect.promise(() =>
89+ mcpPost(new URL(`/mcp?probe=${marker}`, target.baseUrl), "bogus.bogus.bogus", {
90+ ...initializeRequest,
91+ params: {
92+ ...initializeRequest.params,
93+ clientInfo: { name: marker, version: CLIENT_VERSION },
94+ },
95+ }),
96+ );
97+ yield* Effect.promise(() => response.text());
98+ expect(response.status, "the bearer is rejected").toBe(401);
99+
100+ const span = yield* telemetry.expectSpan({
101+ operation: "mcp.request",
102+ attributes: { "mcp.auth.has_bearer": "true", "mcp.auth.verified": "false" },
103+ });
104+ expect(span.span.tags["mcp.request.method"], "the request method is on the span").toBe("POST");
105+ }),
106+);
packages/hosts/cloudflare/src/mcp/do-headers.test.ts+49−0
@@ -0,0 +1,49 @@
1+import { describe, expect, it } from "@effect/vitest";
2+import { Effect, Tracer } from "effect";
3+
4+import { currentPropagationHeaders } from "./do-headers";
5+
6+const TRACE_ID = "0af7651916cd43dd8448eb211c80319c";
7+const SPAN_ID = "b7ad6b7169203331";
8+
9+describe("currentPropagationHeaders", () => {
10+ it("emits a traceparent from an external parent span (the worker edge span)", async () => {
11+ // The worker joins the edge http.server span via OtelTracer.withSpanContext,
12+ // which provides an ExternalSpan (this is its effect-level shape) — not an
13+ // Effect-local Span. Propagation must still see it, or the session DO
14+ // starts a fresh root per request.
15+ const request = new Request("https://executor.sh/mcp", { method: "POST" });
16+ const headers = await Effect.runPromise(
17+ currentPropagationHeaders(request).pipe(
18+ Effect.withParentSpan(
19+ Tracer.externalSpan({ traceId: TRACE_ID, spanId: SPAN_ID, sampled: true }),
20+ ),
21+ ),
22+ );
23+ expect(headers.traceparent).toBe(`00-${TRACE_ID}-${SPAN_ID}-01`);
24+ });
25+
26+ it("emits a traceparent from an Effect-local span", async () => {
27+ const request = new Request("https://executor.sh/mcp", { method: "POST" });
28+ const headers = await Effect.runPromise(
29+ currentPropagationHeaders(request).pipe(Effect.withSpan("test.span")),
30+ );
31+ expect(headers.traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-0[01]$/);
32+ });
33+
34+ it("omits the traceparent when no span is active", async () => {
35+ const request = new Request("https://executor.sh/mcp", { method: "POST" });
36+ const headers = await Effect.runPromise(currentPropagationHeaders(request));
37+ expect(headers.traceparent).toBeUndefined();
38+ });
39+
40+ it("passes through tracestate and baggage from the inbound request", async () => {
41+ const request = new Request("https://executor.sh/mcp", {
42+ method: "POST",
43+ headers: { tracestate: "vendor=state", baggage: "k=v" },
44+ });
45+ const headers = await Effect.runPromise(currentPropagationHeaders(request));
46+ expect(headers.tracestate).toBe("vendor=state");
47+ expect(headers.baggage).toBe("k=v");
48+ });
49+});
packages/hosts/cloudflare/src/mcp/do-headers.ts+7−2
@@ -37,8 +37,13 @@ export type IncomingPropagationHeaders = { · 36 unmodified lines
3737 readonly baggage?: string;
3838 };
3939
40−const currentTraceparent = Effect.map(Effect.currentSpan, (span) => {
41− if (!span || !span.traceId || !span.spanId) return undefined;
40+// `currentParentSpan`, not `currentSpan`: the worker's MCP handler joins the
41+// edge `http.server` span via `OtelTracer.withSpanContext`, which provides an
42+// ExternalSpan as the fiber's ParentSpan without opening an Effect-local span.
43+// `currentSpan` filters to real Spans and would come up empty there, silently
44+// dropping propagation and letting the DO start a fresh root per request.
45+const currentTraceparent = Effect.map(Effect.currentParentSpan, (span) => {
46+ if (!span.traceId || !span.spanId) return undefined;
4247 const flags = span.sampled ? "01" : "00";
4348 return `00-${span.traceId}-${span.spanId}-${flags}`;
4449 }).pipe(Effect.orElseSucceed(() => undefined));