From aaab4caf74edb6e1bd4e4df3bc809ae035969606 Mon Sep 17 00:00:00 2001 From: Soul and Code Date: Sun, 25 Oct 2026 03:42:44 +0700 Subject: [PATCH] Worker and workflow span export is implemented, but I couldn't run anything here, so none of it has been type-checked or **Already in the repo.** The exporter was already shared in `packages/runtime`. The worker entry points already exported per-activity spans, so only the workflow side, links, allowlist and docs were missing. **Open items you should know about** - The integration test (fake collector, Temporal local test server, trace-id continuity, parent/child structure, continue-as-new link, silent replay) needs `@temporalio/testing`. I did not add it to `package.json`, because I can't regenerate the lockfile and `npm ci` would break. The test skips itself when the package or the dev server is missing, so in this setup it will skip, not pass. Add the dependency and lockfile entry to make it run. - The bundler setup in that test (`.js` to `.ts` extension alias, `@space/contracts` alias) is untested and may need adjusting. - `KNOWN_LIMITATIONS.md` is generated from `review-findings.json`, so I left OBS-001/OBS-007 untouched. - There is no Jaeger screenshot for the PR. That manual check still needs to be done. PR title: Export Temporal workflow spans over OTLP: one trace from API request to every activity **What changed** - **Workflow spans:** a workflow interceptor (`workflow-interceptors.ts`) does no I/O and records one span per workflow run. Its ids come from `uuid4` and workflow time. Attempts, status and `reasonCode`/`category` come from existing data. - **Propagation:** the trace context travels in a Temporal header (`space-traceparent`) to activities, child workflows and continue-as-new runs. Child workflows nest under their parent, and activity spans parent to the workflow span. - **Replay:** a sink (`workflow-span-sinks.ts`, `callDuringReplay: false`) turns each record into an OTLP span through the existing exporter. A broken exporter can't fail a workflow task. - **Worker wiring:** `tracingWorkerOptions(spans)` is used by `worker.ts`, `memory-maintenance-worker.ts` and `memory-mutation-worker.ts`. With `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` unset, nothing is added to workfl --- apps/api/src/trace-context.ts | 9 +- apps/worker/src/activity-observability.ts | 105 ++++------ apps/worker/src/memory-maintenance-worker.ts | 4 +- apps/worker/src/memory-mutation-worker.ts | 4 +- .../src/room-background-continuation.ts | 11 +- apps/worker/src/span-attributes.ts | 52 +++++ apps/worker/src/trace-correlation.ts | 73 +++++++ apps/worker/src/trace-propagation.ts | 33 +++ apps/worker/src/worker.ts | 8 +- apps/worker/src/workflow-interceptors.ts | 139 +++++++++++++ apps/worker/src/workflow-span-sinks.ts | 70 +++++++ .../tests/activity-observability.test.ts | 2 +- .../worker/tests/support/tracing-workflows.ts | 20 ++ apps/worker/tests/workflow-tracing.test.ts | 195 ++++++++++++++++++ docs/configuration-reference.md | 20 ++ docs/quality/observability.md | 34 ++- packages/contracts/src/schemas.ts | 3 + packages/runtime/src/otlp-span-exporter.ts | 8 +- 18 files changed, 702 insertions(+), 88 deletions(-) create mode 100644 apps/worker/src/span-attributes.ts create mode 100644 apps/worker/src/trace-correlation.ts create mode 100644 apps/worker/src/trace-propagation.ts create mode 100644 apps/worker/src/workflow-interceptors.ts create mode 100644 apps/worker/src/workflow-span-sinks.ts create mode 100644 apps/worker/tests/support/tracing-workflows.ts create mode 100644 apps/worker/tests/workflow-tracing.test.ts diff --git a/apps/api/src/trace-context.ts b/apps/api/src/trace-context.ts index af9e55efd..1e723ad64 100644 --- a/apps/api/src/trace-context.ts +++ b/apps/api/src/trace-context.ts @@ -94,9 +94,14 @@ export class TraceContextRegistry { export const requestTraceContexts = new TraceContextRegistry(); -/** traceparent for work started on behalf of this Space request id, if known. */ +/** + * traceparent for work started on behalf of this Space request id, if known. + * It names the request's own span, which the OTLP exporter sends as the SERVER + * span, so workflow spans become its children in one connected trace. + */ export function traceparentForTraceId(traceId: string | null | undefined, registry = requestTraceContexts): string | null { - return childTraceparent(registry.lookup(traceId)?.traceparent ?? null); + const traceparent = registry.lookup(traceId)?.traceparent ?? null; + return parseTraceparent(traceparent) ? traceparent : null; } /** Adds the request's traceparent to a turn input unless it already has a valid one. */ diff --git a/apps/worker/src/activity-observability.ts b/apps/worker/src/activity-observability.ts index ea4d021d1..1e2b4f0f0 100644 --- a/apps/worker/src/activity-observability.ts +++ b/apps/worker/src/activity-observability.ts @@ -7,9 +7,11 @@ import { formatTraceparent, isSafeErrorCode, parseTraceparent, - traceIdFromTraceparent, type FailureCategory } from "@space/contracts"; +import { allowlistedSpanAttributes, correlationSpanAttributes } from "./span-attributes.js"; +import { activityCorrelation, argumentTraceparent } from "./trace-correlation.js"; +import { TRACEPARENT_HEADER, decodeTraceHeader } from "./trace-propagation.js"; import type { SpanExporter } from "@space/runtime"; import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js"; import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js"; @@ -64,56 +66,7 @@ export function childTraceparent(traceparent: string | null | undefined): string return formatTraceparent(parsed.traceId, spanId, parsed.flags); } -// Keys whose values are identifiers. Prompts, content and messages are never read. -const correlationKeys = [ - "traceId", - "traceparent", - "requestId", - "roomId", - "paneId", - "missionId", - "roomAgentMissionId", - "agentRunId", - "agentSessionId", - "actionId", - "goalId", - "dotId", - "turnId", - "cliTaskId", - "revisionId" -] as const; -const nestedKeys = ["turn", "item", "bridge", "input", "turnInput"] as const; - -type CorrelationKey = (typeof correlationKeys)[number]; - -function readCorrelation(value: unknown, into: Partial>, depth: number): void { - if (!value || typeof value !== "object" || Array.isArray(value)) return; - const record = value as Record; - for (const key of correlationKeys) { - if (into[key] !== undefined) continue; - const identifier = logIdentifier(record[key]); - if (identifier !== null) into[key] = identifier; - } - if (depth <= 0) return; - for (const key of nestedKeys) readCorrelation(record[key], into, depth - 1); -} - -/** Allowlisted correlation fields from activity arguments (identifiers only). */ -export function activityCorrelation(args: readonly unknown[]): Record { - const found: Partial> = {}; - for (const arg of args) readCorrelation(arg, found, 2); - const correlation: Record = {}; - if (found.traceId) correlation.traceId = found.traceId; - const w3cTraceId = traceIdFromTraceparent(found.traceparent); - if (w3cTraceId) correlation.w3cTraceId = w3cTraceId; - if (found.requestId && found.requestId !== found.traceId) correlation.parentTraceId = found.requestId; - const missionId = found.missionId ?? found.roomAgentMissionId; - if (missionId) correlation.missionId = missionId; - for (const key of ["roomId", "paneId", "agentRunId", "agentSessionId", "actionId", "goalId", "dotId", "turnId", "cliTaskId", "revisionId"] as const) { - if (found[key]) correlation[key] = found[key]!; - } - return correlation; -} +export { activityCorrelation }; function errorCodeOf(error: unknown): string { const record = error as { code?: unknown; type?: unknown; name?: unknown } | null; @@ -137,12 +90,8 @@ export interface ObserveActivityDependencies { cancelled?: () => boolean; /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */ spans?: SpanExporter | null; -} - -function activityTraceparent(args: readonly unknown[]) { - const found: Partial> = {}; - for (const arg of args) readCorrelation(arg, found, 2); - return parseTraceparent(found.traceparent); + /** traceparent the workflow interceptor propagated in a Temporal header (the workflow span). */ + traceparentHeader?: string | null; } export async function observeActivityExecution( @@ -194,19 +143,36 @@ export async function observeActivityExecution( queueMs, rssBytes: process.memoryUsage().rss }); - const parent = dependencies.spans ? activityTraceparent(args) : null; + // The workflow interceptor's header names the workflow span; without it + // (workflow tracing off) the API's traceparent in the input is the parent. + const parent = dependencies.spans ? parseTraceparent(dependencies.traceparentHeader) ?? argumentTraceparent(args) : null; if (parent) { - dependencies.spans!.record({ - traceId: parent.traceId, - spanId: randomBytes(8).toString("hex"), - parentSpanId: parent.parentId, - name: `activity ${activityType}`, - startTimeUnixNano: BigInt(startedAt) * 1_000_000n, - endTimeUnixNano: BigInt(startedAt + durationMs) * 1_000_000n, - attributes: { "space.activity_type": activityType, "space.outcome": outcome, "space.attempt": info.attempt }, - error: outcome === "FAILED" || outcome === "FAILED_RESULT", - kind: 1 - }); + try { + dependencies.spans!.record({ + traceId: parent.traceId, + spanId: randomBytes(8).toString("hex"), + parentSpanId: parent.parentId, + name: `activity ${activityType}`, + startTimeUnixNano: BigInt(startedAt) * 1_000_000n, + endTimeUnixNano: BigInt(startedAt + durationMs) * 1_000_000n, + attributes: allowlistedSpanAttributes({ + ...correlationSpanAttributes(activityCorrelation(args)), + "space.workflow_id": info.workflowExecution?.workflowId, + "space.run_id": info.workflowExecution?.runId, + "space.workflow_type": info.workflowType, + "space.activity_type": activityType, + "space.attempt": info.attempt, + "space.outcome": outcome, + "space.reason_code": reasonCode, + "space.category": category, + "space.duration_ms": durationMs + }), + error: outcome === "FAILED" || outcome === "FAILED_RESULT", + kind: 1 + }); + } catch { + // Tracing must never change an activity's outcome. + } } }; @@ -229,6 +195,7 @@ export function createActivityObservabilityInterceptors(dependencies: Omit observeActivityExecution(context.info, input.args, () => next(input), { ...dependencies, + traceparentHeader: decodeTraceHeader(input.headers, TRACEPARENT_HEADER), cancelled: () => context.cancellationSignal.aborted }) } diff --git a/apps/worker/src/memory-maintenance-worker.ts b/apps/worker/src/memory-maintenance-worker.ts index e12046033..2e72797cc 100644 --- a/apps/worker/src/memory-maintenance-worker.ts +++ b/apps/worker/src/memory-maintenance-worker.ts @@ -8,7 +8,7 @@ import { } from "./memory-maintenance-config.js"; import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js"; import { otlpSpanExporterFromEnv } from "@space/runtime"; -import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { tracingWorkerOptions } from "./workflow-span-sinks.js"; import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js"; const service = "space-memory-maintenance-worker"; @@ -30,7 +30,7 @@ if (!memoryMaintenanceEnabled(process.env)) { workflowsPath, activities, ...memoryMaintenanceWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + ...tracingWorkerOptions(spans) }); console.log(JSON.stringify({ service, diff --git a/apps/worker/src/memory-mutation-worker.ts b/apps/worker/src/memory-mutation-worker.ts index c2eadf657..6cdf982b3 100644 --- a/apps/worker/src/memory-mutation-worker.ts +++ b/apps/worker/src/memory-mutation-worker.ts @@ -4,7 +4,7 @@ import { executePersistedMemoryChangeSet } from "./memory-mutation-activity.js"; import { MEMORY_MUTATION_TASK_QUEUE, memoryMutationsEnabled, memoryMutationWorkerOptions } from "./memory-mutation-config.js"; import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js"; import { otlpSpanExporterFromEnv } from "@space/runtime"; -import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { tracingWorkerOptions } from "./workflow-span-sinks.js"; import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js"; const service = "space-memory-mutation-worker"; @@ -23,7 +23,7 @@ if (!memoryMutationsEnabled(process.env)) { workflowsPath: fileURLToPath(new URL("./memory-mutation-workflows.js", import.meta.url)), activities: { executePersistedMemoryChangeSet }, ...memoryMutationWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + ...tracingWorkerOptions(spans) }); console.log(JSON.stringify({ service, status: "running", address, namespace, taskQueue: MEMORY_MUTATION_TASK_QUEUE })); const stopReloadWatch = reloadWhenServicesCommitChanges(worker, { log: (message) => console.log(JSON.stringify({ service, status: "reloading", message })) }); diff --git a/apps/worker/src/room-background-continuation.ts b/apps/worker/src/room-background-continuation.ts index e15c29a21..9e39d0a60 100644 --- a/apps/worker/src/room-background-continuation.ts +++ b/apps/worker/src/room-background-continuation.ts @@ -1,4 +1,7 @@ +import { randomBytes } from "node:crypto"; import { + formatTraceparent, + parseTraceparent, buildCodexAppServerTurnWorkflowId, buildRoomAgentSupervisorWorkflowId, roomAgentSupervisorQueueItemSchema, @@ -120,6 +123,12 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis "If the entire request is already verified, emit the space-room-verification block with version 1, status VERIFIED and summary.", "Otherwise execute only the remaining work; do not infer completion from zero plans or idle panes." ].join("\n"); + const sourceTraceparent = parseTraceparent(source.queueItem.turn.traceparent); + // The continuation is its own trace (trace:bg-) that links back to the span it continues. + const traceContext = sourceTraceparent + ? { traceparent: formatTraceparent(randomBytes(16).toString("hex"), randomBytes(8).toString("hex"), sourceTraceparent.flags), + continuesTraceparent: formatTraceparent(sourceTraceparent.traceId, sourceTraceparent.parentId, sourceTraceparent.flags) } + : {}; const prefix = `${instructions}\n\nOperator request:\n${request}\n\nSettled action results (bounded; inspect for details):\n`; const prompt = prefix + continuationEvidence(actions, Math.max(0, 8000 - prefix.length)); const item = roomAgentSupervisorQueueItemSchema.parse({ missionId: input.missionId, turn: { @@ -130,7 +139,7 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis agentUserMessageId: `agent_msg:bg-context-${input.continuationId}`, agentAssistantMessageId: `agent_msg:bg-response-${input.continuationId}`, // Resolve the current native thread from the durable session when executing. - agentThreadId: null, traceId: `trace:bg-${input.continuationId}` + agentThreadId: null, traceId: `trace:bg-${input.continuationId}`, ...traceContext } }); try { const enqueued = await store.enqueueRoomAgentMission({ diff --git a/apps/worker/src/span-attributes.ts b/apps/worker/src/span-attributes.ts new file mode 100644 index 000000000..0d699e833 --- /dev/null +++ b/apps/worker/src/span-attributes.ts @@ -0,0 +1,52 @@ +/** + * The only attributes a worker or workflow span may carry. Everything else is + * dropped before export: prompts, model output, file paths, URLs, error + * messages and any other user text never leave the process. String values must + * also look like identifiers (no spaces, slashes beyond `:._-`), numbers must be + * finite. Pure, so the workflow sandbox and the activity side share it. + */ +export const spanAttributeAllowlist = [ + "space.trace_id", + "space.workflow_id", + "space.run_id", + "space.room_id", + "space.pane_id", + "space.mission_id", + "space.workflow_type", + "space.activity_type", + "space.attempt", + "space.status", + "space.outcome", + "space.reason_code", + "space.category", + "space.duration_ms" +] as const; + +const allowed: ReadonlySet = new Set(spanAttributeAllowlist); +const identifierPattern = /^[A-Za-z0-9:._-]{1,200}$/; + +export function allowlistedSpanAttributes(raw: Readonly>): Record { + const attributes: Record = {}; + for (const [key, value] of Object.entries(raw)) { + if (!allowed.has(key)) continue; + if (typeof value === "string" && identifierPattern.test(value)) attributes[key] = value; + else if (typeof value === "number" && Number.isInteger(value)) attributes[key] = value; + } + return attributes; +} + +/** Maps correlation ids (see trace-correlation.ts) to allowlisted span attributes. */ +export function correlationSpanAttributes(correlation: Readonly>): Record { + const attributes: Record = {}; + const names: Record = { + traceId: "space.trace_id", + roomId: "space.room_id", + paneId: "space.pane_id", + missionId: "space.mission_id" + }; + for (const [key, name] of Object.entries(names)) { + const value = correlation[key]; + if (value) attributes[name] = value; + } + return attributes; +} diff --git a/apps/worker/src/trace-correlation.ts b/apps/worker/src/trace-correlation.ts new file mode 100644 index 000000000..573fcb48c --- /dev/null +++ b/apps/worker/src/trace-correlation.ts @@ -0,0 +1,73 @@ +import { parseTraceparent, traceIdFromTraceparent } from "@space/contracts"; +import { logIdentifier } from "./worker-log.js"; + +/** + * Identifier-only correlation read from workflow or activity arguments. Pure + * (no Node APIs), so the workflow interceptors may bundle it into the + * deterministic workflow sandbox. Prompts, content and messages are never read. + */ + +// Keys whose values are identifiers. +const correlationKeys = [ + "traceId", + "traceparent", + "continuesTraceparent", + "requestId", + "roomId", + "paneId", + "missionId", + "roomAgentMissionId", + "agentRunId", + "agentSessionId", + "actionId", + "goalId", + "dotId", + "turnId", + "cliTaskId", + "revisionId" +] as const; +const nestedKeys = ["turn", "item", "bridge", "input", "turnInput"] as const; + +export type CorrelationKey = (typeof correlationKeys)[number]; + +function readCorrelation(value: unknown, into: Partial>, depth: number): void { + if (!value || typeof value !== "object" || Array.isArray(value)) return; + const record = value as Record; + for (const key of correlationKeys) { + if (into[key] !== undefined) continue; + const identifier = key === "traceparent" || key === "continuesTraceparent" + ? (parseTraceparent(record[key]) ? (record[key] as string) : null) + : logIdentifier(record[key]); + if (identifier !== null) into[key] = identifier; + } + if (depth <= 0) return; + for (const key of nestedKeys) readCorrelation(record[key], into, depth - 1); +} + +/** All allowlisted identifier fields found in the arguments (two levels deep). */ +export function readArgumentCorrelation(args: readonly unknown[]): Partial> { + const found: Partial> = {}; + for (const arg of args) readCorrelation(arg, found, 2); + return found; +} + +/** The W3C trace context the API put into the arguments, if any. */ +export function argumentTraceparent(args: readonly unknown[]) { + return parseTraceparent(readArgumentCorrelation(args).traceparent); +} + +/** Allowlisted correlation fields from activity arguments (identifiers only). */ +export function activityCorrelation(args: readonly unknown[]): Record { + const found = readArgumentCorrelation(args); + const correlation: Record = {}; + if (found.traceId) correlation.traceId = found.traceId; + const w3cTraceId = traceIdFromTraceparent(found.traceparent); + if (w3cTraceId) correlation.w3cTraceId = w3cTraceId; + if (found.requestId && found.requestId !== found.traceId) correlation.parentTraceId = found.requestId; + const missionId = found.missionId ?? found.roomAgentMissionId; + if (missionId) correlation.missionId = missionId; + for (const key of ["roomId", "paneId", "agentRunId", "agentSessionId", "actionId", "goalId", "dotId", "turnId", "cliTaskId", "revisionId"] as const) { + if (found[key]) correlation[key] = found[key]!; + } + return correlation; +} diff --git a/apps/worker/src/trace-propagation.ts b/apps/worker/src/trace-propagation.ts new file mode 100644 index 000000000..49b56fc4a --- /dev/null +++ b/apps/worker/src/trace-propagation.ts @@ -0,0 +1,33 @@ +/** + * Temporal header carrying the W3C trace context between workflow, child + * workflow, continue-as-new run and activity. Pure and ASCII only (a + * traceparent is ASCII), so it works inside the workflow sandbox without a + * payload converter or TextEncoder. + */ +export const TRACEPARENT_HEADER = "space-traceparent"; +/** traceparent of the span this run or workflow continues (an OTLP span link). */ +export const LINK_HEADER = "space-trace-link"; + +interface HeaderPayload { + metadata?: Record | null; + data?: Uint8Array | null; +} + +function ascii(value: string): Uint8Array { + const bytes = new Uint8Array(value.length); + for (let i = 0; i < value.length; i += 1) bytes[i] = value.charCodeAt(i) & 0x7f; + return bytes; +} + +export function encodeTraceHeader(value: string): HeaderPayload { + return { metadata: { encoding: ascii("binary/plain") }, data: ascii(value) }; +} + +export function decodeTraceHeader(headers: Record | null | undefined, name: string): string | null { + const payload = headers?.[name] as HeaderPayload | undefined; + const data = payload?.data; + if (!data || data.length === 0 || data.length > 128) return null; + let text = ""; + for (let i = 0; i < data.length; i += 1) text += String.fromCharCode(data[i]!); + return text; +} diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 896269754..86e3b5fb6 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -6,7 +6,7 @@ import { StreamingCredentialStore } from "@space/streaming"; import { TelegramDeliveryWorker, TelegramSecretStore, otlpSpanExporterFromEnv } from "@space/runtime"; import { NativeConnection, Runtime, Worker } from "@temporalio/worker"; import * as activities from "./activities.js"; -import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { tracingWorkerOptions } from "./workflow-span-sinks.js"; import { resolveTemporalCoreMetricsBind, resolveWorkerMetricsListen, startWorkerMetricsServer } from "./worker-metrics.js"; import { workerLog } from "./worker-log.js"; import { sweepCodexCliCompletions } from "./codex-cli-completion.js"; @@ -43,9 +43,9 @@ const worker = await Worker.create({ taskQueue, workflowsPath, activities, - // Structured, correlated activity records and worker metrics. Activity-side - // only: the workflow sandbox and its replay behaviour are unchanged. - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + // Structured, correlated activity records and worker metrics, plus OTLP + // workflow and activity spans when SPACE_OTEL_EXPORTER_OTLP_ENDPOINT is set. + ...tracingWorkerOptions(spans) }); // Optional loopback-only Prometheus endpoint for worker metrics. diff --git a/apps/worker/src/workflow-interceptors.ts b/apps/worker/src/workflow-interceptors.ts new file mode 100644 index 000000000..c5a31fb55 --- /dev/null +++ b/apps/worker/src/workflow-interceptors.ts @@ -0,0 +1,139 @@ +import { + ContinueAsNew, + isCancellation, + proxySinks, + uuid4, + workflowInfo, + type Sinks, + type WorkflowInterceptorsFactory +} from "@temporalio/workflow"; +import { categorizeFailureCode, formatTraceparent, isSafeErrorCode, parseTraceparent, type FailureCategory } from "@space/contracts"; +import { readArgumentCorrelation } from "./trace-correlation.js"; +import { LINK_HEADER, TRACEPARENT_HEADER, decodeTraceHeader, encodeTraceHeader } from "./trace-propagation.js"; + +/** + * Workflow-side tracing. Workflow code does no I/O for it: the interceptors + * only compute ids (deterministic `uuid4`, workflow time) and hand one plain + * record per workflow run to a sink, which the worker process turns into an + * OTLP span. The sink is registered with `callDuringReplay: false`, so replay + * never emits a span again (workflow-span-sinks.ts). The trace context travels + * to activities, child workflows and continue-as-new runs in a Temporal header. + * + * This file is loaded into the workflow sandbox through + * `interceptors.workflowModules`; keep its imports free of Node APIs. + */ + +export type WorkflowSpanStatus = "COMPLETED" | "FAILED" | "CANCELLED" | "CONTINUED_AS_NEW"; + +export interface WorkflowSpanData { + traceId: string; + spanId: string; + parentSpanId: string; + link: { traceId: string; spanId: string } | null; + startMs: number; + endMs: number; + status: WorkflowSpanStatus; + reasonCode: string | null; + category: FailureCategory | null; + /** Identifier-only correlation read from the workflow input (traceId, roomId, paneId, missionId). */ + correlation: Record; +} + +export interface WorkflowSpanSinks extends Sinks { + spans: { + recordSpan(data: WorkflowSpanData): void; + }; +} + +function newSpanId(): string { + const id = uuid4().replace(/-/g, "").slice(0, 16); + return /^0+$/.test(id) ? "0000000000000001" : id; +} + +function failureCode(error: unknown): string | null { + let current = error as { type?: unknown; code?: unknown; cause?: unknown } | null | undefined; + for (let depth = 0; current && typeof current === "object" && depth < 5; depth += 1) { + if (isSafeErrorCode(current.code)) return current.code; + if (isSafeErrorCode(current.type)) return current.type; + current = current.cause as typeof current; + } + return null; +} + +export const interceptors: WorkflowInterceptorsFactory = () => { + const { spans } = proxySinks(); + let active: { traceId: string; spanId: string; flags: string; parentTraceparent: string } | null = null; + + const propagate = (headers: Record): Record => + active ? { ...headers, [TRACEPARENT_HEADER]: encodeTraceHeader(formatTraceparent(active.traceId, active.spanId, active.flags)) } : headers; + + return { + inbound: [{ + async execute(input, next) { + const found = readArgumentCorrelation(input.args); + const fromInput = parseTraceparent(found.traceparent); + const continues = parseTraceparent(found.continuesTraceparent); + const fromHeader = parseTraceparent(decodeTraceHeader(input.headers, TRACEPARENT_HEADER)); + const headerLink = parseTraceparent(decodeTraceHeader(input.headers, LINK_HEADER)); + const memoTraceparent = workflowInfo().memo?.traceparent; + const fromMemo = parseTraceparent(memoTraceparent); + // A continue-as-new run keeps its parent from the header; a background + // continuation starts the trace named by its own input and links back; + // otherwise the parent workflow's span, then the API request's span. + const parent = headerLink ? fromHeader ?? fromInput ?? fromMemo + : continues && fromInput ? fromInput + : fromHeader ?? fromInput ?? fromMemo; + if (!parent) return next(input); + const linked = headerLink ?? continues; + const traceId = parent.traceId; + const spanId = newSpanId(); + const parentTraceparent = formatTraceparent(parent.traceId, parent.parentId, parent.flags); + active = { traceId, spanId, flags: parent.flags, parentTraceparent }; + const startMs = Date.now(); + const finish = (status: WorkflowSpanStatus, error: unknown) => { + const reasonCode = status === "FAILED" ? failureCode(error) : null; + spans.recordSpan({ + traceId, + spanId, + parentSpanId: parent.parentId, + link: linked ? { traceId: linked.traceId, spanId: linked.parentId } : null, + startMs, + endMs: Date.now(), + status, + reasonCode, + category: status === "FAILED" ? categorizeFailureCode(reasonCode) : null, + correlation + }); + }; + const correlation: Record = {}; + const missionId = found.missionId ?? found.roomAgentMissionId; + for (const [key, value] of [["traceId", found.traceId], ["roomId", found.roomId], ["paneId", found.paneId], ["missionId", missionId]] as const) { + if (value) correlation[key] = value; + } + try { + const result = await next(input); + finish("COMPLETED", null); + return result; + } catch (error) { + finish(error instanceof ContinueAsNew ? "CONTINUED_AS_NEW" : isCancellation(error) ? "CANCELLED" : "FAILED", error); + throw error; + } + } + }], + outbound: [{ + scheduleActivity: (input, next) => next({ ...input, headers: propagate(input.headers) as typeof input.headers }), + scheduleLocalActivity: (input, next) => next({ ...input, headers: propagate(input.headers) as typeof input.headers }), + startChildWorkflowExecution: (input, next) => next({ ...input, headers: propagate(input.headers) as typeof input.headers }), + continueAsNew: (input, next) => next({ + ...input, + headers: active + ? { + ...input.headers, + [TRACEPARENT_HEADER]: encodeTraceHeader(active.parentTraceparent), + [LINK_HEADER]: encodeTraceHeader(formatTraceparent(active.traceId, active.spanId, active.flags)) + } as typeof input.headers + : input.headers + }) + }] + }; +}; diff --git a/apps/worker/src/workflow-span-sinks.ts b/apps/worker/src/workflow-span-sinks.ts new file mode 100644 index 000000000..6c57d13d0 --- /dev/null +++ b/apps/worker/src/workflow-span-sinks.ts @@ -0,0 +1,70 @@ +import { existsSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import type { InjectedSinks } from "@temporalio/worker"; +import type { SpanExporter } from "@space/runtime"; +import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { allowlistedSpanAttributes, correlationSpanAttributes } from "./span-attributes.js"; +import type { WorkflowSpanSinks } from "./workflow-interceptors.js"; + +/** + * Worker options for tracing: activity interceptor (always, for logs and + * metrics), plus workflow interceptors and sinks only when span export is on. + * Spread into Worker.create; with no exporter nothing is added to workflows. + */ +export function tracingWorkerOptions(spans: SpanExporter | null | undefined) { + const sinks = createWorkflowSpanSinks(spans); + let interceptorModule = fileURLToPath(new URL("./workflow-interceptors.js", import.meta.url)); + // Running from the TypeScript source (tests, worktree runs) has no built .js. + if (!existsSync(interceptorModule)) interceptorModule = interceptorModule.replace(/\.js$/, ".ts"); + return { + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + ...(sinks ? { workflowModules: [interceptorModule] } : {}) + }, + ...(sinks ? { sinks } : {}) + }; +} + +/** + * Worker-side half of workflow tracing: turns the record the workflow + * interceptor emits into an OTLP INTERNAL span. `callDuringReplay: false` + * means replaying a history never exports a second span. The sink never + * throws, so a broken exporter cannot fail a workflow task. + */ +export function createWorkflowSpanSinks(spans: SpanExporter | null | undefined): InjectedSinks | undefined { + if (!spans) return undefined; + return { + spans: { + recordSpan: { + callDuringReplay: false, + fn(info, data) { + try { + spans.record({ + traceId: data.traceId, + spanId: data.spanId, + parentSpanId: data.parentSpanId, + name: `workflow ${info.workflowType}`, + startTimeUnixNano: BigInt(data.startMs) * 1_000_000n, + endTimeUnixNano: BigInt(Math.max(data.startMs, data.endMs)) * 1_000_000n, + attributes: allowlistedSpanAttributes({ + ...correlationSpanAttributes(data.correlation), + "space.workflow_id": info.workflowId, + "space.run_id": info.runId, + "space.workflow_type": info.workflowType, + "space.status": data.status, + "space.reason_code": data.reasonCode, + "space.category": data.category, + "space.duration_ms": Math.max(0, data.endMs - data.startMs) + }), + error: data.status === "FAILED", + kind: 1, + ...(data.link ? { links: [data.link] } : {}) + }); + } catch { + // Tracing must never fail workflow work. + } + } + } + } + }; +} diff --git a/apps/worker/tests/activity-observability.test.ts b/apps/worker/tests/activity-observability.test.ts index 654bf7962..31193acde 100644 --- a/apps/worker/tests/activity-observability.test.ts +++ b/apps/worker/tests/activity-observability.test.ts @@ -186,7 +186,7 @@ it("registers the activity interceptor on every Temporal worker, including both const { readFile } = await import("node:fs/promises"); for (const entry of ["worker.ts", "memory-maintenance-worker.ts", "memory-mutation-worker.ts"]) { const source = await readFile(new URL(`../src/${entry}`, import.meta.url), "utf8"); - expect(source, entry).toMatch(/interceptors: \{ activity: \[createActivityObservabilityInterceptors\(\{ spans \}\)\] \}/); + expect(source, entry).toMatch(/\.\.\.tracingWorkerOptions\(spans\)/); } }); diff --git a/apps/worker/tests/support/tracing-workflows.ts b/apps/worker/tests/support/tracing-workflows.ts new file mode 100644 index 000000000..19c916efa --- /dev/null +++ b/apps/worker/tests/support/tracing-workflows.ts @@ -0,0 +1,20 @@ +// Workflows for workflow-tracing.test.ts. They import only @temporalio/workflow +// so the workflow bundle stays small; the activity is a stub. +import { continueAsNew, executeChild, proxyActivities } from "@temporalio/workflow"; + +const { echoStep } = proxyActivities<{ echoStep(input: unknown): Promise }>({ startToCloseTimeout: "10 seconds" }); + +export async function childWorkflow(input: { traceId: string }): Promise { + return echoStep(input); +} + +export async function parentWorkflow(input: { traceId: string; roomId: string; prompt: string }): Promise { + await echoStep(input); + await executeChild(childWorkflow, { args: [{ traceId: input.traceId }] }); + return "done"; +} + +export async function continuingWorkflow(input: { traceId: string; round: number }): Promise { + if (input.round === 0) await continueAsNew({ ...input, round: 1 }); + return echoStep(input); +} diff --git a/apps/worker/tests/workflow-tracing.test.ts b/apps/worker/tests/workflow-tracing.test.ts new file mode 100644 index 000000000..bcb256a01 --- /dev/null +++ b/apps/worker/tests/workflow-tracing.test.ts @@ -0,0 +1,195 @@ +import { createServer, type Server } from "node:http"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { MetricsRegistry, createOtlpSpanExporter, type ServerSpan, type SpanExporter } from "@space/runtime"; +import { createWorkerMetrics } from "../src/worker-metrics.js"; +import { observeActivityExecution } from "../src/activity-observability.js"; +import { allowlistedSpanAttributes, spanAttributeAllowlist } from "../src/span-attributes.js"; +import { LINK_HEADER, TRACEPARENT_HEADER, decodeTraceHeader, encodeTraceHeader } from "../src/trace-propagation.js"; +import { createWorkflowSpanSinks, tracingWorkerOptions } from "../src/workflow-span-sinks.js"; + +const traceId = "4bf92f3577b34da6a3ce929d0e0e4736"; +const apiSpanId = "00f067aa0ba902b7"; +const traceparent = `00-${traceId}-${apiSpanId}-01`; + +function recordingExporter(): SpanExporter & { spans: ServerSpan[] } { + const spans: ServerSpan[] = []; + return { spans, record: (span) => { spans.push(span); }, flush: async () => undefined, close: async () => undefined, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; +} + +describe("span attribute allowlist", () => { + it("drops prompts, paths, URLs, error messages and unknown keys", () => { + const attributes = allowlistedSpanAttributes({ + "space.room_id": "room:1", + "space.attempt": 2, + "space.reason_code": "TOOL_UNAVAILABLE", + "space.status": "this is a free text error message", + "space.workflow_id": "https://example.test/secret?token=1", + "space.pane_id": "/home/user/project/file.ts", + "space.prompt": "SECRET prompt", + "space.duration_ms": 1.5, + "http.url": "http://x" + }); + expect(attributes).toEqual({ "space.room_id": "room:1", "space.attempt": 2, "space.reason_code": "TOOL_UNAVAILABLE" }); + expect(Object.keys(attributes).every((key) => (spanAttributeAllowlist as readonly string[]).includes(key))).toBe(true); + }); + + it("exports only allowlisted attributes for activity spans", async () => { + const exporter = recordingExporter(); + const metrics = createWorkerMetrics(new MetricsRegistry()); + const info = { activityType: "runTurn", activityId: "1", attempt: 3, currentAttemptScheduledTimestampMs: 1, workflowType: "parentWorkflow", workflowExecution: { workflowId: "wf:1", runId: "run-1" } } as never; + const prompt = "SECRET prompt text /home/user/file"; + await observeActivityExecution(info, [{ traceId: "req:abc", roomId: "room:1", paneId: "pane:1", prompt, turn: { prompt, traceparent } }], + async () => ({ status: "FAILED", error: prompt }), { metrics, spans: exporter }); + const [span] = exporter.spans; + expect(span).toBeDefined(); + expect(Object.keys(span!.attributes).every((key) => (spanAttributeAllowlist as readonly string[]).includes(key))).toBe(true); + expect(JSON.stringify(span!.attributes)).not.toContain("SECRET"); + expect(span!.attributes).toMatchObject({ "space.attempt": 3, "space.outcome": "FAILED_RESULT", "space.reason_code": "ACTIVITY_RESULT_FAILED", + "space.room_id": "room:1", "space.pane_id": "pane:1", "space.workflow_id": "wf:1", "space.run_id": "run-1" }); + }); +}); + +describe("trace propagation", () => { + it("round-trips a traceparent through a Temporal header payload", () => { + const headers = { [TRACEPARENT_HEADER]: encodeTraceHeader(traceparent) }; + expect(decodeTraceHeader(headers, TRACEPARENT_HEADER)).toBe(traceparent); + expect(decodeTraceHeader(headers, LINK_HEADER)).toBeNull(); + }); + + it("parents an activity span under the workflow span named by the header", async () => { + const exporter = recordingExporter(); + const metrics = createWorkerMetrics(new MetricsRegistry()); + const info = { activityType: "runTurn", activityId: "1", attempt: 1, currentAttemptScheduledTimestampMs: 1 } as never; + const workflowSpanId = "aaaaaaaaaaaaaaaa"; + await observeActivityExecution(info, [{ traceparent }], async () => "ok", + { metrics, spans: exporter, traceparentHeader: `00-${traceId}-${workflowSpanId}-01` }); + expect(exporter.spans[0]).toMatchObject({ traceId, parentSpanId: workflowSpanId }); + }); +}); + +describe("workflow span sink", () => { + it("is registered with callDuringReplay false so replay emits no spans", () => { + expect(createWorkflowSpanSinks(null)).toBeUndefined(); + const sinks = createWorkflowSpanSinks(recordingExporter())!; + expect(sinks.spans!.recordSpan!.callDuringReplay).toBe(false); + }); + + it("adds no workflow interceptors or sinks when export is off", () => { + const options = tracingWorkerOptions(null); + expect(options.interceptors).not.toHaveProperty("workflowModules"); + expect(options).not.toHaveProperty("sinks"); + }); + + it("records a span with a link and only allowlisted attributes", () => { + const exporter = recordingExporter(); + const sinks = createWorkflowSpanSinks(exporter)!; + sinks.spans!.recordSpan!.fn({ workflowType: "parentWorkflow", workflowId: "wf:1", runId: "run-2" } as never, { + traceId, spanId: "bbbbbbbbbbbbbbbb", parentSpanId: apiSpanId, link: { traceId, spanId: "cccccccccccccccc" }, + startMs: 1000, endMs: 1500, status: "FAILED", reasonCode: "TOOL_UNAVAILABLE", category: "TOOL_UNAVAILABLE", + correlation: { traceId: "req:abc", roomId: "room:1", prompt: "SECRET" } + }); + expect(exporter.spans[0]).toMatchObject({ name: "workflow parentWorkflow", parentSpanId: apiSpanId, error: true, kind: 1, + links: [{ traceId, spanId: "cccccccccccccccc" }], startTimeUnixNano: 1_000_000_000n, endTimeUnixNano: 1_500_000_000n }); + expect(JSON.stringify(exporter.spans[0]!.attributes)).not.toContain("SECRET"); + expect(exporter.spans[0]!.attributes).toMatchObject({ "space.workflow_id": "wf:1", "space.run_id": "run-2", "space.room_id": "room:1", "space.duration_ms": 500 }); + }); +}); + +// Real Temporal test server + fake OTLP collector. Needs @temporalio/testing and +// the dev server it downloads; skipped (not failed) when they are unavailable. +interface OtlpSpan { traceId: string; spanId: string; parentSpanId?: string; name: string; links?: Array<{ traceId: string; spanId: string }>; attributes: Array<{ key: string; value: Record }> } + +describe("Temporal integration: one connected trace", () => { + const closers: Array<() => Promise> = []; + afterEach(async () => { for (const close of closers.splice(0).reverse()) await close().catch(() => undefined); }); + + it("exports workflow, child workflow and activity spans under the API span and replays silently", async (context) => { + let testing: typeof import("@temporalio/testing"); + let env: Awaited>; + try { + testing = await import("@temporalio/testing"); + env = await testing.TestWorkflowEnvironment.createLocal(); + } catch { + context.skip(); + return; + } + closers.push(() => env.teardown()); + const { Worker } = await import("@temporalio/worker"); + + const received: OtlpSpan[] = []; + const collector: Server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + if (request.url === "/v1/traces") { + const body = JSON.parse(Buffer.concat(chunks).toString("utf8")); + for (const resource of body.resourceSpans) for (const scope of resource.scopeSpans) received.push(...scope.spans); + } + response.writeHead(200).end("{}"); + }); + }); + await new Promise((resolve) => collector.listen(0, "127.0.0.1", resolve)); + closers.push(() => new Promise((resolve) => collector.close(resolve))); + const port = (collector.address() as { port: number }).port; + const exporter = createOtlpSpanExporter({ endpoint: `http://127.0.0.1:${port}`, serviceName: "space-worker-test", flushIntervalMs: 60_000 }); + closers.push(() => exporter.close()); + + const workflowsPath = fileURLToPath(new URL("./support/tracing-workflows.ts", import.meta.url)); + const contractsSource = fileURLToPath(new URL("../../../packages/contracts/src/index.ts", import.meta.url)); + const bundlerOptions = { + webpackConfigHook: (config: { resolve?: Record }) => ({ + ...config, + resolve: { ...config.resolve, extensionAlias: { ".js": [".ts", ".js"] }, alias: { "@space/contracts$": contractsSource } } + }) + }; + const tracing = tracingWorkerOptions(exporter); + const taskQueue = `tracing-${Date.now()}`; + const worker = await Worker.create({ + connection: env.nativeConnection, + taskQueue, + workflowsPath, + bundlerOptions, + activities: { echoStep: async () => "ok" }, + ...tracing + } as never); + + const spans = () => received.map((span) => ({ ...span, attrs: Object.fromEntries(span.attributes.map((a) => [a.key, Object.values(a.value)[0]])) })); + const input = { traceId: "req:abc123", roomId: "room:1", prompt: "SECRET prompt text", traceparent }; + const [parentHandle, continuedHandle] = await worker.runUntil(async () => { + const parent = await env.client.workflow.start("parentWorkflow", { taskQueue, workflowId: `parent-${taskQueue}`, args: [input] }); + await parent.result(); + const continued = await env.client.workflow.start("continuingWorkflow", { taskQueue, workflowId: `continue-${taskQueue}`, args: [{ traceId: "req:abc123", round: 0, traceparent }] }); + await continued.result(); + return [parent, continued] as const; + }); + await exporter.flush(); + + const all = spans(); + expect(all.every((span) => span.traceId === traceId)).toBe(true); + const workflowSpans = all.filter((span) => span.name.startsWith("workflow ")); + const parentSpan = workflowSpans.find((span) => span.name === "workflow parentWorkflow")!; + const childSpan = workflowSpans.find((span) => span.name === "workflow childWorkflow")!; + expect(parentSpan.parentSpanId).toBe(apiSpanId); + expect(childSpan.parentSpanId).toBe(parentSpan.spanId); + const activitySpans = all.filter((span) => span.name === "activity echoStep"); + expect(activitySpans.filter((span) => span.parentSpanId === parentSpan.spanId)).toHaveLength(1); + expect(activitySpans.filter((span) => span.parentSpanId === childSpan.spanId)).toHaveLength(1); + expect(JSON.stringify(received)).not.toContain("SECRET"); + + // Continue-as-new: the second run links to the first run's span. + const runs = workflowSpans.filter((span) => span.name === "workflow continuingWorkflow"); + expect(runs).toHaveLength(2); + const linked = runs.find((span) => span.links?.length)!; + const first = runs.find((span) => span !== linked)!; + expect(linked.links![0]!.spanId).toBe(first.spanId); + + // Replaying a recorded history must not export anything. + const before = received.length; + const history = await parentHandle.fetchHistory(); + await Worker.runReplayHistory({ workflowsPath, bundlerOptions, ...tracing } as never, history, parentHandle.workflowId); + await exporter.flush(); + expect(received.length).toBe(before); + void continuedHandle; + }, 120_000); +}); diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index e432af4c6..bfb3c78ac 100644 --- a/docs/configuration-reference.md +++ b/docs/configuration-reference.md @@ -52,3 +52,23 @@ background work, and is hidden in the web app. Public installs upgraded from 1.1.x that used Job Copilot, the games, channels, Telegram or the desktop routes set the modules they keep, for example `SPACE_OPTIONAL_MODULES=job-copilot,games`. + +## OTLP span export + +`SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is read by the API and by every Temporal +worker process (`space-worker`, `space-memory-maintenance-worker`, +`space-memory-mutation-worker`). It is a collector base URL such as +`http://127.0.0.1:4318`; spans are posted as OTLP/HTTP JSON to +`/v1/traces`. Unset, empty or invalid (credentials, query or fragment +in the URL), export is off and nothing is added to the process or to workflows. + +With it set, one API request that starts work on Temporal gives one trace: the +API SERVER span, a span for each workflow run (child workflows nest under their +parent), and one span per activity attempt. A continue-as-new run and a Room +Agent background continuation carry an OTLP span link to the run or trace they +continue. Spans carry only ids (trace, workflow, run, room, pane, mission), +workflow and activity type, attempt, status, error code and category, and +duration; prompts, model output, paths, URLs and error messages are never +exported. Export is best effort: a bounded queue drops (and counts) spans, each +POST times out, and a slow or failing collector never delays or fails work. +See [observability](quality/observability.md#traces-w3c-trace-context). diff --git a/docs/quality/observability.md b/docs/quality/observability.md index 488fc3036..d0fc3519d 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -220,8 +220,30 @@ SERVER span per API request when `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is set (no OpenTelemetry dependency; `packages/runtime/src/otlp-span-exporter.ts`). Since 2026-10-16 the worker and both memory workers use the same exporter (services `space-worker`, `space-memory-maintenance-worker`, `space-memory-mutation-worker`): -one INTERNAL span per activity whose input carries a `traceparent`, as a child of -that span, with the activity type, outcome and attempt only. +one INTERNAL span per activity attempt, with the activity type, outcome, attempt, +`reasonCode` / `category` and duration. The workers also export one INTERNAL span +per workflow run, so one API request gives one connected trace: API SERVER span, +workflow span (parent = the propagated `traceparent` span, which is the API +request's own span), child workflow spans under their parent workflow, and +activity spans under their workflow span. +- **How:** Temporal interceptors, no tracing dependency. The workflow + interceptor (`apps/worker/src/workflow-interceptors.ts`) does no I/O: it + computes ids with the deterministic `uuid4` and workflow time, puts the trace + context into a Temporal header for activities, child workflows and + continue-as-new, and hands one plain record per run to a sink. The sink + (`workflow-span-sinks.ts`) is registered with `callDuringReplay: false`, so + replay never exports a span again. A workflow span is exported when the run + ends (completes, fails, is cancelled or continues as new), so a long-running + workflow shows up when it finishes. +- **Links:** a continue-as-new run links to the previous run's span; a Room + Agent background continuation (`trace:bg-`) is its own trace and links to + the span it continues (`continuesTraceparent` in the turn input). +- **Attributes** come from an allowlist (`apps/worker/src/span-attributes.ts`): + trace, workflow, run, room, pane and mission ids, workflow and activity type, + attempt, status or outcome, `reasonCode`, `category` and duration. Prompts, + model output, paths, URLs and error messages are never exported. +- Workflows started without a `traceparent` (for example the scheduled memory + maintenance) produce no spans. - **API:** a valid inbound `traceparent` is continued (same trace ID, new span). An invalid or missing header gets a new trace ID. Every API log line carries `w3cTraceId` (Fastify `childLoggerFactory`). @@ -336,9 +358,11 @@ No credential or transcript export was added. the lookup also returns the Space-native agent runs (`space_agent_runs`) of the trace's workflows, joined by workflow id, with their status and error code, and names a failed run's code as the cause when no event carries one. -- **Hosts emit no spans.** Since 2026-10-16 the OTLP exporter sends API SERVER - spans and worker activity spans (the worker and both memory workers); the CLI - and Browser Hosts still join traces through log lines only. +- **Hosts emit no spans.** The OTLP exporter sends API SERVER spans and worker + workflow and activity spans (the worker and both memory workers); the CLI and + Browser Hosts still join traces through log lines only. A workflow span is + exported when its run ends, and a background continuation is a separate trace + joined to its source by a span link, not by parentage. - **Recovery counters are partial.** Recovery counters exist for plan model retries, CLI terminal reconnects and worker activity retries only. - **Worker endpoints are not scraped yet.** Stable's `deploy/space-worker.service` diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index 696eb6d84..5e0ab16fd 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -6807,6 +6807,9 @@ export const dummyTurnInputSchema = z.object({ // Optional W3C trace context of the originating request (additive; see // docs/quality/observability.md). Never part of a workflow id. traceparent: traceparentSchema.optional(), + // Background continuations start their own trace; this names the span they + // continue so the exporter can emit an OTLP span link (additive). + continuesTraceparent: traceparentSchema.optional(), // Space Dots Stage 3 (additive): a gateway-enabled native Chat turn. dotId: dotIdSchema.optional(), toolGateway: z.literal("space-dot-tools-v1").optional(), diff --git a/packages/runtime/src/otlp-span-exporter.ts b/packages/runtime/src/otlp-span-exporter.ts index 86a0edf05..800751d57 100644 --- a/packages/runtime/src/otlp-span-exporter.ts +++ b/packages/runtime/src/otlp-span-exporter.ts @@ -4,8 +4,9 @@ * `/v1/traces`). No dependency, no collector required: Space logs stay * joinable by W3C trace id without it. * - * Spans carry only the method, route template, status code, service name and - * Space request id; never URLs with parameters, headers, bodies or user data. + * Spans carry only the method, route template, status code, service name, + * identifiers, types, attempt, outcome and error code (see the worker's span + * attribute allowlist); never URLs with parameters, headers, bodies or user data. * Export is best effort: a bounded queue drops (and counts) spans instead of * growing, each POST has a timeout, and a failing collector never affects a * request. @@ -22,6 +23,8 @@ export interface ServerSpan { error: boolean; /** OTLP span kind; SERVER (2) unless given. Worker activities use INTERNAL (1). */ kind?: number; + /** OTLP span links: continue-as-new runs and background continuations link to what they continue. */ + links?: readonly { traceId: string; spanId: string }[]; } export interface SpanExporterStats { @@ -82,6 +85,7 @@ export function otlpTraceRequest(spans: readonly ServerSpan[], serviceName: stri ...(span.parentSpanId ? { parentSpanId: span.parentSpanId } : {}), name: span.name, kind: span.kind ?? 2, + ...(span.links?.length ? { links: span.links.map(link => ({ traceId: link.traceId, spanId: link.spanId })) } : {}), startTimeUnixNano: span.startTimeUnixNano.toString(), endTimeUnixNano: span.endTimeUnixNano.toString(), attributes: Object.entries(span.attributes).map(([key, value]) => ({ key, value: attributeValue(value) })),