Export Temporal workflow spans over OTLP: one trace from API request to every activity

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. */

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<Record<CorrelationKey, string>>, depth: number): void {
-  if (!value || typeof value !== "object" || Array.isArray(value)) return;
-  const record = value as Record<string, unknown>;
-  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<string, string> {
-  const found: Partial<Record<CorrelationKey, string>> = {};
-  for (const arg of args) readCorrelation(arg, found, 2);
-  const correlation: Record<string, string> = {};
-  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<Record<CorrelationKey, string>> = {};
-  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<T>(
@@ -194,19 +143,36 @@ export async function observeActivityExecution<T>(
       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<Obser
     inbound: {
       execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), {
         ...dependencies,
+        traceparentHeader: decodeTraceHeader(input.headers, TRACEPARENT_HEADER),
         cancelled: () => context.cancellationSignal.aborted
       })
     }

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,

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 })) });

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-<id>) 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({

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<string> = new Set(spanAttributeAllowlist);
+const identifierPattern = /^[A-Za-z0-9:._-]{1,200}$/;
+
+export function allowlistedSpanAttributes(raw: Readonly<Record<string, unknown>>): Record<string, string | number> {
+  const attributes: Record<string, string | number> = {};
+  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<string, string | undefined>>): Record<string, string> {
+  const attributes: Record<string, string> = {};
+  const names: Record<string, string> = {
+    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;
+}

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<Record<CorrelationKey, string>>, depth: number): void {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return;
+  const record = value as Record<string, unknown>;
+  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<Record<CorrelationKey, string>> {
+  const found: Partial<Record<CorrelationKey, string>> = {};
+  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<string, string> {
+  const found = readArgumentCorrelation(args);
+  const correlation: Record<string, string> = {};
+  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;
+}

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<string, Uint8Array> | 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<string, unknown> | 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;
+}