Connect API, Temporal workflows and activity attempts in OTLP traces

apps/api/src/trace-context.ts

@@ -96,7 +96,8 @@ export const requestTraceContexts = new TraceContextRegistry();
 
 /** traceparent for work started on behalf of this Space request id, if known. */
 export function traceparentForTraceId(traceId: string | null | undefined, registry = requestTraceContexts): string | null {
-  return childTraceparent(registry.lookup(traceId)?.traceparent ?? null);
+  // Propagate the recorded SERVER span, not an unrecorded intermediate span.
+  return registry.lookup(traceId)?.traceparent ?? null;
 }
 
 /** Adds the request's traceparent to a turn input unless it already has a valid one. */

apps/api/tests/trace-correlation.test.ts

@@ -50,6 +50,7 @@ describe("request trace context", () => {
     registry.remember("req:dddddddddddd", createRequestTraceContext(inbound));
     const filled = withTurnTraceparent({ traceId: "req:dddddddddddd" }, registry);
     expect(parseTraceparent(filled.traceparent)?.traceId).toBe(inboundTraceId);
+    expect(filled.traceparent).toBe(registry.lookup("req:dddddddddddd")?.traceparent);
     const kept = withTurnTraceparent({ traceId: "req:dddddddddddd", traceparent: inbound }, registry);
     expect(kept.traceparent).toBe(inbound);
     expect(withTurnTraceparent({ traceId: "req:unknown00000" }, registry)).toEqual({ traceId: "req:unknown00000" });

apps/api/vitest.config.ts

@@ -8,6 +8,7 @@ const __dirname = path.dirname(fileURLToPath(import.meta.url));
 export default defineConfig({
   resolve: {
     alias: {
+      "@space/contracts/observability": resolve(__dirname, "../../packages/contracts/src/observability.ts"),
       "@space/cli-host": resolve(__dirname, "../../packages/cli-host/src/index.ts"),
       "@space/db": resolve(__dirname, "../../packages/db/src/index.ts"),
       "@space/gateway": resolve(__dirname, "../../packages/gateway/src/index.ts"),

apps/worker/package.json

@@ -26,6 +26,7 @@
     "undici": "8.10.2"
   },
   "devDependencies": {
+    "@temporalio/testing": "1.24.0",
     "typescript": "^5.9.3",
     "vitest": "^4.1.12"
   },

apps/worker/src/activity-observability.ts

@@ -1,3 +1,4 @@
+import { traceFromHeaders, safeTraceAttributes } from "./workflow-trace-context.js";
 import { AsyncLocalStorage } from "node:async_hooks";
 import { randomBytes } from "node:crypto";
 import type { Context } from "@temporalio/activity";
@@ -137,6 +138,7 @@ export interface ObserveActivityDependencies {
   cancelled?: () => boolean;
   /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */
   spans?: SpanExporter | null;
+  traceparent?: string;
 }
 
 function activityTraceparent(args: readonly unknown[]) {
@@ -194,19 +196,22 @@ export async function observeActivityExecution<T>(
       queueMs,
       rssBytes: process.memoryUsage().rss
     });
-    const parent = dependencies.spans ? activityTraceparent(args) : null;
+    const parent = dependencies.spans ? (parseTraceparent(dependencies.traceparent) ?? activityTraceparent(args)) : null;
     if (parent) {
-      dependencies.spans!.record({
+      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: { "space.activity_type": activityType, "space.outcome": outcome, "space.attempt": info.attempt },
+        attributes: safeTraceAttributes({ "space.activity_type": activityType, "space.outcome": outcome, "space.attempt": info.attempt,
+          "space.activity_id": info.activityId, "space.workflow_id": info.workflowExecution?.workflowId,
+          "space.workflow_run_id": info.workflowExecution?.runId, "space.workflow_type": info.workflowType,
+          "space.reason_code": reasonCode, "space.category": category, "space.duration_ms": durationMs }),
         error: outcome === "FAILED" || outcome === "FAILED_RESULT",
         kind: 1
-      });
+      }); } catch { /* Tracing must not change the activity result or retry policy. */ }
     }
   };
 
@@ -229,6 +234,7 @@ export function createActivityObservabilityInterceptors(dependencies: Omit<Obser
     inbound: {
       execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), {
         ...dependencies,
+        traceparent: traceFromHeaders(input.headers) ?? dependencies.traceparent,
         cancelled: () => context.cancellationSignal.aborted
       })
     }

apps/worker/src/memory-maintenance-worker.ts

@@ -1,3 +1,4 @@
+import { workflowTracingOptions } from "./workflow-tracing-worker.js";
 import { fileURLToPath } from "node:url";
 import { NativeConnection, Worker } from "@temporalio/worker";
 import * as activities from "./memory-maintenance-activities.js";
@@ -30,7 +31,8 @@ if (!memoryMaintenanceEnabled(process.env)) {
       workflowsPath,
       activities,
       ...memoryMaintenanceWorkerOptions,
-      interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+      interceptors: { activity: [createActivityObservabilityInterceptors({ spans })], workflowModules: workflowTracingOptions(spans).workflowModules },
+      sinks: workflowTracingOptions(spans).sinks
     });
     console.log(JSON.stringify({
       service,
@@ -41,9 +43,12 @@ if (!memoryMaintenanceEnabled(process.env)) {
       activityConcurrency: memoryMaintenanceWorkerOptions.maxConcurrentActivityTaskExecutions
     }));
     const stopReloadWatch = reloadWhenServicesCommitChanges(worker, { log: (message) => console.log(JSON.stringify({ service, status: "reloading", message })) });
-    await worker.run();
-    stopReloadWatch();
-    await spans?.close();
+    try {
+      await worker.run();
+    } finally {
+      stopReloadWatch();
+      await spans?.close();
+    }
   } catch (error) {
     console.error(JSON.stringify({
       service,

apps/worker/src/memory-mutation-worker.ts

@@ -1,3 +1,4 @@
+import { workflowTracingOptions } from "./workflow-tracing-worker.js";
 import { fileURLToPath } from "node:url";
 import { NativeConnection, Worker } from "@temporalio/worker";
 import { executePersistedMemoryChangeSet } from "./memory-mutation-activity.js";
@@ -23,7 +24,8 @@ if (!memoryMutationsEnabled(process.env)) {
     workflowsPath: fileURLToPath(new URL("./memory-mutation-workflows.js", import.meta.url)),
     activities: { executePersistedMemoryChangeSet },
     ...memoryMutationWorkerOptions,
-    interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+    interceptors: { activity: [createActivityObservabilityInterceptors({ spans })], workflowModules: workflowTracingOptions(spans).workflowModules },
+    sinks: workflowTracingOptions(spans).sinks
   });
   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,
@@ -130,7 +133,11 @@ 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}`,
+    ...(parseTraceparent(source.queueItem.turn.traceparent) ? {
+      traceLink: source.queueItem.turn.traceparent,
+      traceparent: formatTraceparent(randomBytes(16).toString("hex"), randomBytes(8).toString("hex"), "01")
+    } : {})
   } });
   try {
     const enqueued = await store.enqueueRoomAgentMission({

apps/worker/src/worker.ts

@@ -1,3 +1,4 @@
+import { workflowTracingOptions } from "./workflow-tracing-worker.js";
 import { fileURLToPath } from "node:url";
 import { hostname } from "node:os";
 import { assertDevInstanceIsolated, instanceBackgroundJobEnabled } from "@space/contracts";
@@ -43,9 +44,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 })] }
+  // Export activity attempts and replay-suppressed workflow spans.
+  interceptors: { activity: [createActivityObservabilityInterceptors({ spans })], workflowModules: workflowTracingOptions(spans).workflowModules },
+  sinks: workflowTracingOptions(spans).sinks
 });
 
 // Optional loopback-only Prometheus endpoint for worker metrics.

apps/worker/src/workflow-trace-context.ts

@@ -0,0 +1,68 @@
+import { parseTraceparent } from "@space/contracts/observability";
+import type { Headers, Sinks } from "@temporalio/workflow";
+
+export const traceHeader = "space-traceparent";
+export const linkHeader = "space-trace-link";
+
+// ASCII only: these helpers also run in Temporal's deterministic sandbox.
+export function tracePayload(value: string) {
+  return { metadata: { encoding: Uint8Array.from([98, 105, 110, 97, 114, 121, 47, 112, 108, 97, 105, 110]) },
+    data: Uint8Array.from(value, char => char.charCodeAt(0)) };
+}
+
+export function traceFromHeaders(headers: Headers, key = traceHeader): string | undefined {
+  const data = headers[key]?.data;
+  if (!data || data.length !== 55) return undefined;
+  const value = String.fromCharCode(...data);
+  return parseTraceparent(value) ? value : undefined;
+}
+
+export function traceFromArgs(args: readonly unknown[], key = "traceparent"): string | undefined {
+  const visit = (value: unknown, depth: number): string | undefined => {
+    if (!value || typeof value !== "object" || Array.isArray(value)) return undefined;
+    const record = value as Record<string, unknown>;
+    if (parseTraceparent(record[key])) return record[key] as string;
+    if (depth > 0) for (const nested of ["turn", "item", "bridge", "input", "turnInput"]) {
+      const found = visit(record[nested], depth - 1);
+      if (found) return found;
+    }
+    return undefined;
+  };
+  for (const arg of args) {
+    const found = visit(arg, 3);
+    if (found) return found;
+  }
+  return undefined;
+}
+
+const identifier = /^[A-Za-z0-9:._-]{1,200}$/;
+const attributeKeys = new Set([
+  "space.workflow_id", "space.workflow_run_id", "space.workflow_type", "space.activity_id",
+  "space.activity_type", "space.attempt", "space.outcome", "space.reason_code", "space.category", "space.duration_ms"
+]);
+
+/** No arbitrary attributes, free-form errors, arguments, paths or URLs. */
+export function safeTraceAttributes(attributes: Readonly<Record<string, unknown>>): Record<string, string | number> {
+  const result: Record<string, string | number> = {};
+  for (const [key, value] of Object.entries(attributes)) {
+    if (!attributeKeys.has(key)) continue;
+    if (typeof value === "string" && identifier.test(value)) result[key] = value;
+    else if (typeof value === "number" && Number.isFinite(value) && value >= 0) result[key] = value;
+  }
+  return result;
+}
+
+export interface WorkflowTraceSpan {
+  traceId: string;
+  spanId: string;
+  parentSpanId: string | null;
+  startedAt: number;
+  endedAt: number;
+  attributes: Record<string, string | number>;
+  error: boolean;
+  links: { traceId: string; spanId: string }[];
+}
+
+export interface WorkflowTraceSinks extends Sinks {
+  spaceTracing: { record(span: WorkflowTraceSpan): void };
+}

apps/worker/src/workflow-tracing-worker.ts

@@ -0,0 +1,26 @@
+import { fileURLToPath } from "node:url";
+import type { InjectedSinks } from "@temporalio/worker";
+import type { SpanExporter } from "@space/runtime";
+import { safeTraceAttributes, type WorkflowTraceSinks } from "./workflow-trace-context.js";
+
+export function workflowTracingOptions(spans: SpanExporter | null) {
+  const sinks: InjectedSinks<WorkflowTraceSinks> = {
+    spaceTracing: { record: {
+      callDuringReplay: false,
+      fn(_info, span) {
+        try {
+          const attributes = safeTraceAttributes(span.attributes);
+          spans?.record({ traceId: span.traceId, spanId: span.spanId, parentSpanId: span.parentSpanId,
+            name: `workflow ${attributes["space.workflow_type"] ?? "unknown"}`, kind: 1,
+            startTimeUnixNano: BigInt(span.startedAt) * 1_000_000n,
+            endTimeUnixNano: BigInt(span.endedAt) * 1_000_000n,
+            attributes, error: span.error, links: span.links });
+        } catch { /* Export must never fail workflow processing. */ }
+      }
+    } }
+  };
+  return {
+    workflowModules: spans ? [fileURLToPath(new URL("./workflow-tracing.js", import.meta.url))] : [],
+    sinks: spans ? sinks : {}
+  };
+}

apps/worker/src/workflow-tracing.ts

@@ -0,0 +1,63 @@
+import { ContinueAsNew, proxySinks, workflowInfo, type WorkflowInterceptors, type Headers } from "@temporalio/workflow";
+import { formatTraceparent, parseTraceparent } from "@space/contracts/observability";
+import { linkHeader, safeTraceAttributes, traceFromArgs, traceFromHeaders, traceHeader, tracePayload, type WorkflowTraceSinks } from "./workflow-trace-context.js";
+
+/** No I/O or random-number consumption inside workflow code. */
+export function interceptors(): WorkflowInterceptors {
+  const sink = proxySinks<WorkflowTraceSinks>().spaceTracing;
+  let current: string | undefined;
+  const headers = (original: Headers): Headers => current ? { ...original, [traceHeader]: tracePayload(current) } : original;
+  return {
+    inbound: [{
+      async execute(input, next) {
+        const info = workflowInfo();
+        const incoming = traceFromHeaders(input.headers) ?? traceFromArgs(input.args) ?? traceFromArgs([info.memo]);
+        const parent = parseTraceparent(incoming);
+        // Temporal run ids are UUIDs. Reusing their entropy avoids changing the
+        // workflow's seeded random sequence (and hence existing command ids).
+        const runHex = info.runId.replaceAll("-", "");
+        const spanId = runHex.slice(16);
+        const traceId = parent?.traceId ?? runHex;
+        current = formatTraceparent(traceId, spanId, parent?.flags ?? "01");
+        const argumentLink = traceFromArgs(input.args, "traceLink");
+        const selectedLink = traceFromHeaders(input.headers, linkHeader) ?? argumentLink;
+        const link = parseTraceparent(selectedLink);
+        const backgroundRoot = argumentLink === selectedLink && link && link.traceId !== parent?.traceId
+          && parseTraceparent(traceFromArgs(input.args))?.parentId === parent?.parentId;
+        const startedAt = Date.now();
+        let outcome = "COMPLETED";
+        try { return await next(input); }
+        catch (error) {
+          outcome = error instanceof ContinueAsNew ? "CONTINUED_AS_NEW" : "FAILED";
+          throw error;
+        } finally {
+          const endedAt = Date.now();
+          sink.record({ traceId, spanId, parentSpanId: backgroundRoot ? null : parent?.parentId ?? null, startedAt, endedAt,
+            attributes: safeTraceAttributes({ "space.workflow_id": info.workflowId, "space.workflow_run_id": info.runId,
+              "space.workflow_type": info.workflowType, "space.attempt": info.attempt, "space.outcome": outcome,
+              "space.duration_ms": Math.max(0, endedAt - startedAt) }), error: outcome === "FAILED",
+            links: link ? [{ traceId: link.traceId, spanId: link.parentId }] : [] });
+        }
+      }
+    }],
+    outbound: [{
+      scheduleActivity: (input, next) => next({ ...input, headers: headers(input.headers) }),
+      scheduleLocalActivity: (input, next) => next({ ...input, headers: headers(input.headers) }),
+      startChildWorkflowExecution(input, next) {
+        const explicit = traceFromArgs(input.options.args ?? []);
+        const explicitLink = traceFromArgs(input.options.args ?? [], "traceLink");
+        // A supervisor may launch a new mission received on a later request.
+        // Preserve that mission's trace; ordinary children nest under this run.
+        const differentTrace = explicit && parseTraceparent(explicit)?.traceId !== parseTraceparent(current)?.traceId;
+        let childHeaders = headers(input.headers);
+        if (differentTrace) childHeaders = { ...childHeaders, [traceHeader]: tracePayload(explicit) };
+        const link = explicitLink ?? (differentTrace ? current : undefined);
+        if (link) childHeaders = { ...childHeaders, [linkHeader]: tracePayload(link) };
+        return next({ ...input, headers: childHeaders });
+      },
+      continueAsNew(input, next) {
+        return next({ ...input, headers: { ...headers(input.headers), ...(current ? { [linkHeader]: tracePayload(current) } : {}) } });
+      }
+    }]
+  };
+}

apps/worker/tests/activity-observability.test.ts

@@ -186,7 +186,9 @@ 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).toContain("workflowModules: workflowTracingOptions(spans).workflowModules");
+    expect(source, entry).toContain("sinks: workflowTracingOptions(spans).sinks");
+    expect(source, entry).toMatch(/interceptors: \{ activity: \[createActivityObservabilityInterceptors\(\{ spans \}\)\]/);
   }
 });
 
@@ -203,3 +205,12 @@ it("exports one INTERNAL span per activity under the caller's traceparent", asyn
   expect(recorded[0]).toMatchObject({ traceId: "4bf92f3577b34da6a3ce929d0e0e4736", parentSpanId: "00f067aa0ba902b7", name: "activity runTurn",
     kind: 1, error: false, startTimeUnixNano: 1_000_000_000n, endTimeUnixNano: 1_250_000_000n });
 });
+
+
+it("collector failures preserve activity results and original errors", async () => {
+  const spans = { record: () => { throw new Error("collector failure"); }, flush: async () => {},
+    close: async () => {}, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) };
+  await expect(observeActivityExecution(info(), [], async () => "ok", { spans, traceparent })).resolves.toBe("ok");
+  const error = new Error("original activity error");
+  await expect(observeActivityExecution(info(), [], async () => { throw error; }, { spans, traceparent })).rejects.toBe(error);
+});

apps/worker/tests/fixtures/tracing-workflows.ts

@@ -0,0 +1,21 @@
+import { continueAsNew, executeChild, proxyActivities } from "@temporalio/workflow";
+
+const activities = proxyActivities<{ attempt(): Promise<string> }>({
+  startToCloseTimeout: "10 seconds", retry: { initialInterval: "1 second", maximumAttempts: 2 }
+});
+
+export async function tracingChild(): Promise<string> {
+  return activities.attempt();
+}
+
+export async function tracingRoot(input: { traceparent: string; prompt: string }, generation = 0): Promise<string> {
+  await activities.attempt();
+  await executeChild(tracingChild);
+  if (generation === 0) return continueAsNew<typeof tracingRoot>(input, 1);
+  return "complete";
+}
+
+export async function tracingBackground(input: { traceparent: string; traceLink: string }): Promise<string> {
+  await activities.attempt();
+  return "background complete";
+}