fix(observability): connect API and Temporal OTLP traces

CHANGELOG.md

@@ -5,6 +5,20 @@ Semantic Versioning.
 
 ## [Unreleased]
 
+### Observability
+
+- Optional OTLP tracing connects API requests to workflow executions, child
+  workflows and activity attempts in the main worker and both memory workers.
+  Each Room Agent mission stays under its own request; the long-lived supervisor
+  run has its own trace, so its background polling never joins an earlier
+  request's trace. Continued runs and background missions link to the work they
+  continue. Exported correlation attributes accept only values with a bounded
+  identifier syntax.
+- Room Agent queues preserve the originating trace across retries and recovery.
+  Workflow replay emits no spans, and a slow or failing collector does not block
+  business execution. Export remains off unless an endpoint is configured.
+- Fixed an idle exporter flush that could prevent later spans from being sent.
+
 ## [1.1.8] - 2026-10-25
 
 Source: sanitized Space app @ 8d5bf21b32 (4948 files). Verifies SpaceApp's image attestations before an

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 { traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
 
 /**
  * Space Dots Stage 4: Goals.
@@ -26,10 +27,10 @@ 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, traceId?: 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>;
+  ensure?(goalId: string, traceId?: string): Promise<boolean>;
 }
 export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordinator {
   constructor(private readonly options: { address: string; namespace: string; taskQueue: string }) {}
@@ -38,11 +39,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, traceId?: string) {
     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",
+          memo: workflowCorrelationMemo({ traceId, traceparent: traceparentForTraceId(traceId) }) });
       } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; }
     });
   }
@@ -52,11 +54,12 @@ export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordi
       catch (error) { if (!(error instanceof WorkflowNotFoundError)) throw error; }
     });
   }
-  ensure(goalId: string) {
+  ensure(goalId: string, traceId?: string) {
     return this.client(async (client) => {
       try {
         await client.workflow.start("dotGoalWorkflow", { workflowId: dotGoalWorkflowId(goalId), taskQueue: this.options.taskQueue,
-          args: [{ goalId }], workflowIdReusePolicy: "ALLOW_DUPLICATE" });
+          args: [{ goalId }], workflowIdReusePolicy: "ALLOW_DUPLICATE",
+          memo: workflowCorrelationMemo({ traceId, traceparent: traceparentForTraceId(traceId) }) });
         return true;
       } catch (error) {
         if (error instanceof WorkflowExecutionAlreadyStartedError) return false;
@@ -125,7 +128,10 @@ 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?.requestIdForSpace) await deps.workflow.start(goal.goalId, request.requestIdForSpace);
+      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 { traceparentForTraceId, 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 {
@@ -49,7 +51,8 @@ export async function scheduleMemoryConsolidation(
       taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",
-      workflowIdConflictPolicy: "USE_EXISTING"
+      workflowIdConflictPolicy: "USE_EXISTING",
+      memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: traceparentForTraceId(input.traceId) })
     });
     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 { traceparentForTraceId, 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 {
@@ -51,7 +53,8 @@ export async function scheduleMemoryMutation(
       taskQueue: MEMORY_MUTATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",
-      workflowIdConflictPolicy: "USE_EXISTING"
+      workflowIdConflictPolicy: "USE_EXISTING",
+      memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: traceparentForTraceId(input.traceId) })
     });
     return { status: "SCHEDULED", workflowId: handle.workflowId, runId: handle.firstExecutionRunId ?? null };
   } catch (error) {

apps/api/src/room-agent.ts

@@ -225,6 +225,11 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo
         workflowId,
         taskQueue: this.options.taskQueue,
         args: [{ roomId: item.turn.roomId, pending: [], processedCount: 0 }],
+        memo: workflowCorrelationMemo({
+          traceId: item.turn.traceId,
+          traceparent: item.turn.traceparent ?? null,
+          missionId: item.missionId
+        }),
         signal: ROOM_AGENT_ENQUEUE_SIGNAL,
         signalArgs: [item]
       });
@@ -661,7 +666,10 @@ export function createRoomAgentService(options: {
         selectedToolIds: [...roomAgentSelectedToolIds],
         permissionMode: "full_access",
         collaborationMode: "default",
-        traceId
+        traceId,
+        // Persist the originating span before enqueue so recovery and retries
+        // retain it after the API request's in-memory context is evicted.
+        traceparent: traceparentForTraceId(traceId) ?? undefined
       }
     });
     const childWorkflowId = buildCodexAppServerTurnWorkflowId(turn.turn);

apps/api/src/trace-context.ts

@@ -96,7 +96,9 @@ 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 span we actually export. Minting an unrecorded child here
+  // leaves every downstream workflow attached to a missing parent.
+  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/dot-goal-tracing.test.ts

@@ -0,0 +1,47 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+import { parseTraceparent } from "@space/contracts";
+import { TemporalDotGoalWorkflowCoordinator } from "../src/dot-goals.js";
+import { createRequestTraceContext, requestTraceContexts } from "../src/trace-context.js";
+
+const temporal = vi.hoisted(() => ({ start: vi.fn(), close: vi.fn() }));
+
+vi.mock("@temporalio/client", async (importOriginal) => {
+  const original = await importOriginal<typeof import("@temporalio/client")>();
+  return {
+    ...original,
+    Connection: { connect: async () => ({ close: temporal.close }) },
+    Client: class { workflow = { start: temporal.start }; }
+  };
+});
+
+beforeEach(() => {
+  temporal.start.mockReset().mockResolvedValue({});
+  temporal.close.mockReset().mockResolvedValue(undefined);
+});
+
+describe("Dot Goal workflow trace propagation", () => {
+  it("parents both explicit starts and requested restarts to their recorded API spans", async () => {
+    const coordinator = new TemporalDotGoalWorkflowCoordinator({ address: "unused", namespace: "default", taskQueue: "agent-turns" });
+    const first = createRequestTraceContext("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01");
+    const restart = createRequestTraceContext("00-11111111111111111111111111111111-2222222222222222-01");
+    requestTraceContexts.remember("req:goal-create", first);
+    requestTraceContexts.remember("req:goal-restart", restart);
+
+    await coordinator.start("dot_goal:abcdefgh1", "req:goal-create");
+    expect(await coordinator.ensure("dot_goal:abcdefgh1", "req:goal-restart")).toBe(true);
+
+    const start = temporal.start.mock.calls[0]![1];
+    const resumed = temporal.start.mock.calls[1]![1];
+    expect(start).toMatchObject({ args: [{ goalId: "dot_goal:abcdefgh1" }], workflowIdReusePolicy: "REJECT_DUPLICATE" });
+    expect(resumed).toMatchObject({ args: [{ goalId: "dot_goal:abcdefgh1" }], workflowIdReusePolicy: "ALLOW_DUPLICATE" });
+    expect(parseTraceparent(start.memo.traceparent)).toMatchObject({ traceId: first.w3cTraceId, parentId: first.spanId });
+    expect(parseTraceparent(resumed.memo.traceparent)).toMatchObject({ traceId: restart.w3cTraceId, parentId: restart.spanId });
+    expect(temporal.close).toHaveBeenCalledTimes(2);
+  });
+
+  it("does not fabricate a parent for autonomous reconciliation with no request", async () => {
+    const coordinator = new TemporalDotGoalWorkflowCoordinator({ address: "unused", namespace: "default", taskQueue: "agent-turns" });
+    await coordinator.ensure("dot_goal:abcdefgh1");
+    expect(temporal.start.mock.calls[0]![1].memo).toEqual({});
+  });
+});

apps/api/tests/dot-goals.test.ts

@@ -69,6 +69,13 @@ function fixture(overrides: { runStatus?: string; pendingApproval?: boolean; sen
 const input = { sessionId: "agent_session:1", objective: "Ship the report", milestones: ["Draft", "Review"], criteria: ["Report published"], budget: { maxSteps: 2 } };
 
 describe("Dot Goals", () => {
+  it("passes the creating API request's correlation id to the workflow starter", async () => {
+    const { service, deps } = fixture();
+    const request = { requestIdForSpace: "req:dot-goal-tracing" };
+    const goal = await service.create("user:1", dotId, input, request as never);
+    expect(deps.workflow!.start).toHaveBeenCalledWith(goal.goalId, request.requestIdForSpace);
+  });
+
   it("creates one open Goal per conversation for the owner's Dot and starts its workflow", async () => {
     const { service, deps } = fixture();
     const goal = await service.create("user:1", dotId, input, null);

apps/api/tests/memory-consolidation-coordinator.test.ts

@@ -4,11 +4,14 @@ import {
   MEMORY_CONSOLIDATION_WORKFLOW_TYPE,
   scheduleMemoryConsolidation
 } from "../src/memory-consolidation-coordinator.js";
+import { createRequestTraceContext, requestTraceContexts } from "../src/trace-context.js";
 
 describe("Memory consolidation coordinator", () => {
   it("starts a retry-safe workflow on the exclusive maintenance queue", async () => {
     const calls: Array<{ workflowType: string; options: unknown }> = [];
     const input = { runId: "memory_consolidation:run-1", traceId: "trace:memory-consolidation" };
+    const request = createRequestTraceContext("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01");
+    requestTraceContexts.remember(input.traceId, request);
     const workflowId = `space-memory-consolidation:${"a".repeat(64)}`;
 
     const result = await scheduleMemoryConsolidation(input, workflowId, async (workflowType, options) => {
@@ -24,7 +27,8 @@ describe("Memory consolidation coordinator", () => {
         taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
         workflowId,
         workflowIdReusePolicy: "REJECT_DUPLICATE",
-        workflowIdConflictPolicy: "USE_EXISTING"
+        workflowIdConflictPolicy: "USE_EXISTING",
+        memo: { spaceTraceId: input.traceId, traceparent: request.traceparent, w3cTraceId: request.w3cTraceId }
       }
     }]);
   });

apps/api/tests/memory-mutation-coordinator.test.ts

@@ -5,11 +5,14 @@ import {
   buildMemoryMutationWorkflowId,
   scheduleMemoryMutation
 } from "../src/memory-mutation-coordinator.js";
+import { createRequestTraceContext, requestTraceContexts } from "../src/trace-context.js";
 
 describe("Memory mutation coordinator", () => {
   it("starts the dedicated workflow with deterministic duplicate guards", async () => {
     const calls: Array<{ workflowType: string; options: unknown }> = [];
     const input = { changeSetId: "memory_change:approved", traceId: "req:mutation-start" };
+    const request = createRequestTraceContext("00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01");
+    requestTraceContexts.remember(input.traceId, request);
 
     const result = await scheduleMemoryMutation(input, async (workflowType, options) => {
       calls.push({ workflowType, options });
@@ -30,7 +33,8 @@ describe("Memory mutation coordinator", () => {
           taskQueue: MEMORY_MUTATION_TASK_QUEUE,
           workflowId: result.workflowId,
           workflowIdReusePolicy: "REJECT_DUPLICATE",
-          workflowIdConflictPolicy: "USE_EXISTING"
+          workflowIdConflictPolicy: "USE_EXISTING",
+          memo: { spaceTraceId: input.traceId, traceparent: request.traceparent, w3cTraceId: request.w3cTraceId }
         }
       }
     ]);