Trace API requests through Temporal workflows and activities

.github/workflows/ci.yml

@@ -15,6 +15,30 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  worker-tracing:
+    name: Connected Temporal tracing
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
+        with:
+          node-version-file: .nvmrc
+          cache: npm
+      - name: Create sanitized tracing candidate
+        working-directory: ${{ github.workspace }}
+        run: node scripts/public-export.mjs --output "$RUNNER_TEMP/spaceapp-tracing-export"
+      - run: ELECTRON_SKIP_BINARY_DOWNLOAD=1 npm ci
+        working-directory: ${{ runner.temp }}/spaceapp-tracing-export
+      - run: npm run check
+        working-directory: ${{ runner.temp }}/spaceapp-tracing-export
+      - run: npm run telemetry:check -w @space/worker
+        working-directory: ${{ runner.temp }}/spaceapp-tracing-export
+      - run: npm test -w @space/worker -- tests/worker-telemetry.test.ts tests/activity-observability.test.ts
+        working-directory: ${{ runner.temp }}/spaceapp-tracing-export
+
   public-quality:
     name: Sanitized public quality gates
     runs-on: ubuntu-latest

apps/api/src/dot-goals.ts

@@ -1,4 +1,5 @@
 import { createHash } from "node:crypto";
+import { traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js";
 import { z } from "zod";
 import { Client, Connection, WorkflowExecutionAlreadyStartedError, WorkflowNotFoundError } from "@temporalio/client";
 import type { FastifyInstance, FastifyRequest, RouteShorthandOptions } from "fastify";
@@ -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,11 @@ 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 }], memo: workflowCorrelationMemo({ traceId, traceparent: traceparentForTraceId(traceId) }), workflowIdReusePolicy: "REJECT_DUPLICATE" });
       } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; }
     });
   }
@@ -52,11 +53,11 @@ 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 }], memo: workflowCorrelationMemo({ traceId, traceparent: traceparentForTraceId(traceId) }), workflowIdReusePolicy: "ALLOW_DUPLICATE" });
         return true;
       } catch (error) {
         if (error instanceof WorkflowExecutionAlreadyStartedError) return false;
@@ -125,7 +126,7 @@ 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 { await deps.workflow.start(goal.goalId, request?.id); }
     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 };
 
@@ -16,6 +17,7 @@ export interface MemoryConsolidationStartResult {
 }
 
 interface WorkflowStartOptions {
+  memo: Record<string, string>;
   args: [MemoryConsolidationWorkflowInput];
   taskQueue: typeof MEMORY_CONSOLIDATION_TASK_QUEUE;
   workflowId: string;
@@ -46,6 +48,7 @@ export async function scheduleMemoryConsolidation(
   try {
     const handle = await startWorkflow(MEMORY_CONSOLIDATION_WORKFLOW_TYPE, {
       args: [input],
+      memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: traceparentForTraceId(input.traceId) }),
       taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",

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 };
 
@@ -17,6 +18,7 @@ export interface MemoryMutationStartResult {
 }
 
 interface WorkflowStartOptions {
+  memo: Record<string, string>;
   args: [MemoryMutationWorkflowInput];
   taskQueue: typeof MEMORY_MUTATION_TASK_QUEUE;
   workflowId: string;
@@ -48,6 +50,7 @@ export async function scheduleMemoryMutation(
   try {
     const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, {
       args: [input],
+      memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: traceparentForTraceId(input.traceId) }),
       taskQueue: MEMORY_MUTATION_TASK_QUEUE,
       workflowId,
       workflowIdReusePolicy: "REJECT_DUPLICATE",

apps/api/src/room-agent.ts

@@ -225,6 +225,7 @@ 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, missionId: item.missionId }),
         signal: ROOM_AGENT_ENQUEUE_SIGNAL,
         signalArgs: [item]
       });

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);
+  // A propagated parent must name the exported HTTP span. Creating another
+  // unrecorded span here would leave every 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-goals.test.ts

@@ -73,7 +73,7 @@ describe("Dot Goals", () => {
     const { service, deps } = fixture();
     const goal = await service.create("user:1", dotId, input, null);
     expect(goal).toMatchObject({ status: "ACTIVE", budget: { maxSteps: 2, maxMinutes: 180 }, milestones: [{ id: "m1", title: "Draft" }, { id: "m2" }] });
-    expect(deps.workflow!.start).toHaveBeenCalledWith(goal.goalId);
+    expect(deps.workflow!.start).toHaveBeenCalledWith(goal.goalId, undefined);
     await expect(service.create("user:1", dotId, input, null)).rejects.toMatchObject({ status: 409, code: "DOT_GOAL_OPEN" });
     await expect(service.create("user:2", dotId, input, null)).rejects.toBeInstanceOf(DotGoalError);
   });

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

@@ -21,6 +21,7 @@ describe("Memory consolidation coordinator", () => {
       workflowType: MEMORY_CONSOLIDATION_WORKFLOW_TYPE,
       options: {
         args: [input],
+          memo: { spaceTraceId: input.traceId },
         taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE,
         workflowId,
         workflowIdReusePolicy: "REJECT_DUPLICATE",

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

@@ -27,6 +27,7 @@ describe("Memory mutation coordinator", () => {
         workflowType: MEMORY_MUTATION_WORKFLOW_TYPE,
         options: {
           args: [input],
+          memo: { spaceTraceId: input.traceId },
           taskQueue: MEMORY_MUTATION_TASK_QUEUE,
           workflowId: result.workflowId,
           workflowIdReusePolicy: "REJECT_DUPLICATE",

apps/api/tests/otlp-span-exporter.test.ts

@@ -53,6 +53,21 @@ describe("OTLP span export configuration", () => {
 });
 
 describe("OTLP span exporter", () => {
+  it("exports new spans after an empty idle flush and after a drained batch", async () => {
+    const fetchImpl = vi.fn(async () => ({ ok: true }));
+    const created = exporter(fetchImpl);
+    await created.flush();
+    await created.flush();
+    created.record(span());
+    await created.flush();
+    expect(fetchImpl).toHaveBeenCalledTimes(1);
+    await created.flush();
+    created.record(span({ spanId: "1111111111111111" }));
+    await created.close();
+    expect(fetchImpl).toHaveBeenCalledTimes(2);
+    expect(created.stats()).toEqual({ exported: 2, dropped: 0, failed: 0 });
+  });
+
   it("posts OTLP/HTTP JSON server spans with only route-level attributes", async () => {
     const fetchImpl = vi.fn(async () => ({ ok: true }));
     const created = exporter(fetchImpl);

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" });
@@ -203,6 +204,7 @@ describe("workflow starters", () => {
     expect(options.workflowId).toContain(turn.traceId);
     expect(options.args[0]?.traceId).toBe(turn.traceId);
     expect(parseTraceparent(options.args[0]?.traceparent)?.traceId).toBe(inboundTraceId);
+    expect(options.args[0]?.traceparent).toBe(requestTraceContexts.lookup(turn.traceId)?.traceparent);
     expect(options.memo).toMatchObject({ spaceTraceId: turn.traceId, w3cTraceId: inboundTraceId });
     expect(JSON.stringify(options.memo)).not.toContain("PRIVATE PROMPT");
     expect(close).toHaveBeenCalledOnce();

apps/worker/package.json

@@ -10,7 +10,8 @@
     "start": "node dist/worker.js",
     "build": "tsc -p tsconfig.json",
     "check": "tsc -p tsconfig.json --noEmit",
-    "test": "vitest run"
+    "test": "vitest run",
+    "telemetry:check": "tsc -p tsconfig.telemetry-tests.json"
   },
   "dependencies": {
     "@space/codex-app-server": "0.2.5",
@@ -23,11 +24,14 @@
     "@temporalio/client": "1.24.0",
     "@temporalio/worker": "1.24.0",
     "@temporalio/workflow": "1.24.0",
-    "undici": "8.12.3"
+    "undici": "8.12.3",
+    "@temporalio/common": "1.24.0"
   },
   "devDependencies": {
     "typescript": "^5.9.3",
-    "vitest": "^4.1.12"
+    "vitest": "^4.1.12",
+    "@temporalio/testing": "1.24.0",
+    "@temporalio/proto": "1.24.0"
   },
   "license": "MIT"
 }

apps/worker/src/activity-observability.ts

@@ -30,6 +30,7 @@ export type ActivityOutcome = "COMPLETED" | "FAILED_RESULT" | "FAILED" | "CANCEL
 
 interface ActivityObservation {
   reasonCode: string | null;
+  traceparent?: string;
 }
 
 const observations = new AsyncLocalStorage<ActivityObservation>();
@@ -55,8 +56,10 @@ export function failureReasonScope<T>(execute: () => Promise<T>): { promise: Pro
   return { promise: observations.run(observation, execute), reasonCode: () => observation.reasonCode };
 }
 
-/** traceparent for an outbound call: same W3C trace, new span id; null if none. */
+/** Parent an outbound HTTP call to the observed activity span when available. */
 export function childTraceparent(traceparent: string | null | undefined): string | null {
+  const current = observations.getStore()?.traceparent;
+  if (current && parseTraceparent(current)) return current;
   const parsed = parseTraceparent(traceparent);
   if (!parsed) return null;
   let spanId = randomBytes(8).toString("hex");
@@ -135,6 +138,11 @@ export interface ObserveActivityDependencies {
   metrics?: WorkerMetrics;
   now?: () => number;
   cancelled?: () => boolean;
+  traceparent?: string;
+  onFinish?: (observation: {
+    outcome: ActivityOutcome; reasonCode: string | null; category: FailureCategory | null;
+    startTimeMs: number; endTimeMs: number; durationMs: number;
+  }) => void;
   /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */
   spans?: SpanExporter | null;
 }
@@ -172,7 +180,7 @@ export async function observeActivityExecution<T>(
   if (queueMs !== null) metrics.activityScheduleToStart.observe(labels, queueMs / 1000);
   metrics.activitiesInFlight.inc(labels);
   workerLog("info", "activity.started", { ...base, queueMs });
-  const observation: ActivityObservation = { reasonCode: null };
+  const observation: ActivityObservation = { reasonCode: null, traceparent: dependencies.traceparent };
 
   const finish = (outcome: ActivityOutcome, reasonCode: string | null, errorName: string | null) => {
     const durationMs = Math.max(0, now() - startedAt);
@@ -194,7 +202,11 @@ export async function observeActivityExecution<T>(
       queueMs,
       rssBytes: process.memoryUsage().rss
     });
-    const parent = dependencies.spans ? activityTraceparent(args) : null;
+    // Telemetry is best effort; an observer must never change activity results
+    // or create a retry (and must not return a promise to await).
+    try { dependencies.onFinish?.({ outcome, reasonCode, category, startTimeMs: startedAt, endTimeMs: startedAt + durationMs, durationMs }); }
+    catch { /* Keep activity execution independent from instrumentation. */ }
+    const parent = !dependencies.onFinish && dependencies.spans ? activityTraceparent(args) : null;
     if (parent) {
       dependencies.spans!.record({
         traceId: parent.traceId,

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

@@ -1,4 +1,6 @@
 import { fileURLToPath } from "node:url";
+import { createWorkerTelemetry } from "./worker-telemetry.js";
+import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js";
 import { NativeConnection, Worker } from "@temporalio/worker";
 import * as activities from "./memory-maintenance-activities.js";
 import {
@@ -7,9 +9,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 { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js";
+
 
 const service = "space-memory-maintenance-worker";
 
@@ -20,17 +20,19 @@ if (!memoryMaintenanceEnabled(process.env)) {
   const namespace = process.env.SPACE_TEMPORAL_NAMESPACE ?? DEFAULT_TEMPORAL_NAMESPACE;
   const workflowsPath = fileURLToPath(new URL("./memory-maintenance-workflows.js", import.meta.url));
 
+  const telemetry = createWorkerTelemetry({ serviceName: service });
+  let connection: NativeConnection | undefined;
+  let stopReloadWatch = () => {};
   try {
-    const connection = await NativeConnection.connect({ address });
-    const spans = otlpSpanExporterFromEnv(process.env, service);
+    connection = await NativeConnection.connect({ address });
     const worker = await Worker.create({
       connection,
       namespace,
       taskQueue: MEMORY_MAINTENANCE_TASK_QUEUE,
       workflowsPath,
       activities,
       ...memoryMaintenanceWorkerOptions,
-      interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+      ...telemetry.workerOptions
     });
     console.log(JSON.stringify({
       service,
@@ -40,16 +42,19 @@ if (!memoryMaintenanceEnabled(process.env)) {
       taskQueue: MEMORY_MAINTENANCE_TASK_QUEUE,
       activityConcurrency: memoryMaintenanceWorkerOptions.maxConcurrentActivityTaskExecutions
     }));
-    const stopReloadWatch = reloadWhenServicesCommitChanges(worker, { log: (message) => console.log(JSON.stringify({ service, status: "reloading", message })) });
+    stopReloadWatch = reloadWhenServicesCommitChanges(worker, { log: (message) => console.log(JSON.stringify({ service, status: "reloading", message })) });
     await worker.run();
-    stopReloadWatch();
-    await spans?.close();
+
   } catch (error) {
     console.error(JSON.stringify({
       service,
       status: "failed",
       error: error instanceof Error ? error.message : "Unknown maintenance worker failure."
     }));
     process.exitCode = 1;
+  } finally {
+    stopReloadWatch();
+    try { await telemetry.close(); }
+    finally { await connection?.close(); }
   }
 }