feat(worker): export workflow and activity spans over OTLP (#61)

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

apps/worker/src/activity-observability.ts

@@ -11,6 +11,7 @@ import {
   type FailureCategory
 } from "@space/contracts";
 import type { SpanExporter } from "@space/runtime";
+import { SPAN_CONTEXT_HEADER, decodeSpanContextHeader, type WorkflowSpanContextHeader } from "./workflow-span-context.js";
 import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js";
 import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js";
 
@@ -86,6 +87,7 @@ const nestedKeys = ["turn", "item", "bridge", "input", "turnInput"] as const;
 
 type CorrelationKey = (typeof correlationKeys)[number];
 
+/** Recursively collects safe correlation identifiers from activity arguments. */
 function readCorrelation(value: unknown, into: Partial<Record<CorrelationKey, string>>, depth: number): void {
   if (!value || typeof value !== "object" || Array.isArray(value)) return;
   const record = value as Record<string, unknown>;
@@ -115,6 +117,7 @@ export function activityCorrelation(args: readonly unknown[]): Record<string, st
   return correlation;
 }
 
+/** Extracts a bounded, safe error code from an unknown activity failure. */
 function errorCodeOf(error: unknown): string {
   const record = error as { code?: unknown; type?: unknown; name?: unknown } | null;
   if (isSafeErrorCode(record?.code)) return record!.code as string;
@@ -127,6 +130,7 @@ function errorCodeOf(error: unknown): string {
   return "ACTIVITY_ERROR";
 }
 
+/** Checks whether a structured activity return value represents a domain failure. */
 function isFailedResult(value: unknown): boolean {
   return Boolean(value && typeof value === "object" && (value as { status?: unknown }).status === "FAILED");
 }
@@ -137,14 +141,18 @@ export interface ObserveActivityDependencies {
   cancelled?: () => boolean;
   /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */
   spans?: SpanExporter | null;
+  /** The workflow run's span context from the space.span-context header, when the scheduler observed the call. */
+  workflowSpan?: WorkflowSpanContextHeader | null;
 }
 
+/** Extracts the first valid W3C traceparent found in activity arguments. */
 function activityTraceparent(args: readonly unknown[]) {
   const found: Partial<Record<CorrelationKey, string>> = {};
   for (const arg of args) readCorrelation(arg, found, 2);
   return parseTraceparent(found.traceparent);
 }
 
+/** Wraps a single Temporal activity execution with structured logging, metrics, and OTLP span export. */
 export async function observeActivityExecution<T>(
   info: ActivityInfo,
   args: readonly unknown[],
@@ -194,16 +202,33 @@ export async function observeActivityExecution<T>(
       queueMs,
       rssBytes: process.memoryUsage().rss
     });
-    const parent = dependencies.spans ? activityTraceparent(args) : null;
-    if (parent) {
-      dependencies.spans!.record({
+    // Prefer the workflow span the interceptor injected into the Temporal
+    // header (activities nest under their workflow); fall back to the
+    // traceparent the API propagated in the activity input.
+    const workflowSpan = dependencies.workflowSpan ?? null;
+    const parent =
+      workflowSpan && workflowSpan.kind === "parent"
+        ? { traceId: workflowSpan.traceId, parentId: workflowSpan.spanId }
+        : dependencies.spans
+          ? activityTraceparent(args)
+          : null;
+    if (parent && dependencies.spans) {
+      const spanAttributes: Record<string, string | number> = {
+        "space.activity_type": activityType,
+        "space.outcome": outcome,
+        "space.attempt": info.attempt
+      };
+      if (info.workflowType) spanAttributes["space.workflow_type"] = info.workflowType;
+      if (reasonCode) spanAttributes["space.reason_code"] = reasonCode;
+      if (category) spanAttributes["space.category"] = category;
+      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 },
+        attributes: spanAttributes,
         error: outcome === "FAILED" || outcome === "FAILED_RESULT",
         kind: 1
       });
@@ -227,10 +252,12 @@ export async function observeActivityExecution<T>(
 export function createActivityObservabilityInterceptors(dependencies: Omit<ObserveActivityDependencies, "cancelled"> = {}): ActivityInterceptorsFactory {
   return (context) => ({
     inbound: {
-      execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), {
-        ...dependencies,
-        cancelled: () => context.cancellationSignal.aborted
-      })
+      execute: (input, next) =>
+        observeActivityExecution(context.info, input.args, () => next(input), {
+          ...dependencies,
+          cancelled: () => context.cancellationSignal.aborted,
+          workflowSpan: decodeSpanContextHeader(input.headers?.[SPAN_CONTEXT_HEADER])
+        })
     }
   });
 }

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

@@ -9,6 +9,7 @@ import {
 import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js";
 import { otlpSpanExporterFromEnv } from "@space/runtime";
 import { createActivityObservabilityInterceptors } from "./activity-observability.js";
+import { workerSpanSinks } from "./workflow-spans-sink.js";
 import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js";
 
 const service = "space-memory-maintenance-worker";
@@ -23,14 +24,21 @@ if (!memoryMaintenanceEnabled(process.env)) {
   try {
     const connection = await NativeConnection.connect({ address });
     const spans = otlpSpanExporterFromEnv(process.env, service);
+    const workflowInterceptorModules = spans
+      ? [fileURLToPath(new URL("./workflow-observability-interceptors.js", import.meta.url))]
+      : [];
     const worker = await Worker.create({
       connection,
       namespace,
       taskQueue: MEMORY_MAINTENANCE_TASK_QUEUE,
       workflowsPath,
       activities,
       ...memoryMaintenanceWorkerOptions,
-      interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+      interceptors: {
+        activity: [createActivityObservabilityInterceptors({ spans })],
+        workflowModules: workflowInterceptorModules
+      },
+      sinks: workerSpanSinks(spans)
     });
     console.log(JSON.stringify({
       service,

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

@@ -5,6 +5,7 @@ import { MEMORY_MUTATION_TASK_QUEUE, memoryMutationsEnabled, memoryMutationWorke
 import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js";
 import { otlpSpanExporterFromEnv } from "@space/runtime";
 import { createActivityObservabilityInterceptors } from "./activity-observability.js";
+import { workerSpanSinks } from "./workflow-spans-sink.js";
 import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js";
 
 const service = "space-memory-mutation-worker";
@@ -16,14 +17,21 @@ if (!memoryMutationsEnabled(process.env)) {
   const namespace = process.env.SPACE_TEMPORAL_NAMESPACE ?? DEFAULT_TEMPORAL_NAMESPACE;
   const connection = await NativeConnection.connect({ address });
   const spans = otlpSpanExporterFromEnv(process.env, service);
+  const workflowInterceptorModules = spans
+    ? [fileURLToPath(new URL("./workflow-observability-interceptors.js", import.meta.url))]
+    : [];
   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 })] }
+    interceptors: {
+      activity: [createActivityObservabilityInterceptors({ spans })],
+      workflowModules: workflowInterceptorModules
+    },
+    sinks: workerSpanSinks(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

@@ -1,15 +1,18 @@
 import {
   buildCodexAppServerTurnWorkflowId,
   buildRoomAgentSupervisorWorkflowId,
+  parseTraceparent,
   roomAgentSupervisorQueueItemSchema,
   spaceAgentRoomActionRequestSchema,
+  traceparentForTraceId,
   type RoomAgentActionRecord,
   type RoomAgentSupervisorQueueItem
 } from "@space/contracts";
 import { SpaceConflictError, redactMemoryText, type SpaceStore } from "@space/runtime";
 import { roomActionSummary } from "./room-action-summary.js";
 import { DEFAULT_ROOM_AGENT_MAX_CONTINUATIONS, DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS, ROOM_AGENT_CONTINUATION_LIMIT, ROOM_AGENT_TOKEN_LIMIT } from "./room-agent-limits.js";
 
+/** Summarizes completed action records within the given character budget for background continuation. */
 export function continuationEvidence(actions: RoomAgentActionRecord[], budget: number): string {
   // Count the entire mission, including synchronous actions. Background roots
   // are only a scheduling distinction, not the record of completed work.
@@ -45,13 +48,15 @@ export type BackgroundContinuationResult =
 
 interface MissionTokenUsage { usedTokens: number; countedThroughRunId: string }
 
+/** Reads normalized token usage totals from a persisted turn execution state, if present. */
 function recordedTokenUsage(executionState: Record<string, unknown>): MissionTokenUsage | null {
   const value = executionState.tokenUsage as Partial<MissionTokenUsage> | undefined;
   return value && Number.isSafeInteger(value.usedTokens) && (value.usedTokens as number) >= 0 && typeof value.countedThroughRunId === "string"
     ? { usedTokens: value.usedTokens as number, countedThroughRunId: value.countedThroughRunId }
     : null;
 }
 
+/** Evaluates and schedules a bounded background continuation turn when an agent turn hits its budget. */
 export async function prepareBackgroundContinuation(input: {
   roomId: string; missionId: string; continuationId: string;
 }, store: SpaceStore, maxContinuations = DEFAULT_ROOM_AGENT_MAX_CONTINUATIONS,
@@ -122,6 +127,10 @@ 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 continuationTraceId = `trace:bg-${input.continuationId}`;
+  const sourceTraceparent = parseTraceparent(source.queueItem.turn.traceparent)
+    ? source.queueItem.turn.traceparent
+    : traceparentForTraceId(source.queueItem.turn.traceId);
   const item = roomAgentSupervisorQueueItemSchema.parse({ missionId: input.missionId, turn: {
     ...source.queueItem.turn,
     prompt, roomAgentOperatorRequest: request,
@@ -130,7 +139,12 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis
     agentUserMessageId: `agent_msg:bg-context-${input.continuationId}`,
     agentAssistantMessageId: `agent_msg:bg-response-${input.continuationId}`,
     // Resolve the current native thread from the durable session when executing.
-    agentThreadId: null, traceId: `trace:bg-${input.continuationId}`
+    agentThreadId: null, traceId: continuationTraceId,
+    // A continuation always starts its own deterministic W3C trace and links
+    // to the run it continues (the source turn's traceparent, or one derived
+    // from its Space trace id when the source had no inbound header).
+    traceparent: traceparentForTraceId(continuationTraceId),
+    ...(sourceTraceparent ? { sourceTraceparent } : {})
   } });
   try {
     const enqueued = await store.enqueueRoomAgentMission({
@@ -150,6 +164,7 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis
 }
 
 // Without an errorCode the turn was superseded or stopped; with one it failed and keeps that cause.
+/** Marks a queued supervisor turn and its mission as discarded when it cannot be executed. */
 export async function discardSupervisorTurn(item: RoomAgentSupervisorQueueItem, reason: string, store: SpaceStore, errorCode?: string) {
   const record = await store.getRoomAgentTurn(item.missionId, item.turn.agentRunId);
   if (!record || !["QUEUED", "RUNNING"].includes(record.run.status)) return;
@@ -161,6 +176,7 @@ export async function discardSupervisorTurn(item: RoomAgentSupervisorQueueItem,
   await store.updateSpaceAgentMessage(record.run.responseMessageId, { status, content: reason });
 }
 
+/** Resolves whether a queued supervisor turn is ready to run, paused on mission state, or should be skipped. */
 export async function supervisorTurnState(item: RoomAgentSupervisorQueueItem, store: SpaceStore): Promise<"READY" | "PAUSED" | "SKIP"> {
   const mission = await store.getRoomAgentMission(item.turn.roomId, item.missionId);
   const record = await store.getRoomAgentTurn(item.missionId, item.turn.agentRunId);
@@ -181,6 +197,7 @@ export async function supervisorTurnState(item: RoomAgentSupervisorQueueItem, st
   return "READY";
 }
 
+/** Finalizes a supervisor turn and decides whether to schedule a background continuation. */
 export async function finishSupervisorTurn(input: {
   item: RoomAgentSupervisorQueueItem; status: "COMPLETED" | "FAILED" | "INTERRUPTED"; statusReason: string; errorCode?: string;
 }, store: SpaceStore) {

apps/worker/src/worker.ts

@@ -7,6 +7,7 @@ import { TelegramDeliveryWorker, TelegramSecretStore, otlpSpanExporterFromEnv }
 import { NativeConnection, Runtime, Worker } from "@temporalio/worker";
 import * as activities from "./activities.js";
 import { createActivityObservabilityInterceptors } from "./activity-observability.js";
+import { workerSpanSinks } from "./workflow-spans-sink.js";
 import { resolveTemporalCoreMetricsBind, resolveWorkerMetricsListen, startWorkerMetricsServer } from "./worker-metrics.js";
 import { workerLog } from "./worker-log.js";
 import { sweepCodexCliCompletions } from "./codex-cli-completion.js";
@@ -22,6 +23,7 @@ const namespace = process.env.SPACE_TEMPORAL_NAMESPACE ?? DEFAULT_TEMPORAL_NAMES
 const taskQueue = process.env.SPACE_TEMPORAL_TASK_QUEUE ?? DEFAULT_TASK_QUEUE;
 const workflowsPath = fileURLToPath(new URL("./workflows.js", import.meta.url));
 
+/** Parses a polling interval in milliseconds and clamps it between 1s and 60s. */
 function boundedInterval(value: string | undefined): number {
   const parsed = Number.parseInt(value ?? "2000", 10);
   return Number.isFinite(parsed) ? Math.max(1_000, Math.min(parsed, 60_000)) : 2_000;
@@ -37,15 +39,24 @@ if (temporalCoreMetricsBind) {
 
 const connection = await NativeConnection.connect({ address });
 const spans = otlpSpanExporterFromEnv(process.env, "space-worker");
+// Structured, correlated activity records and worker metrics, plus workflow
+// spans: the interceptor module runs inside the workflow sandbox but exports
+// only through the workerSpans sink (callDuringReplay: false), so replay
+// behaviour and determinism are unchanged. Off unless the endpoint is set.
+const workflowInterceptorModules = spans
+  ? [fileURLToPath(new URL("./workflow-observability-interceptors.js", import.meta.url))]
+  : [];
 const worker = await Worker.create({
   connection,
   namespace,
   taskQueue,
   workflowsPath,
   activities,
-  // Structured, correlated activity records and worker metrics. Activity-side
-  // only: the workflow sandbox and its replay behaviour are unchanged.
-  interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] }
+  interceptors: {
+    activity: [createActivityObservabilityInterceptors({ spans })],
+    workflowModules: workflowInterceptorModules
+  },
+  sinks: workerSpanSinks(spans)
 });
 
 // Optional loopback-only Prometheus endpoint for worker metrics.