Export worker/workflow OTLP spans: one trace from API request to Temporal activities (#61)

apps/api/src/dot-goals.ts

@@ -16,6 +16,7 @@ import {
 } from "@space/contracts";
 import { SpaceConflictError, SpaceNotFoundError, type SpaceStore } from "@space/runtime";
 import type { DotGoalsRepository, DotsRepository } from "@space/db";
+import { traceContextForRawRequest, traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
 
 /**
  * Space Dots Stage 4: Goals.
@@ -26,7 +27,7 @@ import type { DotGoalsRepository, DotsRepository } from "@space/db";
  * blocking question, cancellation, three consecutive failures or the budget.
  */
 export interface DotGoalWorkflowCoordinator {
-  start(goalId: string): Promise<void>;
+  start(goalId: string, trace?: { traceId?: string | null; traceparent?: string | null }): Promise<void>;
   wake(goalId: string): Promise<void>;
   /** Starts the Goal's workflow when none is running; true when it had to start one. */
   ensure?(goalId: string): Promise<boolean>;
@@ -38,11 +39,18 @@ export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordi
     try { return await action(new Client({ connection, namespace: this.options.namespace })); }
     finally { await connection.close(); }
   }
-  start(goalId: string) {
+  start(goalId: string, trace?: { traceId?: string | null; traceparent?: string | null }) {
+    const traceparent = trace?.traceparent ?? traceparentForTraceId(trace?.traceId);
+    const memo = workflowCorrelationMemo({ traceId: trace?.traceId, traceparent });
     return this.client(async (client) => {
       try {
-        await client.workflow.start("dotGoalWorkflow", { workflowId: dotGoalWorkflowId(goalId), taskQueue: this.options.taskQueue,
-          args: [{ goalId }], workflowIdReusePolicy: "REJECT_DUPLICATE" });
+        await client.workflow.start("dotGoalWorkflow", {
+          workflowId: dotGoalWorkflowId(goalId),
+          taskQueue: this.options.taskQueue,
+          args: [{ goalId, ...(trace?.traceId ? { traceId: trace.traceId } : {}), ...(traceparent ? { traceparent } : {}) }],
+          workflowIdReusePolicy: "REJECT_DUPLICATE",
+          ...(Object.keys(memo).length ? { memo } : {})
+        });
       } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; }
     });
   }
@@ -125,7 +133,15 @@ export function createDotGoalService(deps: DotGoalServiceDeps) {
       if ((error as { code?: string }).code === "DOT_GOAL_OPEN") throw new DotGoalError(409, "DOT_GOAL_OPEN", "This conversation already has an open Goal.");
       throw error;
     }
-    try { await deps.workflow.start(goal.goalId); }
+    try {
+      if (request) {
+        const traceId = request.id ? String(request.id) : null;
+        const traceparent = request.raw ? traceContextForRawRequest(request.raw).traceparent : traceparentForTraceId(traceId);
+        await deps.workflow.start(goal.goalId, { traceId, traceparent });
+      } else {
+        await deps.workflow.start(goal.goalId);
+      }
+    }
     catch {
       goal = await finish(goal, "FAILED", "The Goal runtime could not start. Try again.");
       throw new DotGoalError(503, "DOT_GOAL_RUNTIME", "The Goal runtime could not start. Try again.");

apps/api/src/memory-consolidation-coordinator.ts

@@ -6,6 +6,7 @@ import {
   type MemoryConsolidationWorkflowInput
 } from "@space/contracts";
 import { SpaceFeatureDisabledError } from "@space/runtime";
+import { withTurnTraceparent, workflowCorrelationMemo } from "./trace-context.js";
 
 export { MEMORY_CONSOLIDATION_TASK_QUEUE, MEMORY_CONSOLIDATION_WORKFLOW_TYPE };
 
@@ -21,6 +22,7 @@ interface WorkflowStartOptions {
   workflowId: string;
   workflowIdReusePolicy: "REJECT_DUPLICATE";
   workflowIdConflictPolicy: "USE_EXISTING";
+  memo?: Record<string, string>;
 }
 
 interface WorkflowStartHandle {
@@ -42,14 +44,16 @@ export async function scheduleMemoryConsolidation(
   workflowId: string,
   startWorkflow: MemoryConsolidationWorkflowStarter
 ): Promise<MemoryConsolidationStartResult> {
-  const input = memoryConsolidationWorkflowInputSchema.parse(rawInput);
+  const input = withTurnTraceparent(memoryConsolidationWorkflowInputSchema.parse(rawInput));
+  const memo = input.traceparent ? workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) : undefined;
   try {
     const handle = await startWorkflow(MEMORY_CONSOLIDATION_WORKFLOW_TYPE, {
       args: [input],
       taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",
-      workflowIdConflictPolicy: "USE_EXISTING"
+      workflowIdConflictPolicy: "USE_EXISTING",
+      ...(memo && Object.keys(memo).length ? { memo } : {})
     });
     return {
       status: "SCHEDULED",

apps/api/src/memory-mutation-coordinator.ts

@@ -7,6 +7,7 @@ import {
   type MemoryMutationWorkflowInput
 } from "@space/contracts";
 import { SpaceFeatureDisabledError } from "@space/runtime";
+import { withTurnTraceparent, workflowCorrelationMemo } from "./trace-context.js";
 
 export { MEMORY_MUTATION_TASK_QUEUE, MEMORY_MUTATION_WORKFLOW_TYPE, buildMemoryMutationWorkflowId };
 
@@ -22,6 +23,7 @@ interface WorkflowStartOptions {
   workflowId: string;
   workflowIdReusePolicy: "REJECT_DUPLICATE";
   workflowIdConflictPolicy: "USE_EXISTING";
+  memo?: Record<string, string>;
 }
 
 interface WorkflowStartHandle {
@@ -43,15 +45,17 @@ export async function scheduleMemoryMutation(
   startWorkflow: MemoryMutationWorkflowStarter,
   commandHash?: string
 ): Promise<MemoryMutationStartResult> {
-  const input = memoryMutationWorkflowInputSchema.parse(rawInput);
+  const input = withTurnTraceparent(memoryMutationWorkflowInputSchema.parse(rawInput));
   const workflowId = buildMemoryMutationWorkflowId(input.changeSetId, commandHash);
+  const memo = input.traceparent ? workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) : undefined;
   try {
     const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, {
       args: [input],
       taskQueue: MEMORY_MUTATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",
-      workflowIdConflictPolicy: "USE_EXISTING"
+      workflowIdConflictPolicy: "USE_EXISTING",
+      ...(memo && Object.keys(memo).length ? { memo } : {})
     });
     return { status: "SCHEDULED", workflowId: handle.workflowId, runId: handle.firstExecutionRunId ?? null };
   } catch (error) {

apps/api/src/room-agent.ts

@@ -188,16 +188,17 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo
 
   async enqueueAction(actionId: string, bridge: SpaceAgentRoomActionBridgeRequest, traceId: string): Promise<void> {
     const lease = await this.connect("roomAgentActionWorkflow");
+    const traceparent = traceparentForTraceId(traceId) ?? traceparentForTraceId(bridge.requestId ?? null);
     try {
       try {
         await lease.client.workflow.start("roomAgentActionWorkflow", {
           workflowId: `space-room-action:${actionId}`, taskQueue: this.options.taskQueue,
-          args: [{ actionId, bridge: { ...bridge, backgroundExecution: true }, traceId }],
+          args: [{ actionId, bridge: { ...bridge, backgroundExecution: true }, traceId, ...(traceparent ? { traceparent } : {}) }],
           workflowIdReusePolicy: "REJECT_DUPLICATE",
           // traceId is this internal request; bridge.requestId is the originating request.
           memo: workflowCorrelationMemo({
             traceId,
-            traceparent: traceparentForTraceId(traceId),
+            traceparent,
             parentTraceId: bridge.requestId ?? null,
             missionId: bridge.missionId
           })
@@ -219,14 +220,20 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo
     // the child turn workflow and its activities (additive optional field).
     const item = { ...parsed, turn: withTurnTraceparent(parsed.turn) };
     const workflowId = buildRoomAgentSupervisorWorkflowId(item.turn.roomId);
+    const memo = workflowCorrelationMemo({
+      traceId: item.turn.traceId,
+      traceparent: item.turn.traceparent,
+      missionId: item.missionId
+    });
     const lease = await this.connect(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE);
     try {
       const handle = await lease.client.workflow.signalWithStart(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE, {
         workflowId,
         taskQueue: this.options.taskQueue,
         args: [{ roomId: item.turn.roomId, pending: [], processedCount: 0 }],
         signal: ROOM_AGENT_ENQUEUE_SIGNAL,
-        signalArgs: [item]
+        signalArgs: [item],
+        ...(Object.keys(memo).length ? { memo } : {})
       });
       recordWorkflowStart(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE, "SIGNALED");
       return { workflowId, runId: handle.signaledRunId };

apps/api/src/trace-context.ts

@@ -96,7 +96,7 @@ 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);
+  return registry.lookup(traceId)?.traceparent ?? null;
 }
 
 /** Adds the request's traceparent to a turn input unless it already has a valid one. */

apps/worker/package.json

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

apps/worker/src/activity-observability.ts

@@ -1,7 +1,7 @@
 import { AsyncLocalStorage } from "node:async_hooks";
 import { randomBytes } from "node:crypto";
 import type { Context } from "@temporalio/activity";
-import type { ActivityInterceptorsFactory } from "@temporalio/worker";
+import { defaultPayloadConverter, type ActivityInterceptorsFactory, type Headers } from "@temporalio/worker";
 import {
   categorizeFailureCode,
   formatTraceparent,
@@ -10,9 +10,11 @@ import {
   traceIdFromTraceparent,
   type FailureCategory
 } from "@space/contracts";
-import type { SpanExporter } from "@space/runtime";
+import { sanitizeSpanAttributes, type SpanExporter } from "@space/runtime";
 import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js";
 import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js";
+import { lookupActiveWorkflowTrace } from "./workflow-observability.js";
+import { WORKFLOW_TRACE_ID_HEADER, WORKFLOW_TRACEPARENT_HEADER } from "./workflow-observability-interceptors.js";
 
 /**
  * One structured start/finish record and metrics per activity execution, with
@@ -135,8 +137,21 @@ export interface ObserveActivityDependencies {
   metrics?: WorkerMetrics;
   now?: () => number;
   cancelled?: () => boolean;
-  /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */
+  /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input or workflow header carries a traceparent. */
   spans?: SpanExporter | null;
+  headers?: Headers;
+}
+
+function readActivityHeader(headers: Headers | undefined, key: string): string | null {
+  if (!headers) return null;
+  const payload = headers[key];
+  if (!payload) return null;
+  try {
+    const decoded = defaultPayloadConverter.fromPayload<unknown>(payload);
+    return typeof decoded === "string" ? decoded : null;
+  } catch {
+    return null;
+  }
 }
 
 function activityTraceparent(args: readonly unknown[]) {
@@ -194,16 +209,57 @@ export async function observeActivityExecution<T>(
       queueMs,
       rssBytes: process.memoryUsage().rss
     });
-    const parent = dependencies.spans ? activityTraceparent(args) : null;
+    const headerTp = parseTraceparent(readActivityHeader(dependencies.headers, WORKFLOW_TRACEPARENT_HEADER));
+    const activeWf = lookupActiveWorkflowTrace(
+      info.workflowExecution?.workflowId ?? null,
+      info.workflowExecution?.runId ?? null
+    );
+    const activeWfTp = parseTraceparent(activeWf?.traceparent);
+    const parent = dependencies.spans ? (headerTp ?? activeWfTp ?? activityTraceparent(args)) : null;
     if (parent) {
+      const corr = activityCorrelation(args);
+      const rawAttributes: Record<string, string | number> = {
+        ...(activeWf?.correlation ?? {}),
+        "space.activity_type": activityType,
+        "space.outcome": outcome,
+        "space.attempt": info.attempt,
+        "space.duration_ms": durationMs
+      };
+      if (info.activityId) rawAttributes["space.activity_id"] = String(info.activityId);
+      if (info.workflowType) rawAttributes["space.workflow_type"] = info.workflowType;
+      if (info.workflowExecution?.workflowId) rawAttributes["space.workflow_id"] = info.workflowExecution.workflowId;
+      if (info.workflowExecution?.runId) rawAttributes["space.run_id"] = info.workflowExecution.runId;
+      if (queueMs !== null) rawAttributes["space.queue_ms"] = queueMs;
+      if (reasonCode) rawAttributes["space.reason_code"] = reasonCode;
+      if (category) rawAttributes["space.category"] = category;
+      const headerTraceId = readActivityHeader(dependencies.headers, WORKFLOW_TRACE_ID_HEADER);
+      if (headerTraceId) rawAttributes["space.trace_id"] = headerTraceId;
+      const corrAttrMap: ReadonlyArray<readonly [string, string]> = [
+        ["traceId", "space.trace_id"],
+        ["parentTraceId", "space.parent_trace_id"],
+        ["missionId", "space.mission_id"],
+        ["roomId", "space.room_id"],
+        ["paneId", "space.pane_id"],
+        ["agentRunId", "space.agent_run_id"],
+        ["agentSessionId", "space.agent_session_id"],
+        ["actionId", "space.action_id"],
+        ["goalId", "space.goal_id"],
+        ["dotId", "space.dot_id"],
+        ["turnId", "space.turn_id"],
+        ["cliTaskId", "space.cli_task_id"],
+        ["revisionId", "space.revision_id"]
+      ];
+      for (const [corrKey, attrKey] of corrAttrMap) {
+        if (corr[corrKey]) rawAttributes[attrKey] = corr[corrKey]!;
+      }
       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: sanitizeSpanAttributes(rawAttributes),
         error: outcome === "FAILED" || outcome === "FAILED_RESULT",
         kind: 1
       });
@@ -224,11 +280,14 @@ export async function observeActivityExecution<T>(
 }
 
 /** Temporal worker interceptor factory: register in Worker.create({ interceptors }). */
-export function createActivityObservabilityInterceptors(dependencies: Omit<ObserveActivityDependencies, "cancelled"> = {}): ActivityInterceptorsFactory {
+export function createActivityObservabilityInterceptors(
+  dependencies: Omit<ObserveActivityDependencies, "cancelled" | "headers"> = {}
+): ActivityInterceptorsFactory {
   return (context) => ({
     inbound: {
       execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), {
         ...dependencies,
+        headers: input.headers,
         cancelled: () => context.cancellationSignal.aborted
       })
     }

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

@@ -9,6 +9,7 @@ import {
 import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js";
 import { otlpSpanExporterFromEnv } from "@space/runtime";
 import { createActivityObservabilityInterceptors } from "./activity-observability.js";
+import { createWorkflowObservabilitySinks, workflowObservabilityInterceptorPath } from "./workflow-observability.js";
 import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js";
 
 const service = "space-memory-maintenance-worker";
@@ -30,7 +31,11 @@ if (!memoryMaintenanceEnabled(process.env)) {
       workflowsPath,
       activities,
       ...memoryMaintenanceWorkerOptions,
-      interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+      interceptors: {
+        activity: [createActivityObservabilityInterceptors({ spans })],
+        workflowModules: [workflowObservabilityInterceptorPath]
+      },
+      sinks: createWorkflowObservabilitySinks({ spans })
     });
     console.log(JSON.stringify({
       service,

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

@@ -5,6 +5,7 @@ import { MEMORY_MUTATION_TASK_QUEUE, memoryMutationsEnabled, memoryMutationWorke
 import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js";
 import { otlpSpanExporterFromEnv } from "@space/runtime";
 import { createActivityObservabilityInterceptors } from "./activity-observability.js";
+import { createWorkflowObservabilitySinks, workflowObservabilityInterceptorPath } from "./workflow-observability.js";
 import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js";
 
 const service = "space-memory-mutation-worker";
@@ -23,7 +24,11 @@ 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: [workflowObservabilityInterceptorPath]
+    },
+    sinks: createWorkflowObservabilitySinks({ 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/worker.ts

@@ -7,6 +7,7 @@ import { TelegramDeliveryWorker, TelegramSecretStore, otlpSpanExporterFromEnv }
 import { NativeConnection, Runtime, Worker } from "@temporalio/worker";
 import * as activities from "./activities.js";
 import { createActivityObservabilityInterceptors } from "./activity-observability.js";
+import { createWorkflowObservabilitySinks, workflowObservabilityInterceptorPath } from "./workflow-observability.js";
 import { resolveTemporalCoreMetricsBind, resolveWorkerMetricsListen, startWorkerMetricsServer } from "./worker-metrics.js";
 import { workerLog } from "./worker-log.js";
 import { sweepCodexCliCompletions } from "./codex-cli-completion.js";
@@ -43,9 +44,11 @@ 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 })] }
+  interceptors: {
+    activity: [createActivityObservabilityInterceptors({ spans })],
+    workflowModules: [workflowObservabilityInterceptorPath]
+  },
+  sinks: createWorkflowObservabilitySinks({ spans })
 });
 
 // Optional loopback-only Prometheus endpoint for worker metrics.