feat(observability): export worker and workflow spans over OTLP (#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 { 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, 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>;
@@ -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, 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" });
+        const traceparent = traceparentForTraceId(traceId);
+        const memo = workflowCorrelationMemo({ traceId, traceparent });
+        await client.workflow.start("dotGoalWorkflow", {
+          workflowId: dotGoalWorkflowId(goalId),
+          taskQueue: this.options.taskQueue,
+          args: [{ goalId }],
+          workflowIdReusePolicy: "REJECT_DUPLICATE",
+          ...(Object.keys(memo).length ? { memo } : {})
+        });
       } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; }
     });
   }
@@ -65,6 +73,7 @@ export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordi
     });
   }
 }
+/** Creates a TemporalDotGoalWorkflowCoordinator when enabled, otherwise null. */
 export function createDotGoalWorkflowCoordinator(options: { enabled: boolean; address: string; namespace: string; taskQueue: string }): DotGoalWorkflowCoordinator | null {
   return options.enabled ? new TemporalDotGoalWorkflowCoordinator(options) : null;
 }
@@ -93,19 +102,25 @@ export type DotGoalStepState =
   | { state: "STARTED" | "BUSY"; runId: string; goal: DotGoal }
   | { state: "PAUSED" | "WAITING" | "DONE"; goal: DotGoal };
 
+/** Generates a deterministic idempotency key for a specific Goal step number. */
 function requestKeyFor(goalId: string, step: number): string {
   return `goal-${createHash("sha256").update(`${goalId}:${step}`).digest("hex").slice(0, 40)}`;
 }
 
+/** Factory for DotGoalService managing Goal lifecycle, chat turns, and worker workflow steering. */
 export function createDotGoalService(deps: DotGoalServiceDeps) {
+  /** Retrieves the active SpaceAgentSession corresponding to the Goal's session. */
   async function activeSession(goal: DotGoal) {
     const session = await deps.store.getSpaceAgentSession(goal.sessionId);
     return session?.isActive && session.dotId === goal.dotId ? session : null;
   }
+
+  /** Transitions Goal state to terminal or paused and persists status reason. */
   async function finish(goal: DotGoal, status: DotGoal["status"], reason: string) {
     return (await deps.goals.transition(goal.goalId, ["ACTIVE", "PAUSED"], status, reason)) ?? goal;
   }
 
+  /** Validates prerequisites and creates a new Dot Goal, starting its Temporal workflow. */
   async function create(owner: string, dotId: string, raw: unknown, request: FastifyRequest | null): Promise<DotGoal> {
     const input = createDotGoalInputSchema.parse(raw);
     if (!deps.workflow) throw new DotGoalError(503, "DOT_GOALS_UNAVAILABLE", "Goals need the Space worker runtime.");
@@ -125,7 +140,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?.id) await deps.workflow.start(goal.goalId, String(request.id));
+      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.");
@@ -134,6 +152,7 @@ export function createDotGoalService(deps: DotGoalServiceDeps) {
     return goal;
   }
 
+  /** Pauses, resumes, or cancels an existing Goal owned by the user. */
   async function control(owner: string, dotId: string, goalId: string, raw: unknown, request: FastifyRequest | null): Promise<DotGoal> {
     const { action } = dotGoalControlInputSchema.parse(raw);
     const goal = await deps.goals.getOwned(owner, dotId, goalId);
@@ -152,6 +171,7 @@ export function createDotGoalService(deps: DotGoalServiceDeps) {
     return next;
   }
 
+  /** Appends steering guidance notes to an active Goal. */
   async function steer(owner: string, dotId: string, goalId: string, raw: unknown, request: FastifyRequest | null): Promise<DotGoal> {
     const { text } = dotGoalSteerInputSchema.parse(raw);
     const goal = await deps.goals.getOwned(owner, dotId, goalId);
@@ -268,6 +288,7 @@ export function createDotGoalService(deps: DotGoalServiceDeps) {
 }
 export type DotGoalService = ReturnType<typeof createDotGoalService>;
 
+/** Registers REST HTTP routes for managing Dot Goals (/api/dots/:dotId/goals). */
 export function registerDotGoalRoutes(app: FastifyInstance, service: DotGoalService | null, deps: { goals: DotGoalsRepository } | null, options: {
   routeOptions: RouteShorthandOptions;
   authenticatedOwner: (request: FastifyRequest) => Promise<string>;
@@ -279,6 +300,7 @@ export function registerDotGoalRoutes(app: FastifyInstance, service: DotGoalServ
     if (error instanceof z.ZodError) return reply.code(400).send({ error: { code: "DOT_GOAL_INVALID", message: "Goal request is invalid." } });
     throw error;
   };
+  /** Authenticated handler wrapper enforcing interactive user session scope for Dot Goal operations. */
   const user = (action: (request: FastifyRequest, owner: string, goals: DotGoalService, repo: DotGoalsRepository) => Promise<unknown>) =>
     async (request: FastifyRequest, reply: import("fastify").FastifyReply) => {
       if (!request.user) return reply.code(401).send({ error: { code: "UNAUTHENTICATED", message: "Sign in to use Dots." } });

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 {
@@ -33,23 +35,28 @@ export type MemoryConsolidationWorkflowStarter = (
   options: WorkflowStartOptions
 ) => Promise<WorkflowStartHandle>;
 
+/** Checks whether the error signifies that the consolidation workflow execution has already started. */
 function isAlreadyStarted(error: unknown): boolean {
   return error instanceof Error && error.name === "WorkflowExecutionAlreadyStartedError";
 }
 
+/** Schedules a memory consolidation workflow on Temporal with correlation memo and traceparent. */
 export async function scheduleMemoryConsolidation(
   rawInput: MemoryConsolidationWorkflowInput,
   workflowId: string,
   startWorkflow: MemoryConsolidationWorkflowStarter
 ): Promise<MemoryConsolidationStartResult> {
-  const input = memoryConsolidationWorkflowInputSchema.parse(rawInput);
+  // Rollout order: Deploy workers before the API so workers accept the optional
+  // traceparent field in workflow input arguments before the API starts sending it.
+  const input = withTurnTraceparent(memoryConsolidationWorkflowInputSchema.parse(rawInput));
   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",
+      ...(input.traceparent ? { memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) } : {})
     });
     return {
       status: "SCHEDULED",
@@ -93,6 +100,7 @@ class TemporalMemoryConsolidationCoordinator implements MemoryConsolidationCoord
   }
 }
 
+/** Creates a live or disabled MemoryConsolidationCoordinator based on configuration flags. */
 export function createMemoryConsolidationCoordinator(options: {
   enabled: boolean;
   address: string;

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 {
@@ -34,24 +36,29 @@ export type MemoryMutationWorkflowStarter = (
   options: WorkflowStartOptions
 ) => Promise<WorkflowStartHandle>;
 
+/** Checks whether the error signifies that the mutation workflow execution has already started. */
 function isAlreadyStarted(error: unknown): boolean {
   return error instanceof Error && error.name === "WorkflowExecutionAlreadyStartedError";
 }
 
+/** Schedules a memory mutation workflow on Temporal with correlation memo and traceparent. */
 export async function scheduleMemoryMutation(
   rawInput: MemoryMutationWorkflowInput,
   startWorkflow: MemoryMutationWorkflowStarter,
   commandHash?: string
 ): Promise<MemoryMutationStartResult> {
-  const input = memoryMutationWorkflowInputSchema.parse(rawInput);
+  // Rollout order: Deploy workers before the API so workers accept the optional
+  // traceparent field in workflow input arguments before the API starts sending it.
+  const input = withTurnTraceparent(memoryMutationWorkflowInputSchema.parse(rawInput));
   const workflowId = buildMemoryMutationWorkflowId(input.changeSetId, commandHash);
   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",
+      ...(input.traceparent ? { memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) } : {})
     });
     return { status: "SCHEDULED", workflowId: handle.workflowId, runId: handle.firstExecutionRunId ?? null };
   } catch (error) {
@@ -91,6 +98,7 @@ class TemporalMemoryMutationCoordinator implements MemoryMutationCoordinator {
   }
 }
 
+/** Creates a live or disabled MemoryMutationCoordinator based on configuration flags. */
 export function createMemoryMutationCoordinator(options: {
   enabled: boolean;
   address: string;

apps/api/src/room-agent.ts

@@ -190,14 +190,15 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo
     const lease = await this.connect("roomAgentActionWorkflow");
     try {
       try {
+        const traceparent = traceparentForTraceId(traceId);
         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
           })
@@ -226,7 +227,12 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo
         taskQueue: this.options.taskQueue,
         args: [{ roomId: item.turn.roomId, pending: [], processedCount: 0 }],
         signal: ROOM_AGENT_ENQUEUE_SIGNAL,
-        signalArgs: [item]
+        signalArgs: [item],
+        memo: workflowCorrelationMemo({
+          traceId: item.turn.traceId,
+          traceparent: item.turn.traceparent ?? null,
+          missionId: item.missionId
+        })
       });
       recordWorkflowStart(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE, "SIGNALED");
       return { workflowId, runId: handle.signaledRunId };
@@ -254,6 +260,7 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo
   }
 }
 
+/** Creates a live TemporalRoomAgentWorkflowCoordinator when enabled, otherwise a DisabledRoomAgentWorkflowCoordinator. */
 export function createRoomAgentWorkflowCoordinator(options: {
   enabled: boolean;
   address: string;

apps/api/src/trace-context.ts

@@ -21,13 +21,15 @@ export interface RequestTraceContext {
   inbound: boolean;
 }
 
+/** Generates non-zero random hex bytes of the specified length for trace/span IDs. */
 function randomHex(bytes: number): string {
   for (;;) {
     const value = randomBytes(bytes).toString("hex");
     if (!/^0+$/.test(value)) return value;
   }
 }
 
+/** Parses incoming traceparent header or generates fresh trace identifiers for an HTTP request. */
 export function createRequestTraceContext(header: unknown, random: (bytes: number) => string = randomHex): RequestTraceContext {
   const inbound = parseTraceparent(Array.isArray(header) ? header[0] : header);
   const w3cTraceId = inbound?.traceId ?? random(16);
@@ -96,7 +98,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-span-exporter.test.ts

@@ -50,6 +50,15 @@ describe("OTLP span export configuration", () => {
     expect(otlpSpanExporterFromEnv({}, "space-api")).toBeNull();
     expect(otlpSpanExporterFromEnv({ SPACE_OTEL_EXPORTER_OTLP_ENDPOINT: "javascript:alert(1)" }, "space-api")).toBeNull();
   });
+
+  it("rejects non-positive and non-finite batchSize values", () => {
+    expect(() => createOtlpSpanExporter({ endpoint: "http://127.0.0.1:4318/", serviceName: "space-api", batchSize: 0 }))
+      .toThrow("batchSize must be a positive finite number.");
+    expect(() => createOtlpSpanExporter({ endpoint: "http://127.0.0.1:4318/", serviceName: "space-api", batchSize: -1 }))
+      .toThrow("batchSize must be a positive finite number.");
+    expect(() => createOtlpSpanExporter({ endpoint: "http://127.0.0.1:4318/", serviceName: "space-api", batchSize: Number.NaN }))
+      .toThrow("batchSize must be a positive finite number.");
+  });
 });
 
 describe("OTLP span exporter", () => {

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"
   },