From 39d71fd33ec8c0e5cb3b8aacda5df5af0deec94b Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Thu, 13 Aug 2026 01:04:55 +0000 Subject: [PATCH 1/4] feat(world-vercel): synthesize per-event client spans on the WS transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #3084 added the opt-in `WORKFLOW_EVENTS_TRANSPORT=ws` path and listed "no client-side span on the WS path" as a known limitation. Because event writes become multiplexed frames on one long-lived socket rather than individual `fetch` calls, the per-event `http POST` CLIENT span that the HTTP transport produced simply disappeared — traces went from one span per event to nothing between the invocation and the server. Restore it by synthesizing a request-shaped span around each frame, and give the upgrade its own span: - Extract `withHttpClientSpan` / `recordClientSpanStatus` from `instrumentedFetch` in `http-core.ts` so the synthetic span is emitted by the same envelope as the real one and cannot drift from it. `InstrumentedFetchOptions` now extends `HttpClientSpanOptions`. - `postEventFrameOverWs` opens `http POST` with `url.full` pointing at the v4 REST endpoint the frame is forwarded into, so per-event traces and latency dashboards keep working across the flag. Extract `eventsV4Url` so that URL cannot drift from the one the HTTP path actually requests. - Tag both transports with `workflow.events.transport` (`http` | `ws`) and `workflow.event.type`; the WS path additionally sets `network.protocol.name=websocket`, `workflow.events.ws.url` (the real wire destination) and `workflow.events.ws.req_id` (join key to the server's log line for the frame), so the span is never mistaken for a real HTTP request. - Add a `workflow.events.ws.connect` span around the upgrade — the one genuinely-HTTP request here, previously the invisible half of every WS write's latency — carrying `workflow.events.ws.reconnect_attempt`. This also puts `resolveUpgradeHeaders`' trace-context injection inside a client span, as AGENTS.md requires. - Fix `parseServer` to treat `wss:` as TLS (port 443, not 80). Out of scope, deliberately: per-frame `traceparent` (needs a frame-meta field plus a server change) and Vercel's outgoing-requests view (that instruments global `fetch`, so a frame structurally cannot appear there). Covered by `ws-transport-spans.test.ts`, which drives the real selection + transport + adapter stack over a fake socket and asserts span shape, failure reporting, retry behaviour and HTTP/WS parity. Co-Authored-By: Claude Opus 5 Co-Authored-By: shalabhchaturvedi-7802 Co-Authored-By: shalabhc --- .changeset/ws-transport-synthetic-spans.md | 5 + AGENTS.md | 2 + docs/content/worlds/v5/vercel.mdx | 12 + packages/world-vercel/src/events-v4.ts | 212 ++++++-- packages/world-vercel/src/http-core.ts | 162 ++++-- packages/world-vercel/src/telemetry.ts | 53 ++ .../src/trace-propagation.test.ts | 24 +- .../src/ws-transport-spans.test.ts | 481 ++++++++++++++++++ packages/world-vercel/src/ws-transport.ts | 56 +- 9 files changed, 901 insertions(+), 106 deletions(-) create mode 100644 .changeset/ws-transport-synthetic-spans.md create mode 100644 packages/world-vercel/src/ws-transport-spans.test.ts diff --git a/.changeset/ws-transport-synthetic-spans.md b/.changeset/ws-transport-synthetic-spans.md new file mode 100644 index 0000000000..208560b73f --- /dev/null +++ b/.changeset/ws-transport-synthetic-spans.md @@ -0,0 +1,5 @@ +--- +'@workflow/world-vercel': patch +--- + +Restore the per-event client span on the WebSocket events transport (`WORKFLOW_EVENTS_TRANSPORT=ws`). Each event write now emits a synthesized `http POST` CLIENT span with the same name, kind and `url.full` as the HTTP path — plus `workflow.events.transport`, `network.protocol.name`, `workflow.events.ws.url` and `workflow.events.ws.req_id` so a synthetic span is never mistaken for a real request. The WebSocket handshake gets its own `workflow.events.ws.connect` span, and injects trace context from inside it. The HTTP path is unchanged apart from gaining `workflow.events.transport: 'http'` and `workflow.event.type`. diff --git a/AGENTS.md b/AGENTS.md index 83324adbc3..3a2ab02db5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -409,3 +409,5 @@ The `executionContext` field on workflow runs is a flexible JSONB/CBOR object th Every outgoing HTTP request from `@workflow/world-vercel` to workflow-server (or the queue) MUST explicitly inject W3C trace context so the server can parent its spans to the caller and traces stay correlated end to end. Call `injectTraceContextIntoHeaders(headers)` (from `packages/world-vercel/src/telemetry.ts`) on the outgoing headers, inside the client span when one exists — `makeRequest` in `utils.ts` is the reference implementation. It is a no-op when no OpenTelemetry SDK is registered. Do **not** rely on ambient OpenTelemetry auto-instrumentation to do this: world-vercel's request paths use custom undici dispatchers / `global fetch`, which auto-instrumentation does not reliably hook. When you add a new request path or API version (e.g. a future v5 events API), wire the injection in the same place you build the request headers. The v4 events path (`fetchV4` in `events-v4.ts`) regressed cross-service correlation precisely by routing around `makeRequest` and skipping this step — workflow-server spans stopped joining the flow-route invocation trace until the injection was added back. Cover new paths with a test in `trace-propagation.test.ts`. + +The same rule covers a request path that is not an HTTP request. A non-`fetch` transport must still open the client span callers read a trace through: use `withHttpClientSpan` (`http-core.ts`), the envelope `instrumentedFetch` is built on, so the span carries the same name, kind and attributes rather than a hand-rolled parallel shape. The WS events transport is the worked example — `postEventFrameOverWs` synthesizes an `http POST` span per frame and tags it `workflow.events.transport: 'ws'`, and the handshake gets its own `workflow.events.ws.connect` span (`ws-transport-spans.test.ts`). Adding a transport that writes events without one silently deletes the per-event view of a run. diff --git a/docs/content/worlds/v5/vercel.mdx b/docs/content/worlds/v5/vercel.mdx index 2c2570c26b..ac96611222 100644 --- a/docs/content/worlds/v5/vercel.mdx +++ b/docs/content/worlds/v5/vercel.mdx @@ -202,6 +202,18 @@ Experimental. Set `WORKFLOW_EVENTS_TRANSPORT=ws` to ship workflow run events to The setting is ignored when the World is configured with `projectConfig` and therefore routes through the `api-workflow` proxy: that endpoint is an HTTP-only REST gateway and does not forward a WebSocket upgrade, so events stay on HTTP and a warning is logged once per process. +Tracing is unaffected by the choice. Each event write emits an `http POST` client span whichever transport carries it, against the same `url.full` — on the WebSocket path that span is synthesized around the frame, since no HTTP request is made. Attributes tell the two apart: + +| Attribute | HTTP | WebSocket | +| --- | --- | --- | +| `workflow.events.transport` | `http` | `ws` | +| `workflow.event.type` | the event type, e.g. `step_started` | same | +| `network.protocol.name` | — | `websocket` | +| `workflow.events.ws.url` | — | the socket the frame went over | +| `workflow.events.ws.req_id` | — | per-connection request id, matching the server's log line | + +The WebSocket handshake is itself a span, `workflow.events.ws.connect`, so the cost of opening (or eagerly reopening) a connection is attributable rather than showing up as unexplained time inside the first write. + ### Programmatic configuration `createWorld()` accepts explicit API configuration. It does not read `WORKFLOW_VERCEL_*` automatically, so pass the environment values yourself when you want a configured World module: diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index e8de126e6b..09587c11c4 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -51,12 +51,23 @@ import { import { errorForResponse, headersToRecord, + httpLog, instrumentedFetch, parseRetryAfter, + recordClientSpanStatus, + withHttpClientSpan, } from './http-core.js'; import { hasSerializedDataFormatPrefix } from './serialized-data.js'; import { deserializeStep, StepWireSchema } from './steps.js'; -import { type APIConfig, getHttpConfig } from './utils.js'; +import { + ErrorType, + NetworkProtocolName, + WorkflowEventsTransport, + WorkflowEventType, + WorkflowWsRequestId, + WorkflowWsUrl, +} from './telemetry.js'; +import { type APIConfig, getHttpConfig, getHttpUrl } from './utils.js'; import type { WsFrameReply } from './ws-transport.js'; import { isWsEventsTransportEnabled } from './ws-transport-enabled.js'; @@ -87,7 +98,8 @@ async function fetchV4( url: string, init: { method: string; headers: Headers; body?: Uint8Array }, config: APIConfig | undefined, - opName: string + opName: string, + attributes?: Record ): Promise { const dispatcher = getEventsDispatcher(config); return instrumentedFetch({ @@ -96,6 +108,10 @@ async function fetchV4( headers: init.headers, body: init.body, dispatcher, + // Named on both transports so a trace or a latency dashboard can tell which + // one served a write — they are otherwise deliberately indistinguishable, + // right down to the span name and `url.full`. See `postEventFrameOverWs`. + attributes: { ...WorkflowEventsTransport('http'), ...attributes }, // Repeated transport failures retire the shared events pool and the next // request builds a fresh one. undici keeps a black-holed HTTP/2 session in // service indefinitely, so without this every request routed onto it fails @@ -120,6 +136,23 @@ async function fetchV4( const EVENT_ID_HEADER = 'x-wf-event-id'; const MAX_EVENTS_HEADER = 'x-wf-max-events'; +/** + * The v4 endpoint one event write targets. + * + * Shared with the WS path, which never requests it but reports it as the + * `url.full` of its synthetic client span: the server forwards a frame into + * this exact route, so naming it is what lets a trace or a dashboard compare + * the two transports write-for-write. Drift between the two would silently + * split that comparison in half. + */ +function eventsV4Url( + baseUrl: string, + runId: string, + eventType: string +): string { + return `${baseUrl}/v4/runs/${encodeURIComponent(runId)}/events/${encodeURIComponent(eventType)}`; +} + interface CreateEventV4InputBase { // runId is required even for run_created, because the payload is keyed under the runId runId: string; @@ -670,12 +703,13 @@ async function postWorkflowRunEventV4( input.payload ?? new Uint8Array(0) ); - const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/${encodeURIComponent(input.eventType)}`; + const url = eventsV4Url(baseUrl, input.runId, input.eventType); return fetchV4( url, { method: 'POST', headers, body: frame }, config, - 'createEvent' + 'createEvent', + WorkflowEventType(input.eventType) ); } @@ -824,6 +858,43 @@ function wsReplyStatus(reply: WsFrameReply, endpoint: string): number { return status; } +/** + * Synthesize the per-write client span the WS path would otherwise not have. + * + * On HTTP every event write goes through `fetchV4` → `instrumentedFetch`, which + * opens an `http POST` CLIENT span, times it, and stamps the response status on + * it. A frame multiplexed onto a shared socket makes no `fetch` call and + * produces no `Response`, so that span simply disappeared when the transport + * flipped — and with it the per-event view of a run's writes, which is the + * thing a trace of a step execution is mostly made of. + * + * Nothing about the request/response *semantics* changed, though: one frame out, + * one correlated reply back, one status. So the span is synthesized here with + * the same name, kind and attributes the fetch path emits, over the v4 REST + * endpoint the server forwards the frame into. Two consequences that are the + * point rather than a side effect: + * + * - a trace looks the same either side of `WORKFLOW_EVENTS_TRANSPORT`, so the + * A/B the flag exists for compares like with like, and + * - dashboards keyed on `http POST` + `url.full` keep working unchanged. + * + * What is *not* elided: `workflow.events.transport: 'ws'`, + * `network.protocol.name: 'websocket'` and `workflow.events.ws.url` say plainly + * that no HTTP request was issued, and `workflow.events.ws.req_id` is the join + * key to the server's log line for the same frame. A synthetic span that hid + * which transport produced it would be a trap, not a convenience. + * + * Two things the HTTP envelope has that this one deliberately does not: the + * cache-bust header (a frame is memoized by nothing) and a per-frame + * `traceparent` (frames carry no headers — trace context rides the upgrade + * instead, so the server parents to the connection's span, not to this one). + * + * One gap this cannot close: Vercel's observability *outgoing requests* view is + * built by instrumenting the global `fetch`, not by reading OTEL spans, so WS + * event writes stay absent from it however faithful the span is. Traces get the + * writes back; that view needs a real request, which is the transport's whole + * point to avoid. + */ async function postEventFrameOverWs( input: CreateEventV4InputBase & { eventType: EventType; @@ -838,54 +909,101 @@ async function postEventFrameOverWs( const { resolveWsTransport } = await import('./ws-transport.js'); const { runId } = input; const resolved = resolveWsTransport(runId, config); + // No span: resolving nothing means no write was attempted here at all — the + // caller falls through to HTTP, which opens its own. if (!resolved) return undefined; const { transport, wsUrl } = resolved; const endpoint = `${wsUrl}#runs/${encodeURIComponent(runId)}/events`; + // Same helper the HTTP path builds its URL with. `resolveWsTransport` already + // returned null for the proxy World, so this is always the direct + // workflow-server origin the socket itself points at. + const restUrl = eventsV4Url( + getHttpUrl(config).baseUrl, + runId, + input.eventType + ); - let reply: WsFrameReply; - try { - // `runId` isn't repeated here — it's already in `wsUrl`, one connection - // per run. The server's request-frame schema is a discriminated union on - // `type` with each type's payload nested under its own name, so a future - // request type is a new variant rather than a reshape of this one. - reply = await transport.request((reqId) => - encodeFrame( - { reqId, type: 'event', event: buildPostFrameMeta(input) }, - input.payload ?? new Uint8Array(0) - ) - ); - } catch (err) { - // Anything `transport.request()` throws means the frame was never acked. - // `code: 'TRANSPORT'` is the shape `utils.ts` gives a failed `fetch`, so - // one classification drives both transports — in-process retry gated by - // event type, then queue redelivery. An unwrapped `WsTransportError` - // would fail `WorkflowWorldError.is()` and classify as a USER_ERROR. - // Application errors are raised below, outside this try. - throw new WorkflowWorldError( - `POST ${endpoint} transport failure: ${ - err instanceof Error ? err.message : String(err) - }`, - { url: wsUrl, code: 'TRANSPORT', cause: err } - ); - } - - const status = wsReplyStatus(reply, endpoint); - const headerRecord = replyMetaToHeaderRecord(reply.meta); - - if (status < 200 || status >= 300) { - throw errorFromV4Response( - status, - headerRecord, - new TextDecoder().decode(reply.body), - 'createEvent', - endpoint - ); - } + return withHttpClientSpan( + { + method: 'POST', + url: restUrl, + attributes: { + ...WorkflowEventsTransport('ws'), + ...WorkflowEventType(input.eventType), + ...NetworkProtocolName('websocket'), + ...WorkflowWsUrl(wsUrl), + }, + }, + async (span) => { + const start = Date.now(); + let reply: WsFrameReply; + try { + // `runId` isn't repeated here — it's already in `wsUrl`, one connection + // per run. The server's request-frame schema is a discriminated union on + // `type` with each type's payload nested under its own name, so a future + // request type is a new variant rather than a reshape of this one. + reply = await transport.request((reqId) => { + // Recorded before the frame is sent so a request that fails, or one + // that never gets a reply, still carries the id the server logged it + // under. Assigned per attempt and per connection, so a retry or a + // reconnect legitimately re-uses low numbers. + span?.setAttributes({ ...WorkflowWsRequestId(reqId) }); + return encodeFrame( + { reqId, type: 'event', event: buildPostFrameMeta(input) }, + input.payload ?? new Uint8Array(0) + ); + }); + } catch (err) { + // Anything `transport.request()` throws means the frame was never acked. + // `code: 'TRANSPORT'` is the shape `utils.ts` gives a failed `fetch`, so + // one classification drives both transports — in-process retry gated by + // event type, then queue redelivery. An unwrapped `WsTransportError` + // would fail `WorkflowWorldError.is()` and classify as a USER_ERROR. + // Application errors are raised below, outside this try. + const error = new WorkflowWorldError( + `POST ${endpoint} transport failure: ${ + err instanceof Error ? err.message : String(err) + }`, + { url: wsUrl, code: 'TRANSPORT', cause: err } + ); + // `error.type: TRANSPORT` rather than an `HTTP ` value: there is + // no status, and the fetch path marks its own no-response failures the + // same way (`TIMEOUT`) instead of inventing one. + span?.setAttributes({ ...ErrorType('TRANSPORT') }); + span?.recordException?.(error); + throw error; + } + const ms = Date.now() - start; + + const status = wsReplyStatus(reply, endpoint); + const headerRecord = replyMetaToHeaderRecord(reply.meta); + const headers = { + get: (name: string) => headerRecord[name.toLowerCase()] ?? null, + }; + + // The same one-line `DEBUG` record the HTTP path emits, so a log grepped + // for event writes reads identically on either transport. + httpLog('POST', 'createEvent', { status, headers }, ms); + recordClientSpanStatus(span, status); + + if (status < 200 || status >= 300) { + const error = errorFromV4Response( + status, + headerRecord, + new TextDecoder().decode(reply.body), + 'createEvent', + endpoint + ); + span?.recordException?.(error); + throw error; + } - return { - headers: { get: (name) => headerRecord[name.toLowerCase()] ?? null }, - arrayBuffer: async () => reply.body.slice().buffer as ArrayBuffer, - }; + return { + headers, + arrayBuffer: async () => reply.body.slice().buffer as ArrayBuffer, + }; + } + ); } /** diff --git a/packages/world-vercel/src/http-core.ts b/packages/world-vercel/src/http-core.ts index 7acbf1f3c6..9e7b4fd35c 100644 --- a/packages/world-vercel/src/http-core.ts +++ b/packages/world-vercel/src/http-core.ts @@ -16,6 +16,7 @@ * dispatchers are passed in by the caller rather than imported here. */ +import type { Span } from '@opentelemetry/api'; import { getVercelOidcToken } from '@vercel/oidc'; import { EntityConflictError, @@ -84,12 +85,21 @@ const DIAGNOSTIC_HEADERS = [ 'x-vercel-mitigated', ] as const; +/** + * The one member the diagnostic/log helpers read headers through. `Headers` + * satisfies it, and so does the header record a WS reply frame's meta is + * flattened into — which has no `Headers` to offer. + */ +export interface HeaderLookup { + get(name: string): string | null; +} + /** * Extract the Vercel diagnostic response headers (x-vercel-id / * x-vercel-error / x-vercel-mitigated) as `key=value` strings, skipping any * that are absent. */ -export function getVercelDiagnostics(headers: Headers): string[] { +export function getVercelDiagnostics(headers: HeaderLookup): string[] { return DIAGNOSTIC_HEADERS.flatMap((header) => { const value = headers.get(header); return value ? [`${header}=${value}`] : []; @@ -100,19 +110,21 @@ export function getVercelDiagnostics(headers: Headers): string[] { * Format the Vercel diagnostic headers as a ` (a=b; c=d)` suffix for error * messages, or an empty string when none are present. */ -export function formatVercelDiagnostics(headers: Headers): string { +export function formatVercelDiagnostics(headers: HeaderLookup): string { const diagnostics = getVercelDiagnostics(headers); return diagnostics.length > 0 ? ` (${diagnostics.join('; ')})` : ''; } /** * One-line request log, emitted only when HTTP debug is enabled. `label` is a - * short request identifier (an endpoint path or full URL). + * short request identifier (an endpoint path or full URL). Takes the + * status/headers pair rather than a `Response` so the WS events transport, + * which has no `Response` to show, logs in the same format as the HTTP path. */ export function httpLog( method: string, label: string, - response: Response, + response: { status: number; headers: HeaderLookup }, ms: number ): void { if (!HTTP_DEBUG_ENABLED) return; @@ -293,20 +305,19 @@ export async function resolveVercelApiToken(opts?: { ); } -/** Parse the server address/port from a URL for OTEL span attributes. */ +/** Parse the server address/port from a URL for OTEL span attributes. + * `wss:` counts as a TLS scheme: the WS events transport reports its upgrade + * URL through here, and defaulting it to 80 would misreport the peer. */ function parseServer(url: string): { serverAddress?: string; serverPort?: number; } { try { const parsed = new URL(url); + const secure = parsed.protocol === 'https:' || parsed.protocol === 'wss:'; return { serverAddress: parsed.hostname, - serverPort: parsed.port - ? parseInt(parsed.port, 10) - : parsed.protocol === 'https:' - ? 443 - : 80, + serverPort: parsed.port ? parseInt(parsed.port, 10) : secure ? 443 : 80, }; } catch { return {}; @@ -338,18 +349,98 @@ export function httpClientSpanAttributes(args: { }; } -export interface InstrumentedFetchOptions { +export interface HttpClientSpanOptions { method: string; url: string; - headers: Headers; - body?: Uint8Array | string; - /** Undici dispatcher (typed `unknown`; see APIConfig.dispatcher). */ - dispatcher: unknown; /** * OTEL peer/rpc service label. 'workflow-server' for backend calls (default), * 'vercel-api' for direct api.vercel.com calls. */ peerService?: string; + /** + * Override the client-span name. Defaults to `http ${method}`. Pass a + * semantic operation name (e.g. `workflow.stream.write`) so the operation is + * discoverable in traces beyond the generic HTTP verb. + */ + spanName?: string; + /** Extra attributes merged on top of the standard HTTP attributes. */ + attributes?: Record; +} + +/** + * Open a CLIENT span for one outgoing request and run `fn` inside it. + * + * Split out of `instrumentedFetch` so a request path that cannot go through + * `fetch` still reports the *same* span: name, kind and the full + * `httpClientSpanAttributes` set. The WS events transport is the reason this + * exists — a frame on a multiplexed socket is a request in every sense the + * caller's trace cares about, but there is no `Response` and no `fetch` call to + * hang a span off, so it synthesizes one here (see `postEventFrameOverWs`). + * + * `fn` runs inside the active span, so anything it injects trace context into + * is parented to this span rather than to the caller's — which is the contract + * CLAUDE.md's trace-propagation rule describes. + */ +export async function withHttpClientSpan( + opts: HttpClientSpanOptions, + fn: (span?: Span) => Promise +): Promise { + const { + method, + url, + peerService = 'workflow-server', + spanName, + attributes, + } = opts; + return trace( + spanName ?? `http ${method}`, + { kind: await getSpanKind('CLIENT') }, + async (span) => { + // Diagnostic (DEBUG only): named spans are created and recording here, + // yet never found in the backend — log the exact span identity so the + // export side can be checked for this specific span id. + if (spanName && HTTP_DEBUG_ENABLED && span) { + const ctx = span.spanContext(); + console.warn( + '[workflow:otel-diag] span-open', + JSON.stringify({ + spanName, + traceId: ctx.traceId, + spanId: ctx.spanId, + recording: span.isRecording(), + }) + ); + } + span?.setAttributes( + httpClientSpanAttributes({ method, url, peerService }) + ); + if (attributes) span?.setAttributes(attributes); + return fn(span); + } + ); +} + +/** + * Stamp a response status onto a client span, marking a non-2xx with the same + * `error.type` the fetch path uses. Shared so a synthesized span reports a 409 + * identically to a real one — the status → error-type contract is what + * dashboards filter on, and it must not depend on which transport answered. + */ +export function recordClientSpanStatus( + span: Span | undefined, + status: number +): void { + span?.setAttributes({ ...HttpResponseStatusCode(status) }); + if (status < 200 || status >= 300) { + span?.setAttributes({ ...ErrorType(`HTTP ${status}`) }); + } +} + +export interface InstrumentedFetchOptions extends HttpClientSpanOptions { + headers: Headers; + body?: Uint8Array | string; + /** Undici dispatcher (typed `unknown`; see APIConfig.dispatcher). */ + dispatcher: unknown; /** * Per-request timeout in ms. Defaults to REQUEST_TIMEOUT_MS. Pass `null` to * disable (e.g. stream writes, which buffer arbitrarily large bodies). @@ -363,17 +454,6 @@ export interface InstrumentedFetchOptions { cacheBust?: boolean; /** Short label for logs (endpoint path). Defaults to the full URL. */ logLabel?: string; - /** - * Override the client-span name. Defaults to `http ${method}`. Pass a - * semantic operation name (e.g. `workflow.stream.write`) so the operation is - * discoverable in traces beyond the generic HTTP verb. - */ - spanName?: string; - /** - * Extra attributes merged onto the client span, in addition to the standard - * HTTP attributes (e.g. stream name / run id on stream operations). - */ - attributes?: Record; /** * When set, stamp the measured request round-trip (dispatch -> response * received, in ms) onto the client span under this attribute key. For a @@ -421,7 +501,7 @@ export async function instrumentedFetch( headers, body, dispatcher, - peerService = 'workflow-server', + peerService, timeoutMs = getRequestTimeoutMs(), signal: callerSignal, injectTraceContext = true, @@ -435,30 +515,9 @@ export async function instrumentedFetch( } = opts; const label = logLabel ?? url; - return trace( - spanName ?? `http ${method}`, - { kind: await getSpanKind('CLIENT') }, + return withHttpClientSpan( + { method, url, peerService, spanName, attributes }, async (span) => { - // Diagnostic (DEBUG only): named spans are created and recording here, - // yet never found in the backend — log the exact span identity so the - // export side can be checked for this specific span id. - if (spanName && HTTP_DEBUG_ENABLED && span) { - const ctx = span.spanContext(); - console.warn( - '[workflow:otel-diag] span-open', - JSON.stringify({ - spanName, - traceId: ctx.traceId, - spanId: ctx.spanId, - recording: span.isRecording(), - }) - ); - } - span?.setAttributes( - httpClientSpanAttributes({ method, url, peerService }) - ); - if (attributes) span?.setAttributes(attributes); - // Explicitly propagate trace context so the receiving server can parent // its spans to this client span — the custom undici dispatcher bypasses // ambient auto-instrumentation. No-ops when no OTEL SDK is registered. @@ -513,11 +572,10 @@ export async function instrumentedFetch( onTransportOutcome?.(); httpLog(method, label, response, ms); - span?.setAttributes({ ...HttpResponseStatusCode(response.status) }); + recordClientSpanStatus(span, response.status); if (durationAttribute) span?.setAttributes({ [durationAttribute]: ms }); if (!response.ok) { - span?.setAttributes({ ...ErrorType(`HTTP ${response.status}`) }); logCurlRepro(method, url, headers); if (buildError) { const error = await buildError(response); diff --git a/packages/world-vercel/src/telemetry.ts b/packages/world-vercel/src/telemetry.ts index 0a08f28e3c..0f75669eb2 100644 --- a/packages/world-vercel/src/telemetry.ts +++ b/packages/world-vercel/src/telemetry.ts @@ -211,6 +211,16 @@ export const HttpResponseStatusCode = SemanticConvention( /** Error type when request fails (standard OTEL: error.type) */ export const ErrorType = SemanticConvention('error.type'); +/** + * Application-layer protocol the request was carried over (standard OTEL: + * network.protocol.name). Only set on the WS events transport, whose client + * span is synthesized rather than produced by a real `fetch` — it is the + * attribute that keeps such a span honest about what actually went on the wire. + */ +export const NetworkProtocolName = SemanticConvention( + 'network.protocol.name' +); + /** Format used for parsing response body (cbor or json) */ export const WorldParseFormat = SemanticConvention<'cbor' | 'json'>( 'workflow.world.parse.format' @@ -254,3 +264,46 @@ export const WorkflowStreamOperation = SemanticConvention( export const WorkflowStreamStartIndex = SemanticConvention( 'workflow.stream.start_index' ); + +/** + * Transport an event write was carried over (workflow.events.transport): + * `http` | `ws`. Set on BOTH paths, deliberately: the two emit the same + * `http POST` client span against the same `url.full`, which is what keeps + * per-event traces and latency dashboards working across + * `WORKFLOW_EVENTS_TRANSPORT`, and this attribute is then the only way to slice + * one against the other. + */ +export const WorkflowEventsTransport = SemanticConvention<'http' | 'ws'>( + 'workflow.events.transport' +); + +/** Event type of a single event write (workflow.event.type), e.g. `step_started`. */ +export const WorkflowEventType = SemanticConvention( + 'workflow.event.type' +); + +/** + * The socket a WS event write actually travelled over + * (workflow.events.ws.url). `url.full` names the v4 REST endpoint the frame is + * forwarded into, so this is where the real wire destination is recorded. + */ +export const WorkflowWsUrl = SemanticConvention( + 'workflow.events.ws.url' +); + +/** + * Per-connection request id this write was multiplexed under + * (workflow.events.ws.req_id). The join key between a client span and the + * server's log line for the same frame. + */ +export const WorkflowWsRequestId = SemanticConvention( + 'workflow.events.ws.req_id' +); + +/** + * Which eager-reconnect attempt opened this socket + * (workflow.events.ws.reconnect_attempt); 0 for the invocation's first connect. + */ +export const WorkflowWsReconnectAttempt = SemanticConvention( + 'workflow.events.ws.reconnect_attempt' +); diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts index c071d59960..573bb5bce5 100644 --- a/packages/world-vercel/src/trace-propagation.test.ts +++ b/packages/world-vercel/src/trace-propagation.test.ts @@ -330,7 +330,7 @@ describe('ws events transport upgrade trace propagation', () => { resetWsEventsTransportsForTest(); }); - it('injects traceparent on the upgrade, parented to the invocation span', async () => { + it('injects traceparent on the upgrade, from its own connect span under the invocation', async () => { const { openWsChannel } = await import('./ws-transport.js'); const tracer = otelTrace.getTracer('test'); @@ -348,15 +348,31 @@ describe('ws events transport upgrade trace propagation', () => { // there is no per-frame traceparent to fall back on, so an uninjected // upgrade orphans the server's spans for the whole run. const traceparent = wsUpgrades[0]?.headers.traceparent; - expect(traceparent).toBe(`00-${traceId}-${spanId}-01`); + expect(traceparent).toMatch(new RegExp(`^00-${traceId}-[0-9a-f]{16}-01$`)); + + // The handshake carries a client span of its own, so what the server + // parents to is that span rather than the invocation directly — the same + // relationship `makeRequest` establishes, and what makes a write that waits + // on a handshake show the wait as a span instead of unattributed time. The + // fake socket never opens, so that span is still recording and hasn't been + // exported; the injected context is the only view of it here, and it must + // not be the invocation's own. `ws-transport-spans.test.ts` asserts the + // finished span against a socket that does open. + expect(traceparent).not.toBe(`00-${traceId}-${spanId}-01`); }); - it('opens the upgrade without traceparent when no span is active', async () => { + it('injects traceparent on the upgrade even when no span is active', async () => { const { openWsChannel } = await import('./ws-transport.js'); openWsChannel('wrun_2', { token: 'test-token' }); await vi.waitFor(() => expect(wsUpgrades).toHaveLength(1)); - expect(wsUpgrades[0]?.headers.traceparent).toBeUndefined(); + // Parity with every HTTP path: `instrumentedFetch` opens a client span + // whether or not one is already active, so the request is always + // correlatable. Before the connect span existed this upgrade went out + // uninjected and the server's spans for the whole run were orphaned. + expect(wsUpgrades[0]?.headers.traceparent).toMatch( + /^00-[0-9a-f]{32}-[0-9a-f]{16}-01$/ + ); // The rest of the upgrade must survive an absent propagator unchanged. expect(wsUpgrades[0]?.headers.authorization).toBe('Bearer test-token'); }); diff --git a/packages/world-vercel/src/ws-transport-spans.test.ts b/packages/world-vercel/src/ws-transport-spans.test.ts new file mode 100644 index 0000000000..1c12645201 --- /dev/null +++ b/packages/world-vercel/src/ws-transport-spans.test.ts @@ -0,0 +1,481 @@ +/** + * The per-write client span on the WebSocket events transport. + * + * On HTTP every event write is an `http POST` CLIENT span, opened by + * `instrumentedFetch`. A frame on a multiplexed socket makes no `fetch` call, so + * that span vanished when `WORKFLOW_EVENTS_TRANSPORT=ws` was introduced, taking + * the per-event view of a run with it. `postEventFrameOverWs` synthesizes an + * equivalent one; these tests pin that it is *equivalent* — same name, same + * kind, same `url.full`, same status/error attributes — while still saying which + * transport produced it. + * + * Unlike `events-v4-ws.test.ts`, nothing here mocks `resolveWsTransport`: the + * spans have to come out of the real selection + transport + adapter stack, over + * a fake socket that actually opens and replies (the harness from + * `ws-protocol-conformance.test.ts`), or they would prove nothing about what a + * deployment emits. + */ + +import { context, trace as otelTrace, propagation } from '@opentelemetry/api'; +import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks'; +import { W3CTraceContextPropagator } from '@opentelemetry/core'; +import { + BasicTracerProvider, + InMemorySpanExporter, + type ReadableSpan, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; +import { encode } from 'cbor-x'; +import { MockAgent } from 'undici'; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from 'vitest'; +import { withEventPostRetry } from './event-retry.js'; +import { createWorkflowRunEventV4 } from './events-v4.js'; +import { type DecodedFrame, decodeFrames, encodeFrame } from './frames.js'; +import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; + +vi.mock('@vercel/oidc', () => ({ + getVercelOidcToken: vi.fn().mockRejectedValue(new Error('no OIDC')), +})); + +const { FakeWebSocket, sockets } = vi.hoisted(() => { + const sockets: FakeSocket[] = []; + class FakeSocket { + static readonly OPEN = 1; + readyState = 0; + binaryType = ''; + readonly sent: Uint8Array[] = []; + private readonly listeners = new Map< + string, + Array<(...a: unknown[]) => void> + >(); + /** Set by the fixture server: called with each frame the client sends. */ + onFrame: ((raw: Uint8Array) => void) | null = null; + + constructor(_url: string, _opts?: unknown) { + sockets.push(this); + } + on(event: string, cb: (...a: unknown[]) => void): this { + const l = this.listeners.get(event) ?? []; + l.push(cb); + this.listeners.set(event, l); + return this; + } + emit(event: string, ...args: unknown[]): void { + for (const cb of [...(this.listeners.get(event) ?? [])]) cb(...args); + } + send(data: Uint8Array, cb?: (err?: Error) => void): void { + this.sent.push(data); + cb?.(); + this.onFrame?.(data); + } + close(code = 1000): void { + if (this.readyState === 3) return; + this.readyState = 3; + this.emit('close', code); + } + open(): void { + this.readyState = 1; + this.emit('open'); + } + deliver(frame: Uint8Array): void { + this.emit('message', Buffer.from(frame)); + } + } + return { FakeWebSocket: FakeSocket, sockets }; +}); + +vi.mock('ws', () => ({ WebSocket: FakeWebSocket })); + +const { openWsChannel, resetWsEventsTransportsForTest } = await import( + './ws-transport.js' +); + +const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; +const REST_URL = `${ORIGIN}/api/v4/runs/wrun_1/events/step_completed`; +const WS_URL = `${ORIGIN.replace(/^http/, 'ws')}/api/websockets/v1/runs/wrun_1`; +const CREATED_AT = '2026-06-10T00:00:00.000Z'; + +const exporter = new InMemorySpanExporter(); +const provider = new BasicTracerProvider(); +const contextManager = new AsyncLocalStorageContextManager(); + +beforeAll(() => { + provider.addSpanProcessor(new SimpleSpanProcessor(exporter)); + contextManager.enable(); + context.setGlobalContextManager(contextManager); + propagation.setGlobalPropagator(new W3CTraceContextPropagator()); + otelTrace.setGlobalTracerProvider(provider); +}); + +afterAll(async () => { + await provider.shutdown(); + context.disable(); + propagation.disable(); + otelTrace.disable(); +}); + +beforeEach(() => { + sockets.length = 0; + exporter.reset(); + resetWsEventsTransportsForTest(); + process.env.WORKFLOW_EVENTS_TRANSPORT = 'ws'; + vi.spyOn(console, 'log').mockImplementation(() => {}); + vi.spyOn(console, 'error').mockImplementation(() => {}); +}); + +afterEach(() => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; + resetWsEventsTransportsForTest(); + vi.restoreAllMocks(); +}); + +const input = { + runId: 'wrun_1', + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', +} as const; + +/** The materialized CBOR body a `step_completed` write answers with. */ +const materializedBody = (eventId = 'evnt_1') => + new Uint8Array( + encode({ + event: { + eventId, + runId: 'wrun_1', + createdAt: CREATED_AT, + eventType: 'step_completed', + specVersion: 2, + correlationId: 'step_1', + eventData: { result: new Uint8Array() }, + }, + step: { + runId: 'wrun_1', + stepId: 'step_1', + stepName: 'step', + status: 'completed', + attempt: 1, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + }, + }) + ); + +async function decodeOne(raw: Uint8Array): Promise { + for await (const frame of decodeFrames( + (async function* () { + yield raw; + })() + )) { + return frame; + } + throw new Error('empty frame'); +} + +interface RouteResponse { + status: number; + body?: Uint8Array; +} + +/** Answer every frame the client sends the way the v4 WS route would. */ +function attachFixtureServer( + socket: InstanceType, + handler: (n: number) => RouteResponse +) { + let n = 0; + socket.onFrame = (raw) => { + void decodeOne(raw).then((frame) => { + const res = handler(++n); + socket.deliver( + encodeFrame( + { reqId: frame.meta.reqId, type: 'event_ack', status: res.status }, + res.body ?? new Uint8Array(0) + ) + ); + }); + }; +} + +/** + * Open the channel the way the flow route does, then let the handshake + * complete with a fixture server attached. Returns once the socket is live, so + * a write issued afterwards is a steady-state write rather than one that pays + * for the handshake. + */ +async function withOpenChannel( + handler: (n: number) => RouteResponse = () => ({ + status: 201, + body: materializedBody(), + }) +) { + const release = openWsChannel(input.runId, { token: 'test-token' }); + await vi.waitFor(() => expect(sockets.length).toBeGreaterThan(0)); + const socket = sockets[0]; + attachFixtureServer(socket, handler); + socket.open(); + return { socket, release }; +} + +const spansNamed = (name: string): ReadableSpan[] => + exporter.getFinishedSpans().filter((s) => s.name === name); + +const writeSpan = (): ReadableSpan => { + const spans = spansNamed('http POST'); + expect(spans).toHaveLength(1); + return spans[0]; +}; + +describe('per-write client span', () => { + it('emits one `http POST` CLIENT span per event write', async () => { + await withOpenChannel(); + + await createWorkflowRunEventV4(input, { token: 'test-token' }); + + const span = writeSpan(); + // SpanKind.CLIENT === 2. Asserted as the literal so a change to the kind + // (which is what makes it render as an outgoing request) is visible here. + expect(span.kind).toBe(2); + expect(span.attributes['http.request.method']).toBe('POST'); + expect(span.attributes['http.response.status_code']).toBe(201); + expect(span.attributes['error.type']).toBeUndefined(); + }); + + it('reports the v4 REST endpoint as url.full, byte-identical to the HTTP path', async () => { + await withOpenChannel(); + + await createWorkflowRunEventV4(input, { token: 'test-token' }); + + // Not the socket URL: the server forwards the frame into this route, and + // naming it is what keeps a dashboard keyed on `http POST` + `url.full` + // working across the transport flag. The HTTP-path assertion in + // `transport parity` below pins that these two strings are the same one. + expect(writeSpan().attributes['url.full']).toBe(REST_URL); + expect(writeSpan().attributes['server.address']).toBe( + new URL(ORIGIN).hostname + ); + expect(writeSpan().attributes['peer.service']).toBe('workflow-server'); + expect(writeSpan().attributes['rpc.service']).toBe('workflow-server'); + }); + + it('says plainly that it was a frame, not an HTTP request', async () => { + await withOpenChannel(); + + await createWorkflowRunEventV4(input, { token: 'test-token' }); + + // A synthetic span that hid its transport would be a trap: the three + // attributes below are what stop `url.full` from being read as "a request + // was made to this URL". + const span = writeSpan(); + expect(span.attributes['workflow.events.transport']).toBe('ws'); + expect(span.attributes['network.protocol.name']).toBe('websocket'); + expect(span.attributes['workflow.events.ws.url']).toBe(WS_URL); + expect(span.attributes['workflow.event.type']).toBe('step_completed'); + }); + + it('carries the reqId that joins it to the server log line for the same frame', async () => { + await withOpenChannel((n) => ({ + status: 201, + body: materializedBody(`evnt_${n}`), + })); + + await createWorkflowRunEventV4(input, { token: 'test-token' }); + await createWorkflowRunEventV4( + { ...input, correlationId: 'step_2' }, + { token: 'test-token' } + ); + + const ids = spansNamed('http POST').map( + (s) => s.attributes['workflow.events.ws.req_id'] + ); + expect(ids).toEqual([1, 2]); + }); + + it('parents each write to the caller, not to the connection', async () => { + await withOpenChannel(); + + const tracer = otelTrace.getTracer('test'); + let invocationSpanId = ''; + await tracer.startActiveSpan('flow-invocation', async (span) => { + invocationSpanId = span.spanContext().spanId; + await createWorkflowRunEventV4(input, { token: 'test-token' }); + span.end(); + }); + + // A write that joins an already-open socket must not be nested under the + // handshake that happened to precede it — otherwise every write in a run + // hangs off the first one and the trace stops showing the step it belongs + // to. + expect(writeSpan().parentSpanId).toBe(invocationSpanId); + }); + + it('does not synthesize a span for a write that never reached the socket', async () => { + // No `openWsChannel`, so `resolveWsTransport` finds nothing and the write + // falls through to HTTP, which opens its own span. Synthesizing one here + // would double-count every write on a run whose channel failed to open. + const agent = new MockAgent(); + agent.disableNetConnect(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/step_completed', + method: 'POST', + }) + .reply(200, materializedBody(), { + headers: { 'x-wf-event-id': 'evnt_1' }, + }); + + await createWorkflowRunEventV4(input, { + token: 'test-token', + dispatcher: agent, + }); + + const spans = spansNamed('http POST'); + expect(spans).toHaveLength(1); + expect(spans[0].attributes['workflow.events.transport']).toBe('http'); + expect(sockets).toHaveLength(0); + }); +}); + +describe('failure reporting', () => { + it('records a non-2xx reply with the same error.type the fetch path uses', async () => { + await withOpenChannel(() => ({ + status: 409, + body: new TextEncoder().encode('{"message":"already applied"}'), + })); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(); + + const span = writeSpan(); + expect(span.attributes['http.response.status_code']).toBe(409); + expect(span.attributes['error.type']).toBe('HTTP 409'); + // SpanStatusCode.ERROR === 2. + expect(span.status.code).toBe(2); + expect(span.events.map((e) => e.name)).toContain('exception'); + }); + + it('records a dead socket as TRANSPORT, with no status to report', async () => { + const { socket } = await withOpenChannel(); + // Answer nothing and drop the connection under the in-flight write. + socket.onFrame = () => socket.close(1006); + + await expect( + createWorkflowRunEventV4(input, { token: 'test-token' }) + ).rejects.toThrow(/transport failure/); + + const span = writeSpan(); + expect(span.attributes['error.type']).toBe('TRANSPORT'); + // Inventing a status here would make a write that was never acked + // indistinguishable from one the server answered. + expect(span.attributes['http.response.status_code']).toBeUndefined(); + expect(span.status.code).toBe(2); + }); + + it('emits one span per attempt, like the HTTP path does', async () => { + // The shared retry policy (`event-retry.ts`) re-enters the adapter rather + // than retrying inside it, so a retried write is two spans — matching + // `instrumentedFetch`, which opens one per `fetch` call rather than one per + // logical write. A single span spanning both attempts would hide the first + // failure and report the retry's latency as the write's. + await withOpenChannel((n) => + n === 1 + ? { + status: 500, + body: new TextEncoder().encode('{"message":"transient"}'), + } + : { status: 201, body: materializedBody() } + ); + + const result = await withEventPostRetry( + () => createWorkflowRunEventV4(input, { token: 'test-token' }), + 'step_completed' + ); + expect(result.event.eventId).toBe('evnt_1'); + + const spans = spansNamed('http POST'); + expect(spans).toHaveLength(2); + expect(spans.map((s) => s.attributes['http.response.status_code'])).toEqual( + [500, 201] + ); + expect(spans.map((s) => s.attributes['error.type'])).toEqual([ + 'HTTP 500', + undefined, + ]); + }); +}); + +describe('connection span', () => { + it('times the handshake under its own operation-named span', async () => { + const { socket } = await withOpenChannel(); + await createWorkflowRunEventV4(input, { token: 'test-token' }); + expect(socket.readyState).toBe(1); + + // Named for the operation, not `http GET`: the upgrade is a real HTTP + // request, but bucketing it with the event writes it enables would make + // both unreadable. + const connect = spansNamed('workflow.events.ws.connect'); + expect(connect).toHaveLength(1); + expect(connect[0].kind).toBe(2); + expect(connect[0].attributes['url.full']).toBe(WS_URL); + expect(connect[0].attributes['workflow.events.transport']).toBe('ws'); + expect(connect[0].attributes['workflow.events.ws.reconnect_attempt']).toBe( + 0 + ); + }); + + it('records a refused upgrade rather than leaving the gap unexplained', async () => { + openWsChannel(input.runId, { token: 'test-token' }); + await vi.waitFor(() => expect(sockets.length).toBeGreaterThan(0)); + sockets[0].close(1006); + + await vi.waitFor(() => + expect(spansNamed('workflow.events.ws.connect')).toHaveLength(1) + ); + const connect = spansNamed('workflow.events.ws.connect')[0]; + expect(connect.attributes['error.type']).toBe('TRANSPORT'); + expect(connect.status.code).toBe(2); + }); +}); + +describe('transport parity', () => { + it('emits the same span name and url.full on HTTP as on ws', async () => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; + const agent = new MockAgent(); + agent.disableNetConnect(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/step_completed', + method: 'POST', + }) + .reply(200, materializedBody(), { + headers: { 'x-wf-event-id': 'evnt_1' }, + }); + + await createWorkflowRunEventV4(input, { + token: 'test-token', + dispatcher: agent, + }); + + // The whole point of synthesizing the WS span: a trace taken either side of + // the flag reads the same, so the A/B the flag exists for compares like + // with like. Only `workflow.events.transport` separates them. + const span = writeSpan(); + expect(span.attributes['url.full']).toBe(REST_URL); + expect(span.attributes['http.request.method']).toBe('POST'); + expect(span.attributes['workflow.event.type']).toBe('step_completed'); + expect(span.attributes['workflow.events.transport']).toBe('http'); + expect(span.attributes['network.protocol.name']).toBeUndefined(); + agent.assertNoPendingInterceptors(); + }); +}); diff --git a/packages/world-vercel/src/ws-transport.ts b/packages/world-vercel/src/ws-transport.ts index d612e98dbf..20527dffe1 100644 --- a/packages/world-vercel/src/ws-transport.ts +++ b/packages/world-vercel/src/ws-transport.ts @@ -25,8 +25,18 @@ import { getVercelOidcToken } from '@vercel/oidc'; import { WebSocket } from 'ws'; import { type DecodedFrame, decodeFrames } from './frames.js'; -import { getRequestTimeoutMs, headersToRecord } from './http-core.js'; -import { injectTraceContextIntoHeaders } from './telemetry.js'; +import { + getRequestTimeoutMs, + headersToRecord, + withHttpClientSpan, +} from './http-core.js'; +import { + ErrorType, + injectTraceContextIntoHeaders, + NetworkProtocolName, + WorkflowEventsTransport, + WorkflowWsReconnectAttempt, +} from './telemetry.js'; import { type APIConfig, getHttpConfig, getHttpUrl } from './utils.js'; import { isWsEventsTransportEnabled } from './ws-transport-enabled.js'; @@ -295,7 +305,7 @@ class WsEventsTransport { // Clearing the slot here rather than from the socket handlers keeps the // bookkeeping in one place, and stops a stale socket's late `close` from // nulling out a newer connect that has since taken the slot. - this.connecting ??= this.connect().finally(() => { + this.connecting ??= this.connectWithSpan().finally(() => { this.connecting = null; }); return this.connecting; @@ -353,6 +363,46 @@ class WsEventsTransport { if (reason === 'auth_expiry') this.needsFreshToken = true; } + /** + * The handshake, wrapped in its own CLIENT span. + * + * The upgrade is the one genuinely-HTTP request this transport makes, and + * until it had a span it was the invisible half of every WS write's latency: + * a write that waits on a handshake shows the wait inside its own span with + * nothing to attribute it to, and an eager reconnect between two writes shows + * up nowhere at all. Named for the operation rather than `http GET` so it + * doesn't land in the same bucket as the event writes it enables. + * + * It also puts `resolveUpgradeHeaders`' trace-context injection inside a + * client span, matching `makeRequest`'s contract (CLAUDE.md): the server's + * per-event spans parent to *this* span rather than directly to whichever + * invocation happened to open the socket, so the handshake and everything the + * server does over the connection hang together. + */ + private connectWithSpan(): Promise { + return withHttpClientSpan( + { + method: 'GET', + url: this.wsUrl, + spanName: 'workflow.events.ws.connect', + attributes: { + ...WorkflowEventsTransport('ws'), + ...NetworkProtocolName('websocket'), + ...WorkflowWsReconnectAttempt(this.reconnectAttempts), + }, + }, + async (span) => { + try { + return await this.connect(); + } catch (err) { + span?.setAttributes({ ...ErrorType('TRANSPORT') }); + if (err instanceof Error) span?.recordException?.(err); + throw err; + } + } + ); + } + private connect(): Promise { return new Promise((resolve, reject) => { void (async () => { From 2389b34d060b19a0091a6f201fbef630b79ce096 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Thu, 13 Aug 2026 01:04:56 +0000 Subject: [PATCH 2/4] chore: trim WS spans changeset to the user-facing summary Co-Authored-By: Claude Opus 5 Co-Authored-By: shalabhchaturvedi-7802 Co-Authored-By: shalabhc --- .changeset/ws-transport-synthetic-spans.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/ws-transport-synthetic-spans.md b/.changeset/ws-transport-synthetic-spans.md index 208560b73f..ff03012414 100644 --- a/.changeset/ws-transport-synthetic-spans.md +++ b/.changeset/ws-transport-synthetic-spans.md @@ -2,4 +2,4 @@ '@workflow/world-vercel': patch --- -Restore the per-event client span on the WebSocket events transport (`WORKFLOW_EVENTS_TRANSPORT=ws`). Each event write now emits a synthesized `http POST` CLIENT span with the same name, kind and `url.full` as the HTTP path — plus `workflow.events.transport`, `network.protocol.name`, `workflow.events.ws.url` and `workflow.events.ws.req_id` so a synthetic span is never mistaken for a real request. The WebSocket handshake gets its own `workflow.events.ws.connect` span, and injects trace context from inside it. The HTTP path is unchanged apart from gaining `workflow.events.transport: 'http'` and `workflow.event.type`. +Restore the per-event client span on the WebSocket events transport (`WORKFLOW_EVENTS_TRANSPORT=ws`), and add a `workflow.events.ws.connect` span for the handshake. Event write spans now carry `workflow.events.transport` and `workflow.event.type` on both transports. From c2ea12b74af4736da22514e21440d6e978145441 Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Thu, 13 Aug 2026 01:09:57 +0000 Subject: [PATCH 3/4] fix(world-vercel): only tag event-write spans with transport Signed-off-by: Shalabh Chaturvedi Co-Authored-By: shalabhchaturvedi-7802 --- packages/world-vercel/src/events-v4.ts | 10 ++-- .../src/ws-transport-spans.test.ts | 53 ++++++++++++++++++- 2 files changed, 56 insertions(+), 7 deletions(-) diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 09587c11c4..93dd6ba96f 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -108,10 +108,7 @@ async function fetchV4( headers: init.headers, body: init.body, dispatcher, - // Named on both transports so a trace or a latency dashboard can tell which - // one served a write — they are otherwise deliberately indistinguishable, - // right down to the span name and `url.full`. See `postEventFrameOverWs`. - attributes: { ...WorkflowEventsTransport('http'), ...attributes }, + attributes, // Repeated transport failures retire the shared events pool and the next // request builds a fresh one. undici keeps a black-holed HTTP/2 session in // service indefinitely, so without this every request routed onto it fails @@ -709,7 +706,10 @@ async function postWorkflowRunEventV4( { method: 'POST', headers, body: frame }, config, 'createEvent', - WorkflowEventType(input.eventType) + { + ...WorkflowEventsTransport('http'), + ...WorkflowEventType(input.eventType), + } ); } diff --git a/packages/world-vercel/src/ws-transport-spans.test.ts b/packages/world-vercel/src/ws-transport-spans.test.ts index 1c12645201..66ac5daf84 100644 --- a/packages/world-vercel/src/ws-transport-spans.test.ts +++ b/packages/world-vercel/src/ws-transport-spans.test.ts @@ -38,8 +38,16 @@ import { vi, } from 'vitest'; import { withEventPostRetry } from './event-retry.js'; -import { createWorkflowRunEventV4 } from './events-v4.js'; -import { type DecodedFrame, decodeFrames, encodeFrame } from './frames.js'; +import { + createWorkflowRunEventV4, + getEventV4, +} from './events-v4.js'; +import { + type DecodedFrame, + decodeFrames, + encodeFrame, + V4_FRAME_CONTENT_TYPE, +} from './frames.js'; import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; vi.mock('@vercel/oidc', () => ({ @@ -448,6 +456,47 @@ describe('connection span', () => { }); describe('transport parity', () => { + it('does not tag an HTTP event read as an event-write transport', async () => { + delete process.env.WORKFLOW_EVENTS_TRANSPORT; + const agent = new MockAgent(); + agent.disableNetConnect(); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/evnt_1?remoteRefBehavior=resolve', + method: 'GET', + }) + .reply( + 200, + encodeFrame( + { + eventId: 'evnt_1', + runId: 'wrun_1', + eventType: 'run_created', + createdAt: CREATED_AT, + eventData: { + deploymentId: 'dpl_1', + workflowName: 'workflow', + input: null, + }, + }, + new Uint8Array() + ), + { headers: { 'content-type': V4_FRAME_CONTENT_TYPE } } + ); + + await getEventV4('wrun_1', 'evnt_1', 'resolve', { + token: 'test-token', + dispatcher: agent, + }); + + const spans = spansNamed('http GET'); + expect(spans).toHaveLength(1); + expect(spans[0].attributes['workflow.events.transport']).toBeUndefined(); + expect(spans[0].attributes['workflow.event.type']).toBeUndefined(); + agent.assertNoPendingInterceptors(); + }); + it('emits the same span name and url.full on HTTP as on ws', async () => { delete process.env.WORKFLOW_EVENTS_TRANSPORT; const agent = new MockAgent(); From daa725986052468bd836730d2fa77a1afbca9fec Mon Sep 17 00:00:00 2001 From: "vercel[bot]" <35613825+vercel[bot]@users.noreply.github.com> Date: Thu, 13 Aug 2026 04:05:10 +0000 Subject: [PATCH 4/4] fix(world-vercel): format WS transport span regression test Signed-off-by: Shalabh Chaturvedi Co-Authored-By: Shalabh Chaturvedi --- packages/world-vercel/src/ws-transport-spans.test.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/packages/world-vercel/src/ws-transport-spans.test.ts b/packages/world-vercel/src/ws-transport-spans.test.ts index 66ac5daf84..9576f7eeba 100644 --- a/packages/world-vercel/src/ws-transport-spans.test.ts +++ b/packages/world-vercel/src/ws-transport-spans.test.ts @@ -38,10 +38,7 @@ import { vi, } from 'vitest'; import { withEventPostRetry } from './event-retry.js'; -import { - createWorkflowRunEventV4, - getEventV4, -} from './events-v4.js'; +import { createWorkflowRunEventV4, getEventV4 } from './events-v4.js'; import { type DecodedFrame, decodeFrames,