fix: [$600] Export worker and workflow spans over OTLP: one trace from API (closes #61)

PR-COMMENT.md

@@ -0,0 +1,13 @@
+Thank you for the review guidance. I understand that maintainer review is still pending.
+
+Verification used Windows 10.0.19045.0, supported Node v22.23.3 and npm 10.9.9, with Temporal CLI 1.9.1 / Server 1.32.0.
+Tested HEAD: 98bcb3859778b85951c6f15902fe82f156562f9e plus the PR working-tree changes.
+Recorded contracts/runtime/worker builds and API/worker type checks passed; runtime tests passed 7 tests and the API selection passed 14.
+OPEN BLOCKER: the initial worker selection timed out (46 passed, 1 failed, 1 skipped). Its unchanged rerun passed, but further sequential repeats passed twice and failed once with the same 5000ms continuity-test timeout.
+Each new passing repeat had 47 passed and 1 skipped; the failed repeat had 46 passed, 1 failed and 1 skipped. The cause remains unknown; no timeout or behavior change was made.
+All 6 Temporal/HTTP integration tests, including replay emitting zero spans and secret-sentinel assertions, passed in both new passing repeats; only the unavailable-server diagnostic was skipped.
+The recorded full worker suite had 242 passed, 14 failed and 1 skipped; the same 14 failures reproduced on the merge-base.
+The recorded root check passed 7 architecture tests before the Windows gateway bundler failed; that bundler failure also reproduced on the merge-base.
+The earlier Node 20 undici error did not recur. Windows socket, symlink, fsync, path and permission limitations remain.
+Removed only the two reported extra EOF blank lines; whole-tree git diff --check now exits 0. Historical failed results are retained.
+See NOTES.md, "Verification for review" and its latest audit follow-up, for commands, counts, baseline comparisons, limitations and log locations. Stable targeted verification is still blocked.

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

@@ -1,3 +1,4 @@
+import { traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
 import { Client, Connection } from "@temporalio/client";
 import {
   MEMORY_CONSOLIDATION_TASK_QUEUE,
@@ -16,6 +17,7 @@ export interface MemoryConsolidationStartResult {
 }
 
 interface WorkflowStartOptions {
+  memo?: Record<string, string>;
   args: [MemoryConsolidationWorkflowInput];
   taskQueue: typeof MEMORY_CONSOLIDATION_TASK_QUEUE;
   workflowId: string;
@@ -43,9 +45,11 @@ export async function scheduleMemoryConsolidation(
   startWorkflow: MemoryConsolidationWorkflowStarter
 ): Promise<MemoryConsolidationStartResult> {
   const input = memoryConsolidationWorkflowInputSchema.parse(rawInput);
+  const traceparent = traceparentForTraceId(input.traceId);
   try {
     const handle = await startWorkflow(MEMORY_CONSOLIDATION_WORKFLOW_TYPE, {
       args: [input],
+      ...(traceparent ? { memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent }) } : {}),
       taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",

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

@@ -1,3 +1,4 @@
+import { traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
 import { Client, Connection } from "@temporalio/client";
 import {
   MEMORY_MUTATION_TASK_QUEUE,
@@ -17,6 +18,7 @@ export interface MemoryMutationStartResult {
 }
 
 interface WorkflowStartOptions {
+  memo?: Record<string, string>;
   args: [MemoryMutationWorkflowInput];
   taskQueue: typeof MEMORY_MUTATION_TASK_QUEUE;
   workflowId: string;
@@ -45,9 +47,11 @@ export async function scheduleMemoryMutation(
 ): Promise<MemoryMutationStartResult> {
   const input = memoryMutationWorkflowInputSchema.parse(rawInput);
   const workflowId = buildMemoryMutationWorkflowId(input.changeSetId, commandHash);
+  const traceparent = traceparentForTraceId(input.traceId);
   try {
     const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, {
       args: [input],
+      ...(traceparent ? { memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent }) } : {}),
       taskQueue: MEMORY_MUTATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",

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/api/tests/otlp-parent-continuity.test.ts

@@ -0,0 +1,21 @@
+import { describe, expect, it } from "vitest";
+import { createRequestTraceContext, requestTraceContexts, traceparentForTraceId } from "../src/trace-context.js";
+import { scheduleMemoryConsolidation } from "../src/memory-consolidation-coordinator.js";
+import { scheduleMemoryMutation } from "../src/memory-mutation-coordinator.js";
+
+describe("OTLP API parent continuity", () => {
+  it("uses the exported request span for direct turn and memory workflow starts", async () => {
+    const context = createRequestTraceContext(undefined);
+    const traceId = "req:otlp-parent-test";
+    requestTraceContexts.remember(traceId, context);
+    expect(traceparentForTraceId(traceId)).toBe(context.traceparent);
+    const memos: unknown[] = [];
+    const start = async (_type: string, options: { workflowId: string; memo?: Record<string, string> }) => {
+      memos.push(options.memo);
+      return { workflowId: options.workflowId };
+    };
+    await scheduleMemoryConsolidation({ runId: "audit:test", traceId }, "workflow:test", start);
+    await scheduleMemoryMutation({ changeSetId: "change:test", traceId }, start);
+    expect(memos).toEqual([0, 1].map(() => ({ spaceTraceId: traceId, traceparent: context.traceparent, w3cTraceId: context.w3cTraceId })));
+  });
+});

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/worker/package.json

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

apps/worker/src/activity-observability.ts

@@ -1,7 +1,8 @@
+import { allowlistSpanAttributes } from "./tracing/span-attributes.js";
 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 ActivityExecuteInput, type ActivityInterceptorsFactory } from "@temporalio/worker";
 import {
   categorizeFailureCode,
   formatTraceparent,
@@ -10,7 +11,8 @@ import {
   traceIdFromTraceparent,
   type FailureCategory
 } from "@space/contracts";
-import type { SpanExporter } from "@space/runtime";
+import { type SpanExporter } from "@space/runtime";
+import { TRACEPARENT_HEADER } from "./workflow-tracing-headers.js";
 import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js";
 import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js";
 
@@ -135,16 +137,42 @@ 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
+   * attempt under the workflow span (`space-traceparent` header) or, without
+   * that header, under the traceparent in the activity input.
+   */
   spans?: SpanExporter | null;
+  /** The workflow span from the activity's `space-traceparent` header. */
+  parentTraceparent?: string | null;
 }
 
-function activityTraceparent(args: readonly unknown[]) {
+function activityTraceparent(args: readonly unknown[], header: string | null | undefined) {
+  const fromHeader = parseTraceparent(header);
+  if (fromHeader) return fromHeader;
   const found: Partial<Record<CorrelationKey, string>> = {};
   for (const arg of args) readCorrelation(arg, found, 2);
   return parseTraceparent(found.traceparent);
 }
 
+/** The traceparent in a Temporal header, or null when absent or not a valid string payload. */
+export function traceparentFromHeaders(headers: ActivityExecuteInput["headers"] | undefined): string | null {
+  const payload = headers?.[TRACEPARENT_HEADER];
+  if (!payload) return null;
+  try {
+    const value: unknown = defaultPayloadConverter.fromPayload(payload);
+    return parseTraceparent(value) ? (value as string) : null;
+  } catch {
+    return null;
+  }
+}
+
+function randomSpanId(): string {
+  let spanId = randomBytes(8).toString("hex");
+  while (/^0+$/.test(spanId)) spanId = randomBytes(8).toString("hex");
+  return spanId;
+}
+
 export async function observeActivityExecution<T>(
   info: ActivityInfo,
   args: readonly unknown[],
@@ -194,19 +222,40 @@ export async function observeActivityExecution<T>(
       queueMs,
       rssBytes: process.memoryUsage().rss
     });
-    const parent = dependencies.spans ? 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
-      });
+    // Export never affects the activity: a full queue drops, a bad span is skipped.
+    try {
+      const parent = dependencies.spans ? activityTraceparent(args, dependencies.parentTraceparent) : null;
+      if (parent) {
+        dependencies.spans!.record({
+          traceId: parent.traceId,
+          spanId: randomSpanId(),
+          parentSpanId: parent.parentId,
+          name: `activity ${activityType}`,
+          startTimeUnixNano: BigInt(startedAt) * 1_000_000n,
+          endTimeUnixNano: BigInt(startedAt + durationMs) * 1_000_000n,
+          attributes: allowlistSpanAttributes({
+            "temporal.activity.type": activityType,
+            "temporal.activity.id": info.activityId,
+            "temporal.activity.attempt": info.attempt,
+            "temporal.workflow.id": info.workflowExecution?.workflowId,
+            "temporal.workflow.run_id": info.workflowExecution?.runId,
+            "temporal.workflow.type": info.workflowType,
+            "space.outcome": outcome,
+            "error.code": reasonCode,
+            "space.error.category": category,
+            "space.duration_ms": durationMs,
+            "space.queue_ms": queueMs,
+            "space.trace_id": base.traceId,
+            "space.room_id": base.roomId,
+            "space.pane_id": base.paneId,
+            "space.mission_id": base.missionId,
+          }),
+          error: outcome === "FAILED" || outcome === "FAILED_RESULT",
+          kind: 1
+        });
+      }
+    } catch {
+      // Tracing is best effort.
     }
   };
 
@@ -229,6 +278,7 @@ export function createActivityObservabilityInterceptors(dependencies: Omit<Obser
     inbound: {
       execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), {
         ...dependencies,
+        parentTraceparent: dependencies.spans ? traceparentFromHeaders(input.headers) : null,
         cancelled: () => context.cancellationSignal.aborted
       })
     }

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

@@ -7,8 +7,7 @@ import {
   memoryMaintenanceWorkerOptions
 } 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 { workerObservabilityOptions, workerTracing } from "./worker-tracing.js";
 
 const service = "space-memory-maintenance-worker";
 
@@ -21,15 +20,15 @@ if (!memoryMaintenanceEnabled(process.env)) {
 
   try {
     const connection = await NativeConnection.connect({ address });
-    const spans = otlpSpanExporterFromEnv(process.env, service);
+    const tracing = workerTracing(process.env, service);
     const worker = await Worker.create({
       connection,
       namespace,
       taskQueue: MEMORY_MAINTENANCE_TASK_QUEUE,
       workflowsPath,
       activities,
       ...memoryMaintenanceWorkerOptions,
-      interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+      ...workerObservabilityOptions(tracing)
     });
     console.log(JSON.stringify({
       service,
@@ -39,8 +38,11 @@ if (!memoryMaintenanceEnabled(process.env)) {
       taskQueue: MEMORY_MAINTENANCE_TASK_QUEUE,
       activityConcurrency: memoryMaintenanceWorkerOptions.maxConcurrentActivityTaskExecutions
     }));
-    await worker.run();
-    await spans?.close();
+    try {
+      await worker.run();
+    } finally {
+      await tracing?.close();
+    }
   } catch (error) {
     console.error(JSON.stringify({
       service,

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

@@ -3,8 +3,7 @@ import { NativeConnection, Worker } from "@temporalio/worker";
 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 { workerObservabilityOptions, workerTracing } from "./worker-tracing.js";
 
 const service = "space-memory-mutation-worker";
 
@@ -14,20 +13,20 @@ if (!memoryMutationsEnabled(process.env)) {
   const address = process.env.SPACE_TEMPORAL_ADDRESS ?? DEFAULT_TEMPORAL_ADDRESS;
   const namespace = process.env.SPACE_TEMPORAL_NAMESPACE ?? DEFAULT_TEMPORAL_NAMESPACE;
   const connection = await NativeConnection.connect({ address });
-  const spans = otlpSpanExporterFromEnv(process.env, service);
+  const tracing = workerTracing(process.env, service);
   const worker = await Worker.create({
     connection,
     namespace,
     taskQueue: MEMORY_MUTATION_TASK_QUEUE,
     workflowsPath: fileURLToPath(new URL("./memory-mutation-workflows.js", import.meta.url)),
     activities: { executePersistedMemoryChangeSet },
     ...memoryMutationWorkerOptions,
-    interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+    ...workerObservabilityOptions(tracing)
   });
   console.log(JSON.stringify({ service, status: "running", address, namespace, taskQueue: MEMORY_MUTATION_TASK_QUEUE }));
   try {
     await worker.run();
   } finally {
-    await spans?.close();
+    await tracing?.close();
   }
 }

apps/worker/src/room-background-continuation.ts

@@ -1,6 +1,9 @@
+import { randomBytes } from "node:crypto";
 import {
   buildCodexAppServerTurnWorkflowId,
   buildRoomAgentSupervisorWorkflowId,
+  formatTraceparent,
+  parseTraceparent,
   roomAgentSupervisorQueueItemSchema,
   spaceAgentRoomActionRequestSchema,
   type RoomAgentActionRecord,
@@ -38,6 +41,15 @@ export function continuationEvidence(actions: RoomAgentActionRecord[], budget: n
   return JSON.stringify(result);
 }
 
+/** Fresh random root W3C traceparent; never all-zero and never the excluded trace id. */
+export function freshRootTraceparent(excludedTraceId?: string): string {
+  let traceId: string;
+  let spanId: string;
+  do { traceId = randomBytes(16).toString("hex"); } while (/^0+$/.test(traceId) || traceId === excludedTraceId);
+  do { spanId = randomBytes(8).toString("hex"); } while (/^0+$/.test(spanId));
+  return formatTraceparent(traceId, spanId, "01");
+}
+
 const activeStatuses = ["QUEUED", "RUNNING", "PAUSED"];
 export type BackgroundContinuationResult =
   | { state: "WAITING" | "DONE" }
@@ -122,10 +134,14 @@ 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));
+  const sourceTraceparent = parseTraceparent(source.queueItem.turn.traceparent);
   const item = roomAgentSupervisorQueueItemSchema.parse({ missionId: input.missionId, turn: {
     ...source.queueItem.turn,
     prompt, roomAgentOperatorRequest: request,
     roomAgentContinuation: { sourceRunId: source.run.runId, generation },
+    // Fresh root W3C context so logs and internal API calls stay joinable without a collector; the interceptor roots the workflow span on this trace id and only links the prior span.
+    traceparent: freshRootTraceparent(sourceTraceparent?.traceId),
+    continuationTraceparent: sourceTraceparent ? source.queueItem.turn.traceparent : undefined,
     artifactIds: [], agentRunId: runId,
     agentUserMessageId: `agent_msg:bg-context-${input.continuationId}`,
     agentAssistantMessageId: `agent_msg:bg-response-${input.continuationId}`,

apps/worker/src/tracing/sinks.ts

@@ -0,0 +1,20 @@
+import type { Sinks } from "@temporalio/workflow";
+
+export type WorkflowSpanOutcome = "COMPLETED" | "FAILED" | "CANCELLED" | "CONTINUED_AS_NEW";
+export type WorkflowSpanLinkReason = "continue_as_new" | "background_continuation";
+
+export interface WorkflowSpanRecord {
+  traceId: string;
+  spanId: string;
+  parentSpanId: string | null;
+  startTimeMs: number;
+  endTimeMs: number;
+  outcome: WorkflowSpanOutcome;
+  errorCode: string | null;
+  correlation: { spaceTraceId?: string; roomId?: string; paneId?: string; missionId?: string; goalId?: string };
+  links: Array<{ traceId: string; spanId: string; reason: WorkflowSpanLinkReason }>;
+}
+
+export interface SpaceTracingSinks extends Sinks {
+  spaceTracing: { workflowSpan(record: WorkflowSpanRecord): void };
+}

apps/worker/src/tracing/span-attributes.ts

@@ -0,0 +1,4 @@
+// This subpath has no Node imports or exporter timers and is safe in a workflow bundle.
+export { SPAN_ATTRIBUTE_ALLOWLIST, allowlistSpanAttributes } from "@space/runtime/span-attributes";
+export { parseTraceparent, formatTraceparent } from "@space/contracts";
+export { TRACEPARENT_HEADER as TRACE_HEADER, LINK_TRACEPARENT_HEADER as LINK_HEADER } from "../workflow-tracing-headers.js";