From 09c613272396007ea0b7882fc21d31d4c53ce0ab Mon Sep 17 00:00:00 2001 From: Hrishikesh Yadav Date: Wed, 28 Oct 2026 21:46:16 +0530 Subject: [PATCH] feat(worker): export worker and workflow spans over OTLP (fixes #61) --- apps/worker/package.json | 1 + apps/worker/src/activity-observability.ts | 39 +++- apps/worker/src/memory-maintenance-worker.ts | 8 +- apps/worker/src/memory-mutation-worker.ts | 8 +- apps/worker/src/worker.ts | 10 +- apps/worker/src/workflow-interceptors.ts | 168 ++++++++++++++ apps/worker/src/workflow-sinks.ts | 72 ++++++ apps/worker/src/workflow-span-registry.ts | 46 ++++ .../tests/activity-observability.test.ts | 100 ++++++++- .../tests/workflow-observability.test.ts | 202 +++++++++++++++++ .../tests/workflow-trace-otlp-e2e.test.ts | 209 ++++++++++++++++++ docs/quality/observability.md | 47 ++-- package-lock.json | 5 +- packages/runtime/src/otlp-span-exporter.ts | 24 +- 14 files changed, 896 insertions(+), 43 deletions(-) create mode 100644 apps/worker/src/workflow-interceptors.ts create mode 100644 apps/worker/src/workflow-sinks.ts create mode 100644 apps/worker/src/workflow-span-registry.ts create mode 100644 apps/worker/tests/workflow-observability.test.ts create mode 100644 apps/worker/tests/workflow-trace-otlp-e2e.test.ts diff --git a/apps/worker/package.json b/apps/worker/package.json index ec58125fa..66ceaacaa 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -18,6 +18,7 @@ "@space/db": "0.2.5", "@space/memory-graph": "0.2.5", "@space/opencode-control": "0.2.5", + "@space/runtime": "0.2.5", "@space/streaming": "0.2.5", "@temporalio/activity": "1.24.0", "@temporalio/client": "1.24.0", diff --git a/apps/worker/src/activity-observability.ts b/apps/worker/src/activity-observability.ts index ea4d021d1..bc7030333 100644 --- a/apps/worker/src/activity-observability.ts +++ b/apps/worker/src/activity-observability.ts @@ -1,6 +1,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { randomBytes } from "node:crypto"; import type { Context } from "@temporalio/activity"; +import { defaultPayloadConverter } from "@temporalio/common"; import type { ActivityInterceptorsFactory } from "@temporalio/worker"; import { categorizeFailureCode, @@ -8,11 +9,13 @@ import { isSafeErrorCode, parseTraceparent, traceIdFromTraceparent, - type FailureCategory + type FailureCategory, + type ParsedTraceparent } from "@space/contracts"; import type { SpanExporter } from "@space/runtime"; import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js"; import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js"; +import { workflowSpanRegistry } from "./workflow-span-registry.js"; /** * One structured start/finish record and metrics per activity execution, with @@ -137,6 +140,7 @@ export interface ObserveActivityDependencies { cancelled?: () => boolean; /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */ spans?: SpanExporter | null; + headers?: Record; } function activityTraceparent(args: readonly unknown[]) { @@ -194,16 +198,38 @@ export async function observeActivityExecution( queueMs, rssBytes: process.memoryUsage().rss }); - const parent = dependencies.spans ? activityTraceparent(args) : null; - if (parent) { - dependencies.spans!.record({ + let parent: ParsedTraceparent | null = null; + if (dependencies.headers?.traceparent) { + try { + const raw = defaultPayloadConverter.fromPayload(dependencies.headers.traceparent as never); + if (typeof raw === "string") parent = parseTraceparent(raw); + } catch {} + } + if (!parent && info.workflowExecution?.runId) { + const registered = workflowSpanRegistry.lookup(info.workflowExecution.runId); + if (registered) { + parent = { traceId: registered.traceId, parentId: registered.spanId, flags: "01" }; + } + } + if (!parent) { + parent = activityTraceparent(args); + } + if (parent && dependencies.spans) { + let spanId = randomBytes(8).toString("hex"); + while (/^0+$/.test(spanId)) spanId = randomBytes(8).toString("hex"); + dependencies.spans.record({ traceId: parent.traceId, - spanId: randomBytes(8).toString("hex"), + spanId, 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: { + "space.activity_type": activityType, + "space.outcome": outcome, + "space.attempt": info.attempt, + ...(reasonCode ? { "space.reason_code": reasonCode, "space.category": category ?? "UNSPECIFIED" } : {}) + }, error: outcome === "FAILED" || outcome === "FAILED_RESULT", kind: 1 }); @@ -229,6 +255,7 @@ export function createActivityObservabilityInterceptors(dependencies: Omit observeActivityExecution(context.info, input.args, () => next(input), { ...dependencies, + headers: input.headers, cancelled: () => context.cancellationSignal.aborted }) } diff --git a/apps/worker/src/memory-maintenance-worker.ts b/apps/worker/src/memory-maintenance-worker.ts index e12046033..35e973bde 100644 --- a/apps/worker/src/memory-maintenance-worker.ts +++ b/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 { createWorkflowTraceSinks } from "./workflow-sinks.js"; import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js"; const service = "space-memory-maintenance-worker"; @@ -23,6 +24,7 @@ if (!memoryMaintenanceEnabled(process.env)) { try { const connection = await NativeConnection.connect({ address }); const spans = otlpSpanExporterFromEnv(process.env, service); + const workflowInterceptorsPath = fileURLToPath(new URL("./workflow-interceptors.js", import.meta.url)); const worker = await Worker.create({ connection, namespace, @@ -30,7 +32,11 @@ if (!memoryMaintenanceEnabled(process.env)) { workflowsPath, activities, ...memoryMaintenanceWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + sinks: createWorkflowTraceSinks(spans), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + workflowModules: [workflowInterceptorsPath] + } }); console.log(JSON.stringify({ service, diff --git a/apps/worker/src/memory-mutation-worker.ts b/apps/worker/src/memory-mutation-worker.ts index c2eadf657..d24913db7 100644 --- a/apps/worker/src/memory-mutation-worker.ts +++ b/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 { createWorkflowTraceSinks } from "./workflow-sinks.js"; import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js"; const service = "space-memory-mutation-worker"; @@ -16,6 +17,7 @@ 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 workflowInterceptorsPath = fileURLToPath(new URL("./workflow-interceptors.js", import.meta.url)); const worker = await Worker.create({ connection, namespace, @@ -23,7 +25,11 @@ if (!memoryMutationsEnabled(process.env)) { workflowsPath: fileURLToPath(new URL("./memory-mutation-workflows.js", import.meta.url)), activities: { executePersistedMemoryChangeSet }, ...memoryMutationWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + sinks: createWorkflowTraceSinks(spans), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + workflowModules: [workflowInterceptorsPath] + } }); 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 })) }); diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 83d23cbc6..c08ff7365 100644 --- a/apps/worker/src/worker.ts +++ b/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 { createWorkflowTraceSinks } from "./workflow-sinks.js"; import { resolveTemporalCoreMetricsBind, resolveWorkerMetricsListen, startWorkerMetricsServer } from "./worker-metrics.js"; import { workerLog } from "./worker-log.js"; import { sweepCodexCliCompletions } from "./codex-cli-completion.js"; @@ -37,15 +38,18 @@ if (temporalCoreMetricsBind) { const connection = await NativeConnection.connect({ address }); const spans = otlpSpanExporterFromEnv(process.env, "space-worker"); +const workflowInterceptorsPath = fileURLToPath(new URL("./workflow-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 })] } + sinks: createWorkflowTraceSinks(spans), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + workflowModules: [workflowInterceptorsPath] + } }); // Optional loopback-only Prometheus endpoint for worker metrics. diff --git a/apps/worker/src/workflow-interceptors.ts b/apps/worker/src/workflow-interceptors.ts new file mode 100644 index 000000000..2b1dec4ed --- /dev/null +++ b/apps/worker/src/workflow-interceptors.ts @@ -0,0 +1,168 @@ +import { + defaultPayloadConverter, + proxySinks, + uuid4, + workflowInfo, + type ActivityInput, + type ContinueAsNewInput, + type StartChildWorkflowExecutionInput, + type WorkflowExecuteInput, + type WorkflowInboundCallsInterceptor, + type WorkflowInterceptors, + type WorkflowOutboundCallsInterceptor +} from "@temporalio/workflow"; +import { formatTraceparent, parseTraceparent, type ParsedTraceparent } from "@space/contracts"; +import type { WorkflowTraceSinks } from "./workflow-sinks.js"; + +const { trace } = proxySinks(); + +interface ActiveTraceContext { + traceId: string; + spanId: string; + parentSpanId: string | null; +} + +export class TracingWorkflowInboundInterceptor implements WorkflowInboundCallsInterceptor { + private activeContext: ActiveTraceContext | null = null; + + getActiveContext(): ActiveTraceContext | null { + return this.activeContext; + } + + async execute(input: WorkflowExecuteInput, next: (input: WorkflowExecuteInput) => Promise): Promise { + const info = workflowInfo(); + const startedAtMs = info.runStartTime.getTime(); + + let parentParsed: ParsedTraceparent | null = null; + + if (input.headers.traceparent) { + try { + const raw = defaultPayloadConverter.fromPayload(input.headers.traceparent); + parentParsed = parseTraceparent(raw); + } catch {} + } + + if (!parentParsed && typeof info.memo?.traceparent === "string") { + parentParsed = parseTraceparent(info.memo.traceparent); + } + + if (!parentParsed && input.args[0] && typeof input.args[0] === "object") { + const arg = input.args[0] as Record; + const candidate = arg.traceparent ?? (arg.turn as Record | undefined)?.traceparent ?? (arg.bridge as Record | undefined)?.traceparent; + if (typeof candidate === "string") { + parentParsed = parseTraceparent(candidate); + } + } + + if (!parentParsed && input.headers.continuationTraceparent) { + try { + const raw = defaultPayloadConverter.fromPayload(input.headers.continuationTraceparent); + parentParsed = parseTraceparent(raw); + } catch {} + } + + if (parentParsed) { + const workflowSpanId = uuid4().replace(/-/g, "").slice(0, 16); + this.activeContext = { + traceId: parentParsed.traceId, + spanId: workflowSpanId, + parentSpanId: parentParsed.parentId + }; + + trace.recordWorkflowStart({ + traceId: this.activeContext.traceId, + spanId: this.activeContext.spanId, + parentSpanId: this.activeContext.parentSpanId, + workflowType: info.workflowType, + workflowId: info.workflowId, + runId: info.runId, + startedAtMs + }); + } + + try { + const result = await next(input); + if (this.activeContext) { + trace.recordWorkflowSpan({ + ...this.activeContext, + workflowType: info.workflowType, + workflowId: info.workflowId, + runId: info.runId, + startedAtMs, + endedAtMs: Date.now(), + status: "COMPLETED" + }); + } + return result; + } catch (error) { + if (this.activeContext) { + trace.recordWorkflowSpan({ + ...this.activeContext, + workflowType: info.workflowType, + workflowId: info.workflowId, + runId: info.runId, + startedAtMs, + endedAtMs: Date.now(), + status: "FAILED", + errorName: error instanceof Error ? error.name : "WorkflowError" + }); + } + throw error; + } + } +} + +export class TracingWorkflowOutboundInterceptor implements WorkflowOutboundCallsInterceptor { + constructor(private readonly inbound: TracingWorkflowInboundInterceptor) {} + + scheduleActivity(input: ActivityInput, next: (input: ActivityInput) => Promise): Promise { + const ctx = this.inbound.getActiveContext(); + if (ctx) { + const traceparent = formatTraceparent(ctx.traceId, ctx.spanId); + const headers = { + ...input.headers, + traceparent: defaultPayloadConverter.toPayload(traceparent) + }; + return next({ ...input, headers }); + } + return next(input); + } + + startChildWorkflowExecution( + input: StartChildWorkflowExecutionInput, + next: (input: StartChildWorkflowExecutionInput) => Promise<[Promise, Promise]> + ): Promise<[Promise, Promise]> { + const ctx = this.inbound.getActiveContext(); + if (ctx) { + const traceparent = formatTraceparent(ctx.traceId, ctx.spanId); + const headers = { + ...input.headers, + traceparent: defaultPayloadConverter.toPayload(traceparent) + }; + return next({ ...input, headers }); + } + return next(input); + } + + continueAsNew(input: ContinueAsNewInput, next: (input: ContinueAsNewInput) => Promise): Promise { + const ctx = this.inbound.getActiveContext(); + if (ctx) { + const traceparent = formatTraceparent(ctx.traceId, ctx.spanId); + const headers = { + ...input.headers, + continuationTraceparent: defaultPayloadConverter.toPayload(traceparent) + }; + return next({ ...input, headers }); + } + return next(input); + } +} + +export function interceptors(): WorkflowInterceptors { + const inbound = new TracingWorkflowInboundInterceptor(); + const outbound = new TracingWorkflowOutboundInterceptor(inbound); + return { + inbound: [inbound], + outbound: [outbound] + }; +} diff --git a/apps/worker/src/workflow-sinks.ts b/apps/worker/src/workflow-sinks.ts new file mode 100644 index 000000000..c8b3881c5 --- /dev/null +++ b/apps/worker/src/workflow-sinks.ts @@ -0,0 +1,72 @@ +import type { InjectedSinks } from "@temporalio/worker"; +import type { Sinks, WorkflowInfo } from "@temporalio/workflow"; +import type { ServerSpan, SpanExporter } from "@space/runtime"; +import { workflowSpanRegistry } from "./workflow-span-registry.js"; + +export interface WorkflowStartPayload { + traceId: string; + spanId: string; + parentSpanId: string | null; + workflowType: string; + workflowId: string; + runId: string; + startedAtMs: number; +} + +export interface WorkflowSpanPayload extends WorkflowStartPayload { + endedAtMs: number; + status: "COMPLETED" | "FAILED" | "CONTINUED_AS_NEW"; + errorName?: string | null; + attributes?: Record; +} + +export interface WorkflowTraceSinks extends Sinks { + trace: { + recordWorkflowStart(payload: WorkflowStartPayload): void; + recordWorkflowSpan(payload: WorkflowSpanPayload): void; + }; +} + +export function createWorkflowTraceSinks(spans: SpanExporter | null): InjectedSinks { + return { + trace: { + recordWorkflowStart: { + fn(_info: WorkflowInfo, payload: WorkflowStartPayload) { + workflowSpanRegistry.register(payload.runId, { + traceId: payload.traceId, + spanId: payload.spanId, + workflowType: payload.workflowType + }); + }, + callDuringReplay: false + }, + recordWorkflowSpan: { + fn(_info: WorkflowInfo, payload: WorkflowSpanPayload) { + workflowSpanRegistry.evict(payload.runId); + if (!spans) return; + + const span: ServerSpan = { + traceId: payload.traceId, + spanId: payload.spanId, + parentSpanId: payload.parentSpanId, + name: `workflow ${payload.workflowType}`, + startTimeUnixNano: BigInt(payload.startedAtMs) * 1_000_000n, + endTimeUnixNano: BigInt(payload.endedAtMs) * 1_000_000n, + attributes: { + "space.workflow_type": payload.workflowType, + "space.workflow_id": payload.workflowId, + "space.temporal_run_id": payload.runId, + "space.outcome": payload.status, + ...(payload.attributes ?? {}) + }, + error: payload.status === "FAILED", + kind: 1 // INTERNAL + }; + + spans.record(span); + }, + callDuringReplay: false + } + } + }; +} diff --git a/apps/worker/src/workflow-span-registry.ts b/apps/worker/src/workflow-span-registry.ts new file mode 100644 index 000000000..c78e30327 --- /dev/null +++ b/apps/worker/src/workflow-span-registry.ts @@ -0,0 +1,46 @@ +/** + * In-memory registry mapping Temporal workflow execution runIds to their active span context. + * Used by the worker process so activity interceptors can automatically parent activity + * spans to their workflow span without needing traceparent explicitly in activity arguments. + */ +export interface RegisteredWorkflowSpan { + traceId: string; + spanId: string; + workflowType: string; +} + +export class WorkflowSpanRegistry { + private readonly spans = new Map(); + + constructor(private readonly capacity = 8192) {} + + register(runId: string, span: RegisteredWorkflowSpan): void { + if (!runId) return; + this.spans.delete(runId); + this.spans.set(runId, span); + while (this.spans.size > this.capacity) { + const oldest = this.spans.keys().next().value; + if (oldest === undefined) break; + this.spans.delete(oldest); + } + } + + lookup(runId: string | null | undefined): RegisteredWorkflowSpan | null { + if (!runId) return null; + return this.spans.get(runId) ?? null; + } + + evict(runId: string | null | undefined): void { + if (runId) this.spans.delete(runId); + } + + clear(): void { + this.spans.clear(); + } + + get size(): number { + return this.spans.size; + } +} + +export const workflowSpanRegistry = new WorkflowSpanRegistry(); diff --git a/apps/worker/tests/activity-observability.test.ts b/apps/worker/tests/activity-observability.test.ts index 654bf7962..05b007f61 100644 --- a/apps/worker/tests/activity-observability.test.ts +++ b/apps/worker/tests/activity-observability.test.ts @@ -186,7 +186,9 @@ it("registers the activity interceptor on every Temporal worker, including both const { readFile } = await import("node:fs/promises"); for (const entry of ["worker.ts", "memory-maintenance-worker.ts", "memory-mutation-worker.ts"]) { const source = await readFile(new URL(`../src/${entry}`, import.meta.url), "utf8"); - expect(source, entry).toMatch(/interceptors: \{ activity: \[createActivityObservabilityInterceptors\(\{ spans \}\)\] \}/); + expect(source, entry).toMatch(/activity:\s*\[createActivityObservabilityInterceptors\(\{ spans \}\)\]/); + expect(source, entry).toMatch(/workflowModules:\s*\[workflowInterceptorsPath\]/); + expect(source, entry).toMatch(/sinks:\s*createWorkflowTraceSinks\(spans\)/); } }); @@ -203,3 +205,99 @@ it("exports one INTERNAL span per activity under the caller's traceparent", asyn expect(recorded[0]).toMatchObject({ traceId: "4bf92f3577b34da6a3ce929d0e0e4736", parentSpanId: "00f067aa0ba902b7", name: "activity runTurn", kind: 1, error: false, startTimeUnixNano: 1_000_000_000n, endTimeUnixNano: 1_250_000_000n }); }); + +it("exports an error span for a failing retryable attempt with reasonCode and category", async () => { + const recorded: Array> = []; + const spans = { record: (s: Record) => recorded.push(s), flush: async () => undefined, close: async () => undefined, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; + const metrics = createWorkerMetrics(new MetricsRegistry()); + const parentTraceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-1122334455667788-01"; + const info = { + activityType: "fetchExternalResource", + activityId: "1", + attempt: 1, + currentAttemptScheduledTimestampMs: 1_000, + workflowExecution: { workflowId: "wf-1", runId: "run-1" } + } as never; + + const thrown = Object.assign(new Error("upstream timeout"), { code: "UPSTREAM_RETRYABLE_UNAVAILABLE" }); + + await expect( + observeActivityExecution( + info, + [{ turn: { traceparent: parentTraceparent } }], + async () => { throw thrown; }, + { metrics, spans: spans as never } + ) + ).rejects.toBe(thrown); + + expect(recorded).toHaveLength(1); + const [span] = recorded; + expect(span.traceId).toBe("4bf92f3577b34da6a3ce929d0e0e4736"); + expect(span.parentSpanId).toBe("1122334455667788"); + expect(span.name).toBe("activity fetchExternalResource"); + expect(span.kind).toBe(1); + expect(span.error).toBe(true); + expect(span.attributes).toMatchObject({ + "space.activity_type": "fetchExternalResource", + "space.attempt": 1, + "space.outcome": "FAILED", + "space.reason_code": "UPSTREAM_RETRYABLE_UNAVAILABLE", + "space.category": "UPSTREAM_UNAVAILABLE" + }); +}); + +it("exports consecutive spans across retried attempts, ending in a completed span", async () => { + const recorded: Array> = []; + const spans = { record: (s: Record) => recorded.push(s), flush: async () => undefined, close: async () => undefined, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; + const metrics = createWorkerMetrics(new MetricsRegistry()); + const parentTraceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-1122334455667788-01"; + + // Attempt 1: Fails + const info1 = { activityType: "runTask", activityId: "1", attempt: 1, currentAttemptScheduledTimestampMs: 1000, workflowExecution: { workflowId: "w", runId: "r" } } as never; + await expect( + observeActivityExecution(info1, [{ traceparent: parentTraceparent }], async () => { + throw Object.assign(new Error("temporarily down"), { code: "TOOL_UNAVAILABLE" }); + }, { metrics, spans: spans as never }) + ).rejects.toThrow(); + + // Attempt 2: Succeeds + const info2 = { activityType: "runTask", activityId: "1", attempt: 2, currentAttemptScheduledTimestampMs: 2000, workflowExecution: { workflowId: "w", runId: "r" } } as never; + await observeActivityExecution(info2, [{ traceparent: parentTraceparent }], async () => { + return { status: "COMPLETED" }; + }, { metrics, spans: spans as never }); + + expect(recorded).toHaveLength(2); + const span0Attrs = recorded[0].attributes as Record; + const span1Attrs = recorded[1].attributes as Record; + expect(span0Attrs["space.attempt"]).toBe(1); + expect(recorded[0].error).toBe(true); + expect(span0Attrs["space.outcome"]).toBe("FAILED"); + expect(span0Attrs["space.category"]).toBe("TOOL_UNAVAILABLE"); + + expect(span1Attrs["space.attempt"]).toBe(2); + expect(recorded[1].error).toBe(false); + expect(span1Attrs["space.outcome"]).toBe("COMPLETED"); +}); + +it("exports an error span for a noted FAILED_RESULT with explicit reasonCode and category", async () => { + const recorded: Array> = []; + const spans = { record: (s: Record) => recorded.push(s), flush: async () => undefined, close: async () => undefined, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; + const metrics = createWorkerMetrics(new MetricsRegistry()); + const parentTraceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-1122334455667788-01"; + const info = { activityType: "runCodexAppServerTurn", activityId: "1", attempt: 1, currentAttemptScheduledTimestampMs: 1000, workflowExecution: { workflowId: "w", runId: "r" } } as never; + + await observeActivityExecution(info, [{ traceparent: parentTraceparent }], async () => { + noteActivityFailureReason("CODEX_APP_SERVER_TURN_TIMEOUT"); + return { status: "FAILED" }; + }, { metrics, spans: spans as never }); + + expect(recorded).toHaveLength(1); + expect(recorded[0].error).toBe(true); + expect(recorded[0].attributes).toMatchObject({ + "space.activity_type": "runCodexAppServerTurn", + "space.attempt": 1, + "space.outcome": "FAILED_RESULT", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT" + }); +}); diff --git a/apps/worker/tests/workflow-observability.test.ts b/apps/worker/tests/workflow-observability.test.ts new file mode 100644 index 000000000..9188b7363 --- /dev/null +++ b/apps/worker/tests/workflow-observability.test.ts @@ -0,0 +1,202 @@ +import { describe, expect, it, vi } from "vitest"; +import { defaultPayloadConverter } from "@temporalio/common"; +import type { ServerSpan, SpanExporter } from "@space/runtime"; +import { createWorkflowTraceSinks } from "../src/workflow-sinks.js"; +import { workflowSpanRegistry } from "../src/workflow-span-registry.js"; +import { + TracingWorkflowInboundInterceptor, + TracingWorkflowOutboundInterceptor +} from "../src/workflow-interceptors.js"; + +describe("workflow trace sinks", () => { + it("protects replay determinism with callDuringReplay: false", () => { + const sinks = createWorkflowTraceSinks(null); + expect(sinks.trace.recordWorkflowStart.callDuringReplay).toBe(false); + expect(sinks.trace.recordWorkflowSpan.callDuringReplay).toBe(false); + }); + + it("registers in workflowSpanRegistry on start and emits INTERNAL span on completion", () => { + const recorded: ServerSpan[] = []; + const exporter: SpanExporter = { + record: (span) => void recorded.push(span), + flush: async () => undefined, + close: async () => undefined, + stats: () => ({ exported: recorded.length, dropped: 0, failed: 0 }) + }; + + const sinks = createWorkflowTraceSinks(exporter); + const runId = "test-run-123"; + + sinks.trace.recordWorkflowStart.fn({} as never, { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "00f067aa0ba902b7", + parentSpanId: "1122334455667788", + workflowType: "dummyTurnWorkflow", + workflowId: "wf-dummy-1", + runId, + startedAtMs: 1_000 + }); + + const registered = workflowSpanRegistry.lookup(runId); + expect(registered).toEqual({ + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "00f067aa0ba902b7", + workflowType: "dummyTurnWorkflow" + }); + + sinks.trace.recordWorkflowSpan.fn({} as never, { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "00f067aa0ba902b7", + parentSpanId: "1122334455667788", + workflowType: "dummyTurnWorkflow", + workflowId: "wf-dummy-1", + runId, + startedAtMs: 1_000, + endedAtMs: 1_500, + status: "COMPLETED" + }); + + expect(workflowSpanRegistry.lookup(runId)).toBeNull(); + expect(recorded).toHaveLength(1); + expect(recorded[0]).toMatchObject({ + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "00f067aa0ba902b7", + parentSpanId: "1122334455667788", + name: "workflow dummyTurnWorkflow", + kind: 1, + error: false, + startTimeUnixNano: 1_000_000_000n, + endTimeUnixNano: 1_500_000_000n, + attributes: { + "space.workflow_type": "dummyTurnWorkflow", + "space.workflow_id": "wf-dummy-1", + "space.temporal_run_id": runId, + "space.outcome": "COMPLETED" + } + }); + }); + + it("emits error span on FAILED workflow completion", () => { + const recorded: ServerSpan[] = []; + const exporter: SpanExporter = { + record: (span) => void recorded.push(span), + flush: async () => undefined, + close: async () => undefined, + stats: () => ({ exported: recorded.length, dropped: 0, failed: 0 }) + }; + + const sinks = createWorkflowTraceSinks(exporter); + const runId = "test-run-fail"; + + sinks.trace.recordWorkflowSpan.fn({} as never, { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "00f067aa0ba902b7", + parentSpanId: null, + workflowType: "roomAgentSupervisorWorkflow", + workflowId: "wf-supervisor-1", + runId, + startedAtMs: 1_000, + endedAtMs: 2_000, + status: "FAILED", + errorName: "ApplicationFailure" + }); + + expect(recorded).toHaveLength(1); + expect(recorded[0].error).toBe(true); + expect(recorded[0].attributes["space.outcome"]).toBe("FAILED"); + }); +}); + +describe("workflow outbound calls interceptor", () => { + it("injects traceparent header into scheduled activities", async () => { + const inbound = new TracingWorkflowInboundInterceptor(); + const outbound = new TracingWorkflowOutboundInterceptor(inbound); + + // Simulate active context + (inbound as any).activeContext = { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "a1b2c3d4e5f60718", + parentSpanId: null + }; + + let scheduledInput: any; + await outbound.scheduleActivity( + { + activityType: "sampleActivity", + args: [], + headers: {}, + options: {} as never, + seq: 1 + }, + async (input) => { + scheduledInput = input; + return "done"; + } + ); + + expect(scheduledInput).toBeDefined(); + expect(scheduledInput.headers.traceparent).toBeDefined(); + const decoded = defaultPayloadConverter.fromPayload(scheduledInput.headers.traceparent); + expect(decoded).toBe("00-4bf92f3577b34da6a3ce929d0e0e4736-a1b2c3d4e5f60718-01"); + }); + + it("injects traceparent header into child workflows", async () => { + const inbound = new TracingWorkflowInboundInterceptor(); + const outbound = new TracingWorkflowOutboundInterceptor(inbound); + + (inbound as any).activeContext = { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "childparent12345", + parentSpanId: null + }; + + let childInput: any; + await outbound.startChildWorkflowExecution( + { + workflowType: "memoryConsolidationWorkflow", + args: [], + headers: {}, + options: {} as never + }, + async (input) => { + childInput = input; + return [Promise.resolve("wf-child-id"), Promise.resolve("result")]; + } + ); + + expect(childInput).toBeDefined(); + expect(childInput.headers.traceparent).toBeDefined(); + const decoded = defaultPayloadConverter.fromPayload(childInput.headers.traceparent); + expect(decoded).toBe("00-4bf92f3577b34da6a3ce929d0e0e4736-childparent12345-01"); + }); + + it("injects continuationTraceparent header into continueAsNew", async () => { + const inbound = new TracingWorkflowInboundInterceptor(); + const outbound = new TracingWorkflowOutboundInterceptor(inbound); + + (inbound as any).activeContext = { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "continue12345678", + parentSpanId: null + }; + + let continueInput: any; + await expect( + outbound.continueAsNew( + { + args: [], + headers: {} + }, + async (input) => { + continueInput = input; + throw new Error("ContinueAsNewExecuted"); + } + ) + ).rejects.toThrow("ContinueAsNewExecuted"); + + expect(continueInput).toBeDefined(); + expect(continueInput.headers.continuationTraceparent).toBeDefined(); + const decoded = defaultPayloadConverter.fromPayload(continueInput.headers.continuationTraceparent); + expect(decoded).toBe("00-4bf92f3577b34da6a3ce929d0e0e4736-continue12345678-01"); + }); +}); diff --git a/apps/worker/tests/workflow-trace-otlp-e2e.test.ts b/apps/worker/tests/workflow-trace-otlp-e2e.test.ts new file mode 100644 index 000000000..2ef3f1708 --- /dev/null +++ b/apps/worker/tests/workflow-trace-otlp-e2e.test.ts @@ -0,0 +1,209 @@ +import http from "node:http"; +import { fileURLToPath } from "node:url"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { ApplicationFailure } from "@temporalio/activity"; +import { TestWorkflowEnvironment } from "@temporalio/testing"; +import { Worker } from "@temporalio/worker"; +import { createOtlpSpanExporter } from "@space/runtime"; +import type { DummyTurnInput, DummyTurnResult } from "@space/contracts"; +import { createActivityObservabilityInterceptors } from "../src/activity-observability.js"; +import { createWorkflowTraceSinks } from "../src/workflow-sinks.js"; + +interface CapturedOtlpSpan { + traceId: string; + spanId: string; + parentSpanId?: string; + name: string; + kind: number; + startTimeUnixNano: string; + endTimeUnixNano: string; + attributes: Array<{ key: string; value: { stringValue?: string; intValue?: string } }>; + status: { code: number }; +} + +function createInMemoryOtlpCollector() { + const spans: CapturedOtlpSpan[] = []; + const server = http.createServer((req, res) => { + if (req.method === "POST" && req.url === "/v1/traces") { + let raw = ""; + req.on("data", (chunk) => { + raw += chunk; + }); + req.on("end", () => { + try { + const parsed = JSON.parse(raw); + const extracted: CapturedOtlpSpan[] = + parsed.resourceSpans?.flatMap((rs: any) => + rs.scopeSpans?.flatMap((ss: any) => ss.spans ?? []) ?? [] + ) ?? []; + spans.push(...extracted); + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ partialSuccess: {} })); + } catch { + res.writeHead(400); + res.end(); + } + }); + } else { + res.writeHead(404); + res.end(); + } + }); + + return { + get spans() { + return spans; + }, + async start(): Promise { + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + const addr = server.address() as { address: string; port: number }; + return `http://${addr.address}:${addr.port}`; + }, + async close(): Promise { + await new Promise((resolve, reject) => server.close((err) => (err ? reject(err) : resolve()))); + } + }; +} + +describe("OTLP Connected Trace Hierarchy E2E", () => { + let collector: ReturnType; + let env: TestWorkflowEnvironment; + let worker: Worker; + let spanExporter: ReturnType; + + const rootTraceId = "4bf92f3577b34da6a3ce929d0e0e4736"; + const apiServerSpanId = "00f067aa0ba902b7"; + const inboundTraceparent = `00-${rootTraceId}-${apiServerSpanId}-01`; + const taskQueue = "test-otlp-trace-queue"; + + let recordStartAttempts = 0; + + beforeAll(async () => { + collector = createInMemoryOtlpCollector(); + const endpoint = await collector.start(); + + spanExporter = createOtlpSpanExporter({ + endpoint, + serviceName: "space-worker", + flushIntervalMs: 50 + }); + + env = await TestWorkflowEnvironment.createTimeSkipping(); + + const workflowsPath = fileURLToPath(new URL("../src/workflows.ts", import.meta.url)); + const workflowInterceptorsPath = fileURLToPath(new URL("../src/workflow-interceptors.ts", import.meta.url)); + + const testActivities = { + recordDummyTurnStarted: async (_input: DummyTurnInput): Promise => { + recordStartAttempts += 1; + if (recordStartAttempts === 1) { + throw ApplicationFailure.retryable("Temporary resource glitch", "TOOL_UNAVAILABLE"); + } + }, + recordDummyTurnCompleted: async (_input: DummyTurnInput): Promise => { + return { + status: "COMPLETED", + providerId: "dummy", + model: "test-model", + totalTimeMs: 100, + tokenUsage: { inputTokens: 10, outputTokens: 20, totalTokens: 30 } + }; + } + }; + + worker = await Worker.create({ + connection: env.nativeConnection, + namespace: env.namespace, + taskQueue, + workflowsPath, + activities: testActivities, + sinks: createWorkflowTraceSinks(spanExporter), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans: spanExporter })], + workflowModules: [workflowInterceptorsPath] + } + }); + }); + + afterAll(async () => { + await spanExporter?.close(); + await collector?.close(); + await env?.teardown(); + }); + + it("exports a complete connected trace tree: API -> workflow -> activity retry attempts -> completion", async () => { + const input: DummyTurnInput = { + turnId: "turn-otlp-1", + roomId: "room-otlp-1", + traceId: `req:${rootTraceId.slice(0, 16)}`, + traceparent: inboundTraceparent, + clientSessionId: "sess-1", + prompt: "hello" + }; + + const handle = await env.client.workflow.start("dummyTurnWorkflow", { + taskQueue, + workflowId: `wf-dummy-otlp-${Date.now()}`, + args: [input], + memo: { + traceparent: inboundTraceparent, + spaceTraceId: input.traceId, + w3cTraceId: rootTraceId + } + }); + + await worker.runUntil(handle.result()); + await spanExporter.flush(); + + const spans = collector.spans; + expect(spans.length).toBeGreaterThanOrEqual(4); + + // 1. All spans must share the exact same 32-hex traceId propagated from API + for (const span of spans) { + expect(span.traceId).toBe(rootTraceId); + } + + // 2. Workflow span + const wfSpan = spans.find((s) => s.name === "workflow dummyTurnWorkflow"); + expect(wfSpan).toBeDefined(); + expect(wfSpan!.parentSpanId).toBe(apiServerSpanId); // Child of API server request + expect(wfSpan!.kind).toBe(1); // INTERNAL + expect(wfSpan!.status.code).toBe(0); // OK + + // 3. Activity attempt 1 (failed with retryable error) + const attempt1 = spans.find( + (s) => + s.name === "activity recordDummyTurnStarted" && + s.attributes.some((a) => a.key === "space.attempt" && a.value.intValue === "1") + ); + expect(attempt1).toBeDefined(); + expect(attempt1!.parentSpanId).toBe(wfSpan!.spanId); // Parent is the workflow span! + expect(attempt1!.status.code).toBe(2); // ERROR + expect(attempt1!.attributes).toEqual( + expect.arrayContaining([ + { key: "space.outcome", value: { stringValue: "FAILED" } }, + { key: "space.reason_code", value: { stringValue: "TOOL_UNAVAILABLE" } }, + { key: "space.category", value: { stringValue: "TOOL_UNAVAILABLE" } } + ]) + ); + + // 4. Activity attempt 2 (succeeded retry) + const attempt2 = spans.find( + (s) => + s.name === "activity recordDummyTurnStarted" && + s.attributes.some((a) => a.key === "space.attempt" && a.value.intValue === "2") + ); + expect(attempt2).toBeDefined(); + expect(attempt2!.parentSpanId).toBe(wfSpan!.spanId); // Parent is the workflow span! + expect(attempt2!.status.code).toBe(0); // OK + expect(attempt2!.attributes).toEqual( + expect.arrayContaining([{ key: "space.outcome", value: { stringValue: "COMPLETED" } }]) + ); + + // 5. Subsequent activity (recordDummyTurnCompleted) + const completedAct = spans.find((s) => s.name === "activity recordDummyTurnCompleted"); + expect(completedAct).toBeDefined(); + expect(completedAct!.parentSpanId).toBe(wfSpan!.spanId); // Parent is the workflow span! + expect(completedAct!.status.code).toBe(0); + }); +}); diff --git a/docs/quality/observability.md b/docs/quality/observability.md index f58323438..d020e4727 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -215,27 +215,36 @@ Without the endpoint, every activity still writes `activity.started` and ## Traces (W3C trace context) -The trace context is local by default. An optional OTLP/HTTP (JSON) exporter sends one -SERVER span per API request when `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is set (no -OpenTelemetry dependency; `packages/runtime/src/otlp-span-exporter.ts`). Since -2026-10-16 the worker and both memory workers use the same exporter (services -`space-worker`, `space-memory-maintenance-worker`, `space-memory-mutation-worker`): -one INTERNAL span per activity whose input carries a `traceparent`, as a child of -that span, with the activity type, outcome and attempt only. +The trace context is local by default. An optional OTLP/HTTP (JSON) exporter sends spans +when `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is set (no OpenTelemetry SDK dependency; +`packages/runtime/src/otlp-span-exporter.ts`), maintaining a bounded queue, dropping on +overflow, applying a POST timeout, and producing zero delay on failure. When unset, +export is completely inactive with zero overhead. + +Every Temporal worker (`worker.ts`, `memory-maintenance-worker.ts`, +`memory-mutation-worker.ts`) and the API server share this exporter: - **API:** a valid inbound `traceparent` is continued (same trace ID, new span). An invalid or missing header gets a new trace ID. Every API log line carries `w3cTraceId` (Fastify `childLoggerFactory`). - **Propagation into turns:** API turn starters and the Room Agent coordinator add - the request's `traceparent` to the turn input (an optional contract field) and - to the Temporal memo. A bounded in-memory map from request ID to trace context - (`apps/api/src/trace-context.ts`) does this without passing the request object - through every service. -- **Worker:** worker logs read the trace ID from the activity input. Internal - room-action calls from the worker send a child `traceparent`, so the internal - API request joins the same trace. -- **Workflow code is unchanged.** Correlation is done in the API starter - (input/memo) and in an activity interceptor. The workflow bundle only gains the - optional contract field. + the request's `traceparent` to the turn input and to the Temporal memo. +- **Workflow spans:** Workflow inbound interceptors (`TracingWorkflowInboundInterceptor`) + extract the inbound `traceparent` (from headers, memo, args, or continuation headers) + and assign a deterministic 16-character hex span ID. Temporal sinks (`recordWorkflowStart`, + `recordWorkflowSpan`) with `callDuringReplay: false` emit spans outside the workflow sandbox + upon workflow start and completion/failure, preserving strict workflow determinism. + `workflowSpanRegistry` maintains an LRU mapping of run IDs to active span IDs outside the sandbox. +- **Activity attempt spans:** Outbound workflow interceptors inject `traceparent` payload headers + into activity invocations. `observeActivityExecution` exports one INTERNAL span per activity attempt, + child of the workflow span, reporting `space.attempt`, `space.outcome`, and on failure + `space.reason_code` and `space.category`. Retried attempts export consecutive failure spans ending + with the concluding attempt's outcome. +- **Child workflows & Continuations:** Outbound interceptors propagate `traceparent` headers to + `startChildWorkflowExecution` (e.g. `memoryMaintenanceWorkflow` → `memoryConsolidationWorkflow`) + nesting child workflow spans under parent workflow spans. Background continuations (`continueAsNew`) + pass `continuationTraceparent` linking the new run to the preceding workflow span. +- **Worker-to-API calls:** Internal room-action calls from the worker send a child `traceparent`, + so the internal API request joins the same trace. ### Replay compatibility @@ -290,7 +299,9 @@ No credential or transcript export was added. | `apps/api/tests/observability-metrics.test.ts` | New API metrics, `/metrics` access rule, redaction, CLI disconnect and reconnect counters | | `apps/api/src/__tests__/files-search-errors.test.ts` | `EMPTY_RESULT`/`OK` and each failure code | | `apps/api/tests/files-search-routes.test.ts` | The HTTP error envelope | -| `apps/worker/tests/activity-observability.test.ts` | Interceptor records, outcomes, categories, metrics, loopback endpoint | +| `apps/worker/tests/activity-observability.test.ts` | Interceptor records, outcomes, categories, metrics, retry attempts, loopback endpoint | +| `apps/worker/tests/workflow-observability.test.ts` | Interceptor replay determinism, workflow start/completion sinks, failure status, outbound header injection | +| `apps/worker/tests/workflow-trace-otlp-e2e.test.ts` | End-to-end trace tree: API traceparent -> workflow span -> activity retry attempts -> completion with in-memory OTLP collector | | `apps/api/tests/observability-fault-injection.test.ts` | Four injected executor faults and one Temporal start failure, end to end | ## Not covered diff --git a/package-lock.json b/package-lock.json index 04bf09e25..7d8ce7677 100644 --- a/package-lock.json +++ b/package-lock.json @@ -184,6 +184,7 @@ "@space/db": "0.2.5", "@space/memory-graph": "0.2.5", "@space/opencode-control": "0.2.5", + "@space/runtime": "0.2.5", "@space/streaming": "0.2.5", "@temporalio/activity": "1.24.0", "@temporalio/client": "1.24.0", @@ -11641,12 +11642,12 @@ } }, "packages/run-spaceapp": { - "version": "1.0.35", + "version": "1.1.8", "cpu": [ "x64", "arm64" ], - "license": "Apache-2.0", + "license": "MIT", "os": [ "darwin", "linux", diff --git a/packages/runtime/src/otlp-span-exporter.ts b/packages/runtime/src/otlp-span-exporter.ts index 86a0edf05..0a3226954 100644 --- a/packages/runtime/src/otlp-span-exporter.ts +++ b/packages/runtime/src/otlp-span-exporter.ts @@ -120,17 +120,19 @@ export function createOtlpSpanExporter(options: OtlpSpanExporterOptions): SpanEx stats.failed += batch.length; } }; - const flush = (): Promise => { - if (!inFlight) { - inFlight = (async () => { - try { - while (queue.length) await flushOnce(); - } finally { - inFlight = null; - } - })(); - } - return inFlight; + const flush = async (): Promise => { + while (inFlight) await inFlight; + if (!queue.length) return; + let promise: Promise | null = null; + promise = (async () => { + try { + while (queue.length) await flushOnce(); + } finally { + if (inFlight === promise) inFlight = null; + } + })(); + inFlight = promise; + await inFlight; }; const timer = setInterval(() => void flush(), options.flushIntervalMs ?? 5000); timer.unref?.();