diff --git a/.changeset/inbound-http-channel-trace.md b/.changeset/inbound-http-channel-trace.md new file mode 100644 index 000000000..d9058543e --- /dev/null +++ b/.changeset/inbound-http-channel-trace.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Add an opt-in OpenTelemetry `SERVER` span around each inbound HTTP channel request. Enable it with `defineInstrumentation({ traceChannelRequests: true })` — it is off by default. When enabled, the span is named for the registered route (e.g. `POST /eve/v1/session/:sessionId`), respects an incoming `traceparent` (becoming a child of the upstream span, or a trace root when there is none), and becomes the parent of nested channel and Workflow spans such as `hook.resume` and outgoing HTTP calls. It records only low-cardinality, non-sensitive attributes (`http.request.method`, `http.route` — the path template alone, per the OTel HTTP convention — `http.response.status_code`, `url.scheme`, `server.address`, `eve.channel.name`, `eve.channel.kind`) — never session ids, tokens, headers, bodies, or query parameters. This is observability-only: it does not change request handling and performs no synchronous span export in the request path, adding only minimal in-process tracing overhead. diff --git a/docs/guides/instrumentation.md b/docs/guides/instrumentation.md index 39acdf485..78305a9df 100644 --- a/docs/guides/instrumentation.md +++ b/docs/guides/instrumentation.md @@ -155,6 +155,17 @@ ai.eve.turn {eve.session.id} eve creates the `ai.eve.turn` parent span per turn and passes enriched telemetry to the AI SDK so model calls and tool executions are traced automatically. Session, turn, step, and channel context is injected as the framework half of the runtime context (`eve.version`, `eve.session.id`, `eve.environment`, `eve.turn.id`, `eve.turn.sequence`, `eve.step.index`, `eve.channel.kind`) and rides onto the spans alongside any values your `events["step.started"]` callback returns under `runtimeContext`. +Set `traceChannelRequests: true` on `defineInstrumentation` to also wrap each inbound channel HTTP request in a single OpenTelemetry `SERVER` span named for the registered route, which parents the turn tree above (and any `hook.resume` and outgoing HTTP spans): + +```text +POST /eve/v1/session/:sessionId + └── hook.resume + ├── GET hooks/by-token + └── POST hook_received +``` + +The span stays low-cardinality (route template in `http.route`, method in `http.request.method`, never the concrete URL) and records no session ids, tokens, headers, bodies, or query parameters. It adopts an incoming `traceparent` as its parent when present, so eve requests correlate with upstream traces. It defaults to `false`; enable it only when you want these request spans. + ## Workflow run tags Separately from OpenTelemetry, eve tags every workflow run with reserved `$eve.*` attributes. These live on the Vercel Workflow run, queryable in the Workflow dashboard, not on OTel spans, and you do not configure them: they are framework-owned and emitted automatically on every session, turn, and subagent run, whether or not an `instrumentation.ts` file is present. Authored code cannot set or override the `$eve.` namespace. diff --git a/packages/eve/package.json b/packages/eve/package.json index 52877a835..057080be8 100644 --- a/packages/eve/package.json +++ b/packages/eve/package.json @@ -320,6 +320,7 @@ "@chat-adapter/twilio": "4.34.0", "@clack/core": "1.3.1", "@nuxt/kit": "^4.0.0", + "@opentelemetry/context-async-hooks": "catalog:", "@opentelemetry/otlp-transformer": "0.214.0", "@opentelemetry/sdk-trace-base": "catalog:", "@standard-schema/spec": "1.1.0", diff --git a/packages/eve/scripts/vendor-compiled/declarations/@opentelemetry/api.d.ts b/packages/eve/scripts/vendor-compiled/declarations/@opentelemetry/api.d.ts index 7803d8add..4e1ddf70b 100644 --- a/packages/eve/scripts/vendor-compiled/declarations/@opentelemetry/api.d.ts +++ b/packages/eve/scripts/vendor-compiled/declarations/@opentelemetry/api.d.ts @@ -20,7 +20,11 @@ export interface Span { export interface Tracer { startSpan( name: string, - options?: { attributes?: Attributes | undefined; root?: boolean | undefined }, + options?: { + attributes?: Attributes | undefined; + kind?: SpanKind | undefined; + root?: boolean | undefined; + }, context?: Context, ): Span; } @@ -40,6 +44,20 @@ export declare const context: { with(context: Context, fn: () => T): T; }; +/** + * Reads carrier values during context extraction. `Carrier` is the + * transport-specific container (e.g. a `Headers` instance for inbound + * HTTP requests). + */ +export interface TextMapGetter { + get(carrier: Carrier, key: string): string | string[] | undefined; + keys(carrier: Carrier): string[]; +} + +export declare const propagation: { + extract(context: Context, carrier: Carrier, getter: TextMapGetter): Context; +}; + export declare const trace: { getActiveSpan(): Span | undefined; getTracer(name: string, version?: string): Tracer; diff --git a/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts b/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts index a12c9db45..3b18322e6 100644 --- a/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts +++ b/packages/eve/src/internal/nitro/routes/channel-dispatch.test.ts @@ -1,8 +1,30 @@ import type { H3Event } from "nitro"; -import { describe, expect, it, vi } from "vitest"; +import { + context as apiContext, + propagation as apiPropagation, + trace as apiTrace, +} from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { + BasicTracerProvider, + InMemorySpanExporter, + type ReadableSpan, + SimpleSpanProcessor, +} from "@opentelemetry/sdk-trace-base"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { trace as vendoredTrace } from "#compiled/@opentelemetry/api/index.js"; -import { CHANNEL_SENTINEL, type CompiledChannel } from "#channel/compiled-channel.js"; +import { + CHANNEL_SENTINEL, + type CompiledChannel, + setChannelInstrumentationKind, +} from "#channel/compiled-channel.js"; import type { RouteHandlerArgs } from "#channel/routes.js"; +import { + getInstrumentationConfig, + registerInstrumentationConfig, +} from "#harness/instrumentation-config.js"; import type { CancelTurnInput, DeliverInput, RunInput, Runtime } from "#channel/types.js"; import { readVercelProjectLink } from "#internal/vercel/project-link.js"; import type { RouteContext } from "#public/definitions/channel.js"; @@ -48,11 +70,18 @@ function createDeferred() { } function createEvent(input?: { + readonly body?: string; readonly headers?: Record; + readonly method?: string; readonly requestIp?: string; + readonly url?: string; readonly waitUntil?: (task: Promise) => void; }): H3Event { - const request = new Request("https://eve.test/slack", { headers: input?.headers }); + const request = new Request(input?.url ?? "https://eve.test/slack", { + body: input?.body, + headers: input?.headers, + method: input?.method ?? "GET", + }); Object.assign(request, { ip: input?.requestIp ?? "127.0.0.1", }); @@ -627,3 +656,406 @@ describe("dispatchChannelRequest", () => { await expect(response.text()).resolves.toBe("ok"); }); }); + +// --------------------------------------------------------------------------- +// Inbound HTTP tracing +// +// The vendored `#compiled/@opentelemetry/api` and the real `@opentelemetry/api` +// share one process-global registry (same major version), so a provider, the +// official `AsyncLocalStorageContextManager`, and a propagator registered +// through the real package are what the production code's vendored calls +// observe. That lets these unit tests assert real recording spans, +// active-context inheritance across awaits, and W3C extraction without a live +// exporter. +// --------------------------------------------------------------------------- + +/** Minimal W3C `traceparent` extractor — enough to model an inbound trace context. */ +const w3cPropagator = { + fields: () => ["traceparent"], + inject: () => {}, + extract(context: unknown, carrier: unknown, getter: { get(c: unknown, k: string): unknown }) { + const raw = getter.get(carrier, "traceparent"); + const value = Array.isArray(raw) ? raw[0] : raw; + if (typeof value !== "string") { + return context; + } + const match = /^00-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec(value); + if (match === null) { + return context; + } + return apiTrace.setSpanContext(context as never, { + isRemote: true, + spanId: match[2]!, + traceFlags: parseInt(match[3]!, 16), + traceId: match[1]!, + }); + }, +}; + +function parentSpanId(span: ReadableSpan): string | undefined { + return (span as { parentSpanContext?: { spanId?: string } }).parentSpanContext?.spanId; +} + +function slackChannel( + fetch: ResolvedChannelDefinition["fetch"], + overrides: Partial = {}, +): ResolvedChannelDefinition { + return { + fetch, + logicalPath: "agent/channels/slack.ts", + method: "POST", + name: "slack", + sourceId: "channel-slack", + sourceKind: "module", + urlPath: "/slack", + ...overrides, + } satisfies ResolvedChannelDefinition; +} + +describe("dispatchChannelRequest tracing", () => { + let exporter: InMemorySpanExporter; + let provider: BasicTracerProvider; + let priorConfig: ReturnType; + + beforeEach(() => { + exporter = new InMemorySpanExporter(); + provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + apiContext.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()); + apiPropagation.setGlobalPropagator(w3cPropagator as never); + apiTrace.setGlobalTracerProvider(provider); + // Request spans are opt-in; enable them for this suite. + priorConfig = getInstrumentationConfig(); + registerInstrumentationConfig({ traceChannelRequests: true }, { agentName: "test" }); + }); + + afterEach(() => { + registerInstrumentationConfig(priorConfig ?? {}, { agentName: "test" }); + apiTrace.disable(); + apiContext.disable(); + apiPropagation.disable(); + }); + + async function finishedSpans(): Promise { + await provider.forceFlush(); + return exporter.getFinishedSpans(); + } + + it("emits one SERVER span named for the route with method, channel, and status attributes", async () => { + let spansDuringHandler = -1; + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + slackChannel( + async () => { + spansDuringHandler = exporter.getFinishedSpans().length; + return new Response("ok"); + }, + { adapter: { kind: "channel:slack" } }, + ), + ], + runtime, + }); + + const response = await dispatchChannelRequest( + createEvent({ method: "POST", waitUntil: vi.fn() }), + "POST /slack", + {} as never, + ); + + expect(response.status).toBe(200); + // The span had not ended while the handler was still running. + expect(spansDuringHandler).toBe(0); + + const spans = await finishedSpans(); + expect(spans).toHaveLength(1); + const [span] = spans; + expect(span!.name).toBe("POST /slack"); + expect(span!.kind).toBe(1 /* SpanKind.SERVER */); + expect(span!.attributes).toMatchObject({ + "eve.channel.kind": "channel:slack", + "eve.channel.name": "slack", + "http.request.method": "POST", + "http.response.status_code": 200, + "http.route": "/slack", + "server.address": "eve.test", + "url.scheme": "https", + }); + expect(span!.status.code).toBe(0 /* SpanStatusCode.UNSET */); + }); + + it("stamps the instrumentation kind for a behaviorless authored channel, not the http adapter kind", async () => { + // Behaviorless authored channels keep adapter kind "http"; the stamped + // instrumentation kind carries the real channel:. + const definition: CompiledChannel = { + __kind: CHANNEL_SENTINEL, + routes: [], + adapter: { kind: "http" }, + }; + setChannelInstrumentationKind(definition, "channel:support"); + + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + slackChannel(async () => new Response("ok"), { + adapter: { kind: "http" }, + definition, + }), + ], + runtime, + }); + + await dispatchChannelRequest(createEvent({ waitUntil: vi.fn() }), "POST /slack", {} as never); + + const [span] = await finishedSpans(); + expect(span!.attributes["eve.channel.kind"]).toBe("channel:support"); + }); + + it("makes a span created inside the handler a child of the request span", async () => { + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + slackChannel(async () => { + // Model hook.resume: a nested span opened under the active context. + await Promise.resolve(); + vendoredTrace.getTracer("test.nested").startSpan("hook.resume").end(); + return new Response("ok"); + }), + ], + runtime, + }); + + await dispatchChannelRequest(createEvent({ waitUntil: vi.fn() }), "POST /slack", {} as never); + + const spans = await finishedSpans(); + const server = spans.find((span) => span.name === "POST /slack"); + const nested = spans.find((span) => span.name === "hook.resume"); + expect(server).toBeDefined(); + expect(nested).toBeDefined(); + expect(nested!.spanContext().traceId).toBe(server!.spanContext().traceId); + expect(parentSpanId(nested!)).toBe(server!.spanContext().spanId); + }); + + it("adopts an incoming traceparent as the request span's parent", async () => { + const traceId = "0af7651916cd43dd8448eb211c80319c"; + const parentId = "b7ad6b7169203331"; + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [slackChannel(async () => new Response("ok"))], + runtime, + }); + + await dispatchChannelRequest( + createEvent({ + headers: { traceparent: `00-${traceId}-${parentId}-01` }, + waitUntil: vi.fn(), + }), + "POST /slack", + {} as never, + ); + + const [span] = await finishedSpans(); + expect(span!.spanContext().traceId).toBe(traceId); + expect(parentSpanId(span!)).toBe(parentId); + }); + + it("marks the span an error and records the exception once for a handler that throws", async () => { + const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {}); + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + slackChannel(async () => { + throw new Error("handler exploded"); + }), + ], + runtime, + }); + + const response = await dispatchChannelRequest( + createEvent({ waitUntil: vi.fn() }), + "POST /slack", + {} as never, + ); + + // The existing JSON 500 conversion is unchanged. + expect(response.status).toBe(500); + const body = (await response.json()) as { errorId: string; ok: boolean }; + expect(body.ok).toBe(false); + expect(typeof body.errorId).toBe("string"); + + const [span] = await finishedSpans(); + expect(span!.attributes["http.response.status_code"]).toBe(500); + expect(span!.status.code).toBe(2 /* SpanStatusCode.ERROR */); + const exceptions = span!.events.filter((event) => event.name === "exception"); + expect(exceptions).toHaveLength(1); + errorSpy.mockRestore(); + }); + + it("records the exception and rethrows when runtime-bundle resolution fails", async () => { + mockedResolveNitroChannelRuntimeBundle.mockRejectedValue(new Error("bundle boom")); + + await expect( + dispatchChannelRequest(createEvent({ waitUntil: vi.fn() }), "POST /slack", {} as never), + ).rejects.toThrow("bundle boom"); + + const [span] = await finishedSpans(); + expect(span!.name).toBe("POST /slack"); + expect(span!.status.code).toBe(2 /* SpanStatusCode.ERROR */); + expect(span!.events.filter((event) => event.name === "exception")).toHaveLength(1); + // No channel was resolved, so no channel identity is attached. + expect(span!.attributes["eve.channel.name"]).toBeUndefined(); + }); + + it("still emits a 404 span with the route but no channel identity when nothing matches", async () => { + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ channels: [], runtime }); + + const response = await dispatchChannelRequest( + createEvent({ waitUntil: vi.fn() }), + "POST /missing", + {} as never, + ); + + expect(response.status).toBe(404); + + const [span] = await finishedSpans(); + expect(span!.name).toBe("POST /missing"); + expect(span!.attributes["http.response.status_code"]).toBe(404); + expect(span!.attributes["http.route"]).toBe("/missing"); + expect(span!.attributes["eve.channel.name"]).toBeUndefined(); + expect(span!.status.code).toBe(0 /* SpanStatusCode.UNSET */); + }); + + it("keeps session ids, tokens, auth headers, bodies, and query strings off the span", async () => { + const sessionId = "sess_secret123"; + const hookToken = "hooktoken_secret"; + const authSecret = "bearer_authsecret"; + const bodySecret = "body_payload_secret"; + + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [ + { + fetch: async () => new Response("ok"), + logicalPath: "agent/channels/eve.ts", + method: "POST", + name: "eve", + sourceId: "channel-eve", + sourceKind: "module", + urlPath: "/eve/v1/session/:sessionId", + } satisfies ResolvedChannelDefinition, + ], + runtime, + }); + + const event = createEvent({ + body: bodySecret, + headers: { authorization: authSecret, cookie: "eve_session=cookiesecret" }, + method: "POST", + url: `https://eve.test/eve/v1/session/${sessionId}?token=${hookToken}`, + }); + + await dispatchChannelRequest(event, "POST /eve/v1/session/:sessionId", {} as never); + + const [span] = await finishedSpans(); + expect(span!.name).toBe("POST /eve/v1/session/:sessionId"); + const serialized = JSON.stringify({ + attributes: span!.attributes, + events: span!.events, + name: span!.name, + status: span!.status, + }); + for (const secret of [sessionId, hookToken, authSecret, bodySecret, "cookiesecret"]) { + expect(serialized).not.toContain(secret); + } + }); +}); + +describe("dispatchChannelRequest without an OTel provider", () => { + let exporter: InMemorySpanExporter; + let provider: BasicTracerProvider; + let priorConfig: ReturnType; + + beforeEach(() => { + // Register an exporter to prove nothing reaches it, but leave the global + // tracer provider as the no-op default. Enable the opt-in flag so this + // exercises the no-provider path rather than the disabled-flag bypass. + exporter = new InMemorySpanExporter(); + provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + apiTrace.disable(); + apiContext.disable(); + apiPropagation.disable(); + priorConfig = getInstrumentationConfig(); + registerInstrumentationConfig({ traceChannelRequests: true }, { agentName: "test" }); + }); + + afterEach(() => { + registerInstrumentationConfig(priorConfig ?? {}, { agentName: "test" }); + apiTrace.disable(); + }); + + it("handles the request identically and records no spans", async () => { + const waitUntil = vi.fn<(task: Promise) => void>(); + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [slackChannel(async () => new Response("ok"))], + runtime, + }); + + const response = await dispatchChannelRequest( + createEvent({ waitUntil }), + "POST /slack", + {} as never, + ); + + expect(response.status).toBe(200); + await expect(response.text()).resolves.toBe("ok"); + await provider.forceFlush(); + expect(exporter.getFinishedSpans()).toHaveLength(0); + }); +}); + +describe("dispatchChannelRequest with request tracing not enabled", () => { + let exporter: InMemorySpanExporter; + let provider: BasicTracerProvider; + let priorConfig: ReturnType; + + beforeEach(() => { + // A live provider proves the opt-in flag — not a missing provider — is what + // gates the span. + exporter = new InMemorySpanExporter(); + provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + apiContext.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()); + apiPropagation.setGlobalPropagator(w3cPropagator as never); + apiTrace.setGlobalTracerProvider(provider); + priorConfig = getInstrumentationConfig(); + }); + + afterEach(() => { + // Restore the process-global config so the toggle does not leak. + registerInstrumentationConfig(priorConfig ?? {}, { agentName: "test" }); + apiTrace.disable(); + apiContext.disable(); + apiPropagation.disable(); + }); + + async function expectNoSpan(): Promise { + const waitUntil = vi.fn<(task: Promise) => void>(); + mockedResolveNitroChannelRuntimeBundle.mockResolvedValue({ + channels: [slackChannel(async () => new Response("ok"))], + runtime, + }); + + const response = await dispatchChannelRequest( + createEvent({ waitUntil }), + "POST /slack", + {} as never, + ); + + expect(response.status).toBe(200); + await expect(response.text()).resolves.toBe("ok"); + await provider.forceFlush(); + expect(exporter.getFinishedSpans()).toHaveLength(0); + } + + it("emits no request span by default when the flag is unset", async () => { + registerInstrumentationConfig({}, { agentName: "test" }); + await expectNoSpan(); + }); + + it("emits no request span when the flag is explicitly false", async () => { + registerInstrumentationConfig({ traceChannelRequests: false }, { agentName: "test" }); + await expectNoSpan(); + }); +}); diff --git a/packages/eve/src/internal/nitro/routes/channel-dispatch.ts b/packages/eve/src/internal/nitro/routes/channel-dispatch.ts index e3684ab4d..f8d3ffd3b 100644 --- a/packages/eve/src/internal/nitro/routes/channel-dispatch.ts +++ b/packages/eve/src/internal/nitro/routes/channel-dispatch.ts @@ -1,5 +1,6 @@ import type { H3Event } from "nitro"; import type { Agent, RouteContext } from "#public/definitions/channel.js"; +import { getChannelInstrumentationKind } from "#channel/compiled-channel.js"; import { createCrossChannelReceiveFn, toCrossChannelTargets, @@ -19,6 +20,7 @@ import { attachRouteAgent, } from "#internal/nitro/routes/channel-route-context.js"; import type { NitroArtifactsConfig } from "#internal/nitro/routes/runtime-artifacts.js"; +import { traceChannelRequest } from "#internal/nitro/routes/channel-request-instrumentation.js"; import { resolveNitroChannelRuntimeBundle } from "#internal/nitro/routes/runtime-stack.js"; import { readVercelProjectLink } from "#internal/vercel/project-link.js"; import { withVercelOidcProjectResolver } from "#runtime/governance/auth/vercel-oidc-project.js"; @@ -53,54 +55,74 @@ export async function dispatchChannelRequest( routeKey: string, config: NitroArtifactsConfig, ): Promise { - const bundle = await resolveNitroChannelRuntimeBundle(config); - - const matchedChannel = bundle.channels.find( - (channel) => `${channel.method.toUpperCase()} ${channel.urlPath}` === routeKey, - ); + return await traceChannelRequest({ request: event.req, routeKey }, async (span) => { + const bundle = await resolveNitroChannelRuntimeBundle(config); - if (matchedChannel === undefined) { - return Response.json( - { error: "No matching channel for this request.", ok: false }, - { status: 404 }, + const matchedChannel = bundle.channels.find( + (channel) => `${channel.method.toUpperCase()} ${channel.urlPath}` === routeKey, ); - } - const routeArgs = buildRouteArgs(event, bundle, matchedChannel.name, config); + if (matchedChannel === undefined) { + return Response.json( + { error: "No matching channel for this request.", ok: false }, + { status: 404 }, + ); + } - let response: Response; + // Channel identity is known only after resolution; a 404 span carries + // just the route. `span` is undefined when instrumentation opted out of + // channel-request tracing. Prefer the stamped instrumentation kind + // (`channel:`) over the raw adapter kind — behaviorless authored + // channels keep adapter kind `"http"`, so the adapter alone would report + // `"http"` where the rest of the trace reports `channel:`. + span?.setAttribute("eve.channel.name", matchedChannel.name); + const channelKind = + getChannelInstrumentationKind(matchedChannel.definition) ?? matchedChannel.adapter?.kind; + if (channelKind !== undefined) { + span?.setAttribute("eve.channel.kind", channelKind); + } - try { - response = await withDevelopmentVercelOidcContext(config, event.req, async () => { - if (matchedChannel.handler) { - // Authored CompiledChannel route — build RouteHandlerArgs. - return await matchedChannel.handler(event.req, routeArgs.args); - } + const routeArgs = buildRouteArgs(event, bundle, matchedChannel.name, config); - // Framework-internal fetch-only channel (e.g. the connection - // callback route). Build a RouteContext with the agent handle. - const ctx: RouteContext = { - agent: routeArgs.agent, - waitUntil: routeArgs.args.waitUntil, - params: routeArgs.args.params, - requestIp: routeArgs.args.requestIp, - }; + let response: Response; - return await matchedChannel.fetch(event.req, ctx); - }); - } catch (error) { - // Without this a handler throw is only Nitro's default 5xx, with no eve log. - const errorId = logError(log, "channel handler threw", error, { - routeKey, - channel: matchedChannel.name, - }); - flushBackgroundTasks(event, routeArgs.backgroundTasks, routeKey, matchedChannel.name); - return Response.json({ error: "Channel handler failed.", errorId, ok: false }, { status: 500 }); - } + try { + response = await withDevelopmentVercelOidcContext(config, event.req, async () => { + if (matchedChannel.handler) { + // Authored CompiledChannel route — build RouteHandlerArgs. + return await matchedChannel.handler(event.req, routeArgs.args); + } + + // Framework-internal fetch-only channel (e.g. the connection + // callback route). Build a RouteContext with the agent handle. + const ctx: RouteContext = { + agent: routeArgs.agent, + waitUntil: routeArgs.args.waitUntil, + params: routeArgs.args.params, + requestIp: routeArgs.args.requestIp, + }; - flushBackgroundTasks(event, routeArgs.backgroundTasks, routeKey, matchedChannel.name); + return await matchedChannel.fetch(event.req, ctx); + }); + } catch (error) { + // Without this a handler throw is only Nitro's default 5xx, with no eve + // log. logError records the exception against the active request span, + // so traceChannelRequest must not record the returned 500 again. + const errorId = logError(log, "channel handler threw", error, { + routeKey, + channel: matchedChannel.name, + }); + flushBackgroundTasks(event, routeArgs.backgroundTasks, routeKey, matchedChannel.name); + return Response.json( + { error: "Channel handler failed.", errorId, ok: false }, + { status: 500 }, + ); + } + + flushBackgroundTasks(event, routeArgs.backgroundTasks, routeKey, matchedChannel.name); - return response; + return response; + }); } export async function dispatchChannelWebSocketRequest( diff --git a/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts b/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts new file mode 100644 index 000000000..dbe8d3472 --- /dev/null +++ b/packages/eve/src/internal/nitro/routes/channel-request-instrumentation.ts @@ -0,0 +1,141 @@ +import { + type Attributes, + context, + propagation, + type Span, + SpanKind, + SpanStatusCode, + type TextMapGetter, + trace, +} from "#compiled/@opentelemetry/api/index.js"; +import { getInstrumentationConfig } from "#harness/instrumentation-config.js"; +import { recordErrorOnSpan } from "#internal/logging.js"; + +/** + * Stable tracer name for every inbound eve channel HTTP request. Kept + * low-cardinality and independent of the agent so dashboards can group + * request spans across deployments. + */ +const TRACER_NAME = "eve.channel"; + +/** + * Reads OTel context from an inbound request's `Headers`. Header lookups + * are case-insensitive, so this is a thin adapter over `Headers.get`. + */ +const headersGetter: TextMapGetter = { + get(carrier, key) { + return carrier.get(key) ?? undefined; + }, + keys(carrier) { + return [...carrier.keys()]; + }, +}; + +/** Inputs for {@link traceChannelRequest}. */ +export interface TraceChannelRequestInput { + readonly request: Request; + readonly routeKey: string; +} + +/** + * Wraps one inbound channel HTTP request in an OTel `SERVER` span. + * + * The span is named for the low-cardinality registered `routeKey` + * (`"POST /eve/v1/session/:sessionId"`), never the concrete URL, while the + * `http.route` attribute carries only the path template — the OTel HTTP + * semantic convention reserves `http.route` for the route template alone, + * with the method in `http.request.method`. The span is started from the + * context extracted off the incoming request headers so an upstream + * `traceparent` becomes its parent and nested channel and Workflow spans + * (`hook.resume` and outgoing HTTP) become its descendants. + * + * The handler runs inside the span's active context. When it returns, the + * response status is recorded and a `>= 500` status marks the span as an + * error. A thrown handler is recorded once and rethrown — handler + * exceptions that eve already converts to a JSON 500 return normally and + * are recorded by `logError` against this active span, so they are not + * recorded a second time here. The span always ends in `finally`, without + * waiting for `event.waitUntil()` work or streamed response bodies. + * + * Emitting these spans is opt-in: unless authored instrumentation enables it + * via `defineInstrumentation({ traceChannelRequests: true })`, the handler runs + * with no span (`undefined`) and no context extraction — a true bypass, not a + * non-recording span. + * + * This is observability-only: it never changes the response and performs no + * synchronous span export in the request path, adding only minimal in-process + * tracing overhead. + */ +export async function traceChannelRequest( + input: TraceChannelRequestInput, + handler: (span: Span | undefined) => Promise, +): Promise { + if (getInstrumentationConfig()?.traceChannelRequests !== true) { + return await handler(undefined); + } + + const { request, routeKey } = input; + const parentContext = propagation.extract(context.active(), request.headers, headersGetter); + const span = trace + .getTracer(TRACER_NAME) + .startSpan( + routeKey, + { attributes: baseAttributes(request, routeKey), kind: SpanKind.SERVER }, + parentContext, + ); + const activeContext = trace.setSpan(parentContext, span); + + try { + const response = await context.with(activeContext, () => handler(span)); + span.setAttribute("http.response.status_code", response.status); + if (response.status >= 500) { + span.setStatus({ code: SpanStatusCode.ERROR }); + } + return response; + } catch (error) { + recordErrorOnSpan(span, error); + throw error; + } finally { + span.end(); + } +} + +/** + * Standard, non-sensitive request attributes known before the channel is + * resolved. `eve.channel.name` / `eve.channel.kind` are attached later by + * the caller once the matching channel is known. Never records session + * ids, tokens, auth or cookie headers, bodies, or query parameters. + */ +function baseAttributes(request: Request, routeKey: string): Attributes { + const attributes: Attributes = { + "http.request.method": request.method, + "http.route": routeTemplate(routeKey), + }; + + const url = parseRequestUrl(request.url); + if (url !== undefined) { + attributes["url.scheme"] = url.protocol.replace(/:$/, ""); + attributes["server.address"] = url.hostname; + } + + return attributes; +} + +/** + * Extracts the path template from a `"METHOD /path/:param"` route key. The + * OTel HTTP convention keeps `http.route` free of the method so route + * facets align with other server spans; the method lives in + * `http.request.method` and the full key remains the span name. + */ +function routeTemplate(routeKey: string): string { + const separator = routeKey.indexOf(" "); + return separator === -1 ? routeKey : routeKey.slice(separator + 1); +} + +function parseRequestUrl(rawUrl: string): URL | undefined { + try { + return new URL(rawUrl); + } catch { + return undefined; + } +} diff --git a/packages/eve/src/public/instrumentation/index.ts b/packages/eve/src/public/instrumentation/index.ts index 0e73ace9e..a5d38f09d 100644 --- a/packages/eve/src/public/instrumentation/index.ts +++ b/packages/eve/src/public/instrumentation/index.ts @@ -153,6 +153,13 @@ export interface InstrumentationDefinition { * `true` when `instrumentation.ts` is present. */ readonly recordOutputs?: boolean; + /** + * Whether to emit the inbound HTTP `SERVER` span that wraps each channel + * request (the parent of the turn trace and any `hook.resume`/outgoing + * HTTP spans). Defaults to `false`. Set `true` to emit these request spans + * alongside the rest of the trace. + */ + readonly traceChannelRequests?: boolean; /** * Setup callback invoked at server startup with the resolved agent name. * Use it to call `registerOTel` or other OTel provider setup; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index eaf7c900e..6bbefd134 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -27,6 +27,9 @@ catalogs: '@ai-sdk/provider-utils': specifier: ^5.0.12 version: 5.0.13 + '@opentelemetry/context-async-hooks': + specifier: 2.6.1 + version: 2.6.1 '@opentelemetry/core': specifier: 2.6.1 version: 2.6.1 @@ -1245,6 +1248,9 @@ importers: '@nuxt/kit': specifier: ^4.0.0 version: 4.4.6(magicast@0.5.3) + '@opentelemetry/context-async-hooks': + specifier: 'catalog:' + version: 2.6.1(@opentelemetry/api@1.9.1) '@opentelemetry/otlp-transformer': specifier: 0.214.0 version: 0.214.0(@opentelemetry/api@1.9.1) @@ -3697,6 +3703,12 @@ packages: peerDependencies: '@opentelemetry/api': '>=1.0.0 <1.10.0' + '@opentelemetry/context-async-hooks@2.6.1': + resolution: {integrity: sha512-XHzhwRNkBpeP8Fs/qjGrAf9r9PRv67wkJQ/7ZPaBQQ68DYlTBBx5MF9LvPx7mhuXcDessKK2b+DcxqwpgkcivQ==} + engines: {node: ^18.19.0 || >=20.6.0} + peerDependencies: + '@opentelemetry/api': '>=1.0.0 <1.10.0' + '@opentelemetry/core@2.10.0': resolution: {integrity: sha512-/wNZ8twnEQQA4HoHu22+vcsdru6pWPWxW+7w+FlxT6Id7PE/WIbZmVKkte+PF72e0F2dnImFeHD2syyE1Mw6MQ==} engines: {node: ^18.19.0 || >=20.6.0} @@ -17757,6 +17769,10 @@ snapshots: dependencies: '@opentelemetry/api': 1.9.1 + '@opentelemetry/context-async-hooks@2.6.1(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@opentelemetry/core@2.10.0(@opentelemetry/api@1.9.1)': dependencies: '@opentelemetry/api': 1.9.1 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 707282ee8..42ad44eb8 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -31,6 +31,7 @@ catalog: "@ai-sdk/otel": "^1.0.34" "@ai-sdk/provider": "^4.0.3" "@ai-sdk/provider-utils": "^5.0.12" + "@opentelemetry/context-async-hooks": "2.6.1" "@opentelemetry/core": "2.6.1" "@opentelemetry/sdk-trace-base": "2.6.1" "@types/node": "25.9.1"