feat(observability): export workflow and activity-attempt spans over OTLP (closes #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 { workflowCorrelationMemo } from "./trace-context.js";
 
 /**
  * Space Dots Stage 4: Goals.
@@ -26,7 +27,8 @@ import type { DotGoalsRepository, DotsRepository } from "@space/db";
  * blocking question, cancellation, three consecutive failures or the budget.
  */
 export interface DotGoalWorkflowCoordinator {
-  start(goalId: string): Promise<void>;
+  /** `traceparent`: the starting request's W3C context, kept in the memo for worker spans. */
+  start(goalId: string, traceparent?: string): 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 +40,12 @@ 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, traceparent?: string) {
+    const memo = traceparent ? workflowCorrelationMemo({ traceId: null, traceparent }) : {};
     return this.client(async (client) => {
       try {
         await client.workflow.start("dotGoalWorkflow", { workflowId: dotGoalWorkflowId(goalId), taskQueue: this.options.taskQueue,
-          args: [{ goalId }], workflowIdReusePolicy: "REJECT_DUPLICATE" });
+          args: [{ goalId }], workflowIdReusePolicy: "REJECT_DUPLICATE", ...(Object.keys(memo).length ? { memo } : {}) });
       } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; }
     });
   }
@@ -125,7 +128,8 @@ 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); }
+    const traceparent = request?.traceContext?.traceparent;
+    try { await (traceparent ? deps.workflow.start(goal.goalId, traceparent) : 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,14 @@ import {
   type MemoryConsolidationWorkflowInput
 } from "@space/contracts";
 import { SpaceFeatureDisabledError } from "@space/runtime";
+import { traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
+
+// Only when the starting request's trace context is known (an API request in
+// flight); otherwise the start options are unchanged.
+function correlationMemo(traceId: string): { memo?: Record<string, string> } {
+  const traceparent = traceparentForTraceId(traceId);
+  return traceparent ? { memo: workflowCorrelationMemo({ traceId, traceparent }) } : {};
+}
 
 export { MEMORY_CONSOLIDATION_TASK_QUEUE, MEMORY_CONSOLIDATION_WORKFLOW_TYPE };
 
@@ -46,6 +54,8 @@ export async function scheduleMemoryConsolidation(
   try {
     const handle = await startWorkflow(MEMORY_CONSOLIDATION_WORKFLOW_TYPE, {
       args: [input],
+      // Correlation ids and the starting request's W3C context (worker spans).
+      ...correlationMemo(input.traceId),
       taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",

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

@@ -7,6 +7,14 @@ import {
   type MemoryMutationWorkflowInput
 } from "@space/contracts";
 import { SpaceFeatureDisabledError } from "@space/runtime";
+import { traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
+
+// Only when the starting request's trace context is known (an API request in
+// flight); otherwise the start options are unchanged.
+function correlationMemo(traceId: string): { memo?: Record<string, string> } {
+  const traceparent = traceparentForTraceId(traceId);
+  return traceparent ? { memo: workflowCorrelationMemo({ traceId, traceparent }) } : {};
+}
 
 export { MEMORY_MUTATION_TASK_QUEUE, MEMORY_MUTATION_WORKFLOW_TYPE, buildMemoryMutationWorkflowId };
 
@@ -48,6 +56,8 @@ export async function scheduleMemoryMutation(
   try {
     const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, {
       args: [input],
+      // Correlation ids and the starting request's W3C context (worker spans).
+      ...correlationMemo(input.traceId),
       taskQueue: MEMORY_MUTATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",

apps/api/src/trace-context.ts

@@ -94,9 +94,15 @@ 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.
+ * Its parent id is the request's own SERVER span id (the span the API exports),
+ * so a workflow span that names it as parent joins the request's trace instead
+ * of pointing at a span that is never exported.
+ */
 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/api/tests/trace-context-workflow-parent.test.ts

@@ -0,0 +1,31 @@
+import { describe, expect, it } from "vitest";
+import { parseTraceparent } from "@space/contracts";
+import { createRequestTraceContext, requestTraceContexts, traceparentForTraceId, TraceContextRegistry } from "../src/trace-context.js";
+import { scheduleMemoryMutation } from "../src/memory-mutation-coordinator.js";
+import { scheduleMemoryConsolidation } from "../src/memory-consolidation-coordinator.js";
+
+describe("trace context handed to Temporal workflows", () => {
+  it("names the request's exported SERVER span as the workflow's parent", () => {
+    const registry = new TraceContextRegistry();
+    const context = createRequestTraceContext(undefined);
+    registry.remember("req:parent-span-01", context);
+    const traceparent = traceparentForTraceId("req:parent-span-01", registry);
+    expect(parseTraceparent(traceparent)).toEqual({ traceId: context.w3cTraceId, parentId: context.spanId, flags: "01" });
+    expect(traceparentForTraceId("req:unknown-000001", registry)).toBeNull();
+  });
+
+  it("adds the correlation memo to memory workflows started during a traced request", async () => {
+    const context = createRequestTraceContext(undefined);
+    requestTraceContexts.remember("req:memory-start-01", context);
+    const options: unknown[] = [];
+    const start = async (_type: string, option: { workflowId: string }) => { options.push(option); return { workflowId: option.workflowId, firstExecutionRunId: "run:1" }; };
+    await scheduleMemoryMutation({ changeSetId: "memory_change:traced", traceId: "req:memory-start-01" }, start as never);
+    await scheduleMemoryConsolidation({ runId: "memory_run:traced", traceId: "req:memory-start-01" }, "space-memory-consolidation:traced", start as never);
+    for (const option of options) {
+      expect((option as { memo: Record<string, string> }).memo).toEqual({
+        spaceTraceId: "req:memory-start-01", traceparent: context.traceparent, w3cTraceId: context.w3cTraceId
+      });
+    }
+    expect(options).toHaveLength(2);
+  });
+});

apps/worker/package.json

@@ -21,11 +21,13 @@
     "@space/streaming": "0.2.5",
     "@temporalio/activity": "1.24.0",
     "@temporalio/client": "1.24.0",
+    "@temporalio/common": "1.24.0",
     "@temporalio/worker": "1.24.0",
     "@temporalio/workflow": "1.24.0",
     "undici": "8.12.3"
   },
   "devDependencies": {
+    "@temporalio/testing": "1.24.0",
     "typescript": "^5.9.3",
     "vitest": "^4.1.12"
   },

apps/worker/src/activity-observability.ts

@@ -11,6 +11,7 @@ import {
   type FailureCategory
 } from "@space/contracts";
 import type { SpanExporter } from "@space/runtime";
+import { allowlistSpanAttributes, correlationAttributes, readTraceHeader } from "./trace-carrier.js";
 import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js";
 import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js";
 
@@ -135,8 +136,15 @@ 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 INTERNAL span per
+   * activity attempt, child of the scheduling workflow span (the `space-trace`
+   * Temporal header), or of the input's traceparent for work scheduled by a
+   * workflow without tracing.
+   */
   spans?: SpanExporter | null;
+  /** traceparent from the `space-trace` header; its parent id is the workflow span. */
+  parentTraceparent?: string | null;
 }
 
 function activityTraceparent(args: readonly unknown[]) {
@@ -194,19 +202,38 @@ export async function observeActivityExecution<T>(
       queueMs,
       rssBytes: process.memoryUsage().rss
     });
-    const parent = dependencies.spans ? activityTraceparent(args) : null;
+    const parent = dependencies.spans
+      ? parseTraceparent(dependencies.parentTraceparent) ?? activityTraceparent(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: allowlistSpanAttributes({
+            ...correlationAttributes(args),
+            "space.activity_type": activityType,
+            "space.activity_id": info.activityId,
+            "space.workflow_type": info.workflowType,
+            "space.workflow_id": info.workflowExecution?.workflowId,
+            "space.run_id": info.workflowExecution?.runId,
+            "space.attempt": info.attempt,
+            "space.outcome": outcome,
+            "space.reason_code": reasonCode,
+            "space.category": category,
+            "space.duration_ms": durationMs,
+            "space.queue_ms": queueMs
+          }),
+          error: outcome === "FAILED" || outcome === "FAILED_RESULT",
+          kind: 1
+        });
+      } catch {
+        // Span export is best effort and never affects the activity.
+      }
     }
   };
 
@@ -229,6 +256,7 @@ export function createActivityObservabilityInterceptors(dependencies: Omit<Obser
     inbound: {
       execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), {
         ...dependencies,
+        ...(dependencies.spans ? { parentTraceparent: readTraceHeader(input.headers)?.traceparent ?? null } : {}),
         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 { workerTracingOptions } from "./worker-tracing.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 })] }
+      ...workerTracingOptions(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 { workerTracingOptions } from "./worker-tracing.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 })] }
+    ...workerTracingOptions(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

@@ -122,10 +122,18 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis
   ].join("\n");
   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));
+  // A continuation is a new trace (trace:bg-<id>) with an OTLP span link to the
+  // operator request it continues; it does not reuse that request's
+  // traceparent as its parent. Later generations keep linking to the same request.
+  const { traceparent: sourceRequestTraceparent, ...sourceTurn } = source.queueItem.turn;
+  const sourceTraceparent = sourceRequestTraceparent ?? source.queueItem.turn.roomAgentContinuation?.sourceTraceparent;
   const item = roomAgentSupervisorQueueItemSchema.parse({ missionId: input.missionId, turn: {
-    ...source.queueItem.turn,
+    ...sourceTurn,
     prompt, roomAgentOperatorRequest: request,
-    roomAgentContinuation: { sourceRunId: source.run.runId, generation },
+    roomAgentContinuation: {
+      sourceRunId: source.run.runId, generation,
+      ...(sourceTraceparent ? { sourceTraceparent } : {})
+    },
     artifactIds: [], agentRunId: runId,
     agentUserMessageId: `agent_msg:bg-context-${input.continuationId}`,
     agentAssistantMessageId: `agent_msg:bg-response-${input.continuationId}`,