From 9ff8784d4f7707d3bd3dc5eb33d9aca3395a6d5a Mon Sep 17 00:00:00 2001 From: Yusef Atteyih Date: Sat, 24 Oct 2026 20:30:50 +0300 Subject: [PATCH 1/4] Connect API, Temporal workflows and activity attempts in OTLP traces Signed-off-by: Yusef Atteyih --- apps/api/src/trace-context.ts | 3 +- apps/api/tests/trace-correlation.test.ts | 1 + apps/worker/package.json | 1 + apps/worker/src/activity-observability.ts | 14 +- apps/worker/src/memory-maintenance-worker.ts | 4 +- apps/worker/src/memory-mutation-worker.ts | 4 +- .../src/room-background-continuation.ts | 9 +- apps/worker/src/worker.ts | 7 +- apps/worker/src/workflow-trace-context.ts | 68 ++++++++ apps/worker/src/workflow-tracing-worker.ts | 26 +++ apps/worker/src/workflow-tracing.ts | 63 +++++++ .../tests/activity-observability.test.ts | 13 +- .../tests/fixtures/tracing-workflows.ts | 21 +++ .../room-background-continuation.test.ts | 12 ++ .../worker/tests/room-continuation-fixture.ts | 4 +- apps/worker/tests/workflow-tracing.test.ts | 156 ++++++++++++++++++ apps/worker/vitest.config.ts | 1 + docs/configuration-reference.md | 10 ++ docs/quality/observability.md | 58 ++++++- package-lock.json | 24 ++- packages/contracts/package.json | 4 + packages/contracts/src/schemas.ts | 1 + packages/runtime/src/otlp-span-exporter.ts | 13 +- .../runtime/tests/otlp-span-exporter.test.ts | 41 +++++ 24 files changed, 535 insertions(+), 23 deletions(-) create mode 100644 apps/worker/src/workflow-trace-context.ts create mode 100644 apps/worker/src/workflow-tracing-worker.ts create mode 100644 apps/worker/src/workflow-tracing.ts create mode 100644 apps/worker/tests/fixtures/tracing-workflows.ts create mode 100644 apps/worker/tests/workflow-tracing.test.ts create mode 100644 packages/runtime/tests/otlp-span-exporter.test.ts diff --git a/apps/api/src/trace-context.ts b/apps/api/src/trace-context.ts index af9e55efd..f41f21367 100644 --- a/apps/api/src/trace-context.ts +++ b/apps/api/src/trace-context.ts @@ -96,7 +96,8 @@ export const requestTraceContexts = new TraceContextRegistry(); /** traceparent for work started on behalf of this Space request id, if known. */ export function traceparentForTraceId(traceId: string | null | undefined, registry = requestTraceContexts): string | null { - return childTraceparent(registry.lookup(traceId)?.traceparent ?? null); + // Propagate the recorded SERVER span, not an unrecorded intermediate span. + return registry.lookup(traceId)?.traceparent ?? null; } /** Adds the request's traceparent to a turn input unless it already has a valid one. */ diff --git a/apps/api/tests/trace-correlation.test.ts b/apps/api/tests/trace-correlation.test.ts index ff8a47758..61d7e49fd 100644 --- a/apps/api/tests/trace-correlation.test.ts +++ b/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" }); diff --git a/apps/worker/package.json b/apps/worker/package.json index d7a484ac8..936ad0ea0 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -26,6 +26,7 @@ "undici": "8.10.2" }, "devDependencies": { + "@temporalio/testing": "1.24.0", "typescript": "^5.9.3", "vitest": "^4.1.12" }, diff --git a/apps/worker/src/activity-observability.ts b/apps/worker/src/activity-observability.ts index ea4d021d1..33603d80f 100644 --- a/apps/worker/src/activity-observability.ts +++ b/apps/worker/src/activity-observability.ts @@ -1,3 +1,4 @@ +import { traceFromHeaders, safeTraceAttributes } from "./workflow-trace-context.js"; import { AsyncLocalStorage } from "node:async_hooks"; import { randomBytes } from "node:crypto"; import type { Context } from "@temporalio/activity"; @@ -137,6 +138,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; + traceparent?: string; } function activityTraceparent(args: readonly unknown[]) { @@ -194,19 +196,22 @@ export async function observeActivityExecution( queueMs, rssBytes: process.memoryUsage().rss }); - const parent = dependencies.spans ? activityTraceparent(args) : null; + const parent = dependencies.spans ? (parseTraceparent(dependencies.traceparent) ?? activityTraceparent(args)) : null; if (parent) { - dependencies.spans!.record({ + try { 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: safeTraceAttributes({ "space.activity_type": activityType, "space.outcome": outcome, "space.attempt": info.attempt, + "space.activity_id": info.activityId, "space.workflow_id": info.workflowExecution?.workflowId, + "space.workflow_run_id": info.workflowExecution?.runId, "space.workflow_type": info.workflowType, + "space.reason_code": reasonCode, "space.category": category, "space.duration_ms": durationMs }), error: outcome === "FAILED" || outcome === "FAILED_RESULT", kind: 1 - }); + }); } catch { /* Tracing must not change the activity result or retry policy. */ } } }; @@ -229,6 +234,7 @@ export function createActivityObservabilityInterceptors(dependencies: Omit observeActivityExecution(context.info, input.args, () => next(input), { ...dependencies, + traceparent: traceFromHeaders(input.headers) ?? dependencies.traceparent, cancelled: () => context.cancellationSignal.aborted }) } diff --git a/apps/worker/src/memory-maintenance-worker.ts b/apps/worker/src/memory-maintenance-worker.ts index e12046033..d1996594f 100644 --- a/apps/worker/src/memory-maintenance-worker.ts +++ b/apps/worker/src/memory-maintenance-worker.ts @@ -1,3 +1,4 @@ +import { workflowTracingOptions } from "./workflow-tracing-worker.js"; import { fileURLToPath } from "node:url"; import { NativeConnection, Worker } from "@temporalio/worker"; import * as activities from "./memory-maintenance-activities.js"; @@ -30,7 +31,8 @@ if (!memoryMaintenanceEnabled(process.env)) { workflowsPath, activities, ...memoryMaintenanceWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + interceptors: { activity: [createActivityObservabilityInterceptors({ spans })], workflowModules: workflowTracingOptions(spans).workflowModules }, + sinks: workflowTracingOptions(spans).sinks }); 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..a596680ff 100644 --- a/apps/worker/src/memory-mutation-worker.ts +++ b/apps/worker/src/memory-mutation-worker.ts @@ -1,3 +1,4 @@ +import { workflowTracingOptions } from "./workflow-tracing-worker.js"; import { fileURLToPath } from "node:url"; import { NativeConnection, Worker } from "@temporalio/worker"; import { executePersistedMemoryChangeSet } from "./memory-mutation-activity.js"; @@ -23,7 +24,8 @@ if (!memoryMutationsEnabled(process.env)) { workflowsPath: fileURLToPath(new URL("./memory-mutation-workflows.js", import.meta.url)), activities: { executePersistedMemoryChangeSet }, ...memoryMutationWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + interceptors: { activity: [createActivityObservabilityInterceptors({ spans })], workflowModules: workflowTracingOptions(spans).workflowModules }, + sinks: workflowTracingOptions(spans).sinks }); 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/room-background-continuation.ts b/apps/worker/src/room-background-continuation.ts index e15c29a21..d2fef8c9e 100644 --- a/apps/worker/src/room-background-continuation.ts +++ b/apps/worker/src/room-background-continuation.ts @@ -1,4 +1,7 @@ +import { randomBytes } from "node:crypto"; import { + formatTraceparent, + parseTraceparent, buildCodexAppServerTurnWorkflowId, buildRoomAgentSupervisorWorkflowId, roomAgentSupervisorQueueItemSchema, @@ -130,7 +133,11 @@ 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: `trace:bg-${input.continuationId}`, + ...(parseTraceparent(source.queueItem.turn.traceparent) ? { + traceLink: source.queueItem.turn.traceparent, + traceparent: formatTraceparent(randomBytes(16).toString("hex"), randomBytes(8).toString("hex"), "01") + } : {}) } }); try { const enqueued = await store.enqueueRoomAgentMission({ diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 896269754..38dbd386c 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -1,3 +1,4 @@ +import { workflowTracingOptions } from "./workflow-tracing-worker.js"; import { fileURLToPath } from "node:url"; import { hostname } from "node:os"; import { assertDevInstanceIsolated, instanceBackgroundJobEnabled } from "@space/contracts"; @@ -43,9 +44,9 @@ const worker = await Worker.create({ 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 })] } + // Export activity attempts and replay-suppressed workflow spans. + interceptors: { activity: [createActivityObservabilityInterceptors({ spans })], workflowModules: workflowTracingOptions(spans).workflowModules }, + sinks: workflowTracingOptions(spans).sinks }); // Optional loopback-only Prometheus endpoint for worker metrics. diff --git a/apps/worker/src/workflow-trace-context.ts b/apps/worker/src/workflow-trace-context.ts new file mode 100644 index 000000000..b769c1195 --- /dev/null +++ b/apps/worker/src/workflow-trace-context.ts @@ -0,0 +1,68 @@ +import { parseTraceparent } from "@space/contracts/observability"; +import type { Headers, Sinks } from "@temporalio/workflow"; + +export const traceHeader = "space-traceparent"; +export const linkHeader = "space-trace-link"; + +// ASCII only: these helpers also run in Temporal's deterministic sandbox. +export function tracePayload(value: string) { + return { metadata: { encoding: Uint8Array.from([98, 105, 110, 97, 114, 121, 47, 112, 108, 97, 105, 110]) }, + data: Uint8Array.from(value, char => char.charCodeAt(0)) }; +} + +export function traceFromHeaders(headers: Headers, key = traceHeader): string | undefined { + const data = headers[key]?.data; + if (!data || data.length !== 55) return undefined; + const value = String.fromCharCode(...data); + return parseTraceparent(value) ? value : undefined; +} + +export function traceFromArgs(args: readonly unknown[], key = "traceparent"): string | undefined { + const visit = (value: unknown, depth: number): string | undefined => { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const record = value as Record; + if (parseTraceparent(record[key])) return record[key] as string; + if (depth > 0) for (const nested of ["turn", "item", "bridge", "input", "turnInput"]) { + const found = visit(record[nested], depth - 1); + if (found) return found; + } + return undefined; + }; + for (const arg of args) { + const found = visit(arg, 3); + if (found) return found; + } + return undefined; +} + +const identifier = /^[A-Za-z0-9:._-]{1,200}$/; +const attributeKeys = new Set([ + "space.workflow_id", "space.workflow_run_id", "space.workflow_type", "space.activity_id", + "space.activity_type", "space.attempt", "space.outcome", "space.reason_code", "space.category", "space.duration_ms" +]); + +/** No arbitrary attributes, free-form errors, arguments, paths or URLs. */ +export function safeTraceAttributes(attributes: Readonly>): Record { + const result: Record = {}; + for (const [key, value] of Object.entries(attributes)) { + if (!attributeKeys.has(key)) continue; + if (typeof value === "string" && identifier.test(value)) result[key] = value; + else if (typeof value === "number" && Number.isFinite(value) && value >= 0) result[key] = value; + } + return result; +} + +export interface WorkflowTraceSpan { + traceId: string; + spanId: string; + parentSpanId: string | null; + startedAt: number; + endedAt: number; + attributes: Record; + error: boolean; + links: { traceId: string; spanId: string }[]; +} + +export interface WorkflowTraceSinks extends Sinks { + spaceTracing: { record(span: WorkflowTraceSpan): void }; +} diff --git a/apps/worker/src/workflow-tracing-worker.ts b/apps/worker/src/workflow-tracing-worker.ts new file mode 100644 index 000000000..61e6b8287 --- /dev/null +++ b/apps/worker/src/workflow-tracing-worker.ts @@ -0,0 +1,26 @@ +import { fileURLToPath } from "node:url"; +import type { InjectedSinks } from "@temporalio/worker"; +import type { SpanExporter } from "@space/runtime"; +import { safeTraceAttributes, type WorkflowTraceSinks } from "./workflow-trace-context.js"; + +export function workflowTracingOptions(spans: SpanExporter | null) { + const sinks: InjectedSinks = { + spaceTracing: { record: { + callDuringReplay: false, + fn(_info, span) { + try { + const attributes = safeTraceAttributes(span.attributes); + spans?.record({ traceId: span.traceId, spanId: span.spanId, parentSpanId: span.parentSpanId, + name: `workflow ${attributes["space.workflow_type"] ?? "unknown"}`, kind: 1, + startTimeUnixNano: BigInt(span.startedAt) * 1_000_000n, + endTimeUnixNano: BigInt(span.endedAt) * 1_000_000n, + attributes, error: span.error, links: span.links }); + } catch { /* Export must never fail workflow processing. */ } + } + } } + }; + return { + workflowModules: spans ? [fileURLToPath(new URL("./workflow-tracing.js", import.meta.url))] : [], + sinks: spans ? sinks : {} + }; +} diff --git a/apps/worker/src/workflow-tracing.ts b/apps/worker/src/workflow-tracing.ts new file mode 100644 index 000000000..4f8fcc354 --- /dev/null +++ b/apps/worker/src/workflow-tracing.ts @@ -0,0 +1,63 @@ +import { ContinueAsNew, proxySinks, workflowInfo, type WorkflowInterceptors, type Headers } from "@temporalio/workflow"; +import { formatTraceparent, parseTraceparent } from "@space/contracts/observability"; +import { linkHeader, safeTraceAttributes, traceFromArgs, traceFromHeaders, traceHeader, tracePayload, type WorkflowTraceSinks } from "./workflow-trace-context.js"; + +/** No I/O or random-number consumption inside workflow code. */ +export function interceptors(): WorkflowInterceptors { + const sink = proxySinks().spaceTracing; + let current: string | undefined; + const headers = (original: Headers): Headers => current ? { ...original, [traceHeader]: tracePayload(current) } : original; + return { + inbound: [{ + async execute(input, next) { + const info = workflowInfo(); + const incoming = traceFromHeaders(input.headers) ?? traceFromArgs(input.args) ?? traceFromArgs([info.memo]); + const parent = parseTraceparent(incoming); + // Temporal run ids are UUIDs. Reusing their entropy avoids changing the + // workflow's seeded random sequence (and hence existing command ids). + const runHex = info.runId.replaceAll("-", ""); + const spanId = runHex.slice(16); + const traceId = parent?.traceId ?? runHex; + current = formatTraceparent(traceId, spanId, parent?.flags ?? "01"); + const argumentLink = traceFromArgs(input.args, "traceLink"); + const selectedLink = traceFromHeaders(input.headers, linkHeader) ?? argumentLink; + const link = parseTraceparent(selectedLink); + const backgroundRoot = argumentLink === selectedLink && link && link.traceId !== parent?.traceId + && parseTraceparent(traceFromArgs(input.args))?.parentId === parent?.parentId; + const startedAt = Date.now(); + let outcome = "COMPLETED"; + try { return await next(input); } + catch (error) { + outcome = error instanceof ContinueAsNew ? "CONTINUED_AS_NEW" : "FAILED"; + throw error; + } finally { + const endedAt = Date.now(); + sink.record({ traceId, spanId, parentSpanId: backgroundRoot ? null : parent?.parentId ?? null, startedAt, endedAt, + attributes: safeTraceAttributes({ "space.workflow_id": info.workflowId, "space.workflow_run_id": info.runId, + "space.workflow_type": info.workflowType, "space.attempt": info.attempt, "space.outcome": outcome, + "space.duration_ms": Math.max(0, endedAt - startedAt) }), error: outcome === "FAILED", + links: link ? [{ traceId: link.traceId, spanId: link.parentId }] : [] }); + } + } + }], + outbound: [{ + scheduleActivity: (input, next) => next({ ...input, headers: headers(input.headers) }), + scheduleLocalActivity: (input, next) => next({ ...input, headers: headers(input.headers) }), + startChildWorkflowExecution(input, next) { + const explicit = traceFromArgs(input.options.args ?? []); + const explicitLink = traceFromArgs(input.options.args ?? [], "traceLink"); + // A supervisor may launch a new mission received on a later request. + // Preserve that mission's trace; ordinary children nest under this run. + const differentTrace = explicit && parseTraceparent(explicit)?.traceId !== parseTraceparent(current)?.traceId; + let childHeaders = headers(input.headers); + if (differentTrace) childHeaders = { ...childHeaders, [traceHeader]: tracePayload(explicit) }; + const link = explicitLink ?? (differentTrace ? current : undefined); + if (link) childHeaders = { ...childHeaders, [linkHeader]: tracePayload(link) }; + return next({ ...input, headers: childHeaders }); + }, + continueAsNew(input, next) { + return next({ ...input, headers: { ...headers(input.headers), ...(current ? { [linkHeader]: tracePayload(current) } : {}) } }); + } + }] + }; +} diff --git a/apps/worker/tests/activity-observability.test.ts b/apps/worker/tests/activity-observability.test.ts index 654bf7962..86819a091 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).toContain("workflowModules: workflowTracingOptions(spans).workflowModules"); + expect(source, entry).toContain("sinks: workflowTracingOptions(spans).sinks"); + expect(source, entry).toMatch(/interceptors: \{ activity: \[createActivityObservabilityInterceptors\(\{ spans \}\)\]/); } }); @@ -203,3 +205,12 @@ 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("collector failures preserve activity results and original errors", async () => { + const spans = { record: () => { throw new Error("collector failure"); }, flush: async () => {}, + close: async () => {}, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; + await expect(observeActivityExecution(info(), [], async () => "ok", { spans, traceparent })).resolves.toBe("ok"); + const error = new Error("original activity error"); + await expect(observeActivityExecution(info(), [], async () => { throw error; }, { spans, traceparent })).rejects.toBe(error); +}); diff --git a/apps/worker/tests/fixtures/tracing-workflows.ts b/apps/worker/tests/fixtures/tracing-workflows.ts new file mode 100644 index 000000000..2610b2632 --- /dev/null +++ b/apps/worker/tests/fixtures/tracing-workflows.ts @@ -0,0 +1,21 @@ +import { continueAsNew, executeChild, proxyActivities } from "@temporalio/workflow"; + +const activities = proxyActivities<{ attempt(): Promise }>({ + startToCloseTimeout: "10 seconds", retry: { initialInterval: "1 second", maximumAttempts: 2 } +}); + +export async function tracingChild(): Promise { + return activities.attempt(); +} + +export async function tracingRoot(input: { traceparent: string; prompt: string }, generation = 0): Promise { + await activities.attempt(); + await executeChild(tracingChild); + if (generation === 0) return continueAsNew(input, 1); + return "complete"; +} + +export async function tracingBackground(input: { traceparent: string; traceLink: string }): Promise { + await activities.attempt(); + return "background complete"; +} diff --git a/apps/worker/tests/room-background-continuation.test.ts b/apps/worker/tests/room-background-continuation.test.ts index 127ccb649..ad356f3ea 100644 --- a/apps/worker/tests/room-background-continuation.test.ts +++ b/apps/worker/tests/room-background-continuation.test.ts @@ -184,3 +184,15 @@ describe("durable background continuation", () => { expect(c.store.getRoomAgentMission(c.room.id, c.initial.mission.id)?.status).toBe("INTERRUPTED"); }); }); + + +it("persists a fresh background trace and its source link across retry", async () => { + const parent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + const c = continuationFixture("OPEN_TYPES", parent); + const prepared = await prepareBackgroundContinuation(c.input, c.store); + if (prepared.state !== "READY") throw Error("Missing continuation"); + expect(prepared.item.turn.traceLink).toBe(parent); + expect(prepared.item.turn.traceparent).toMatch(/^00-[a-f0-9]{32}-[a-f0-9]{16}-01$/); + expect(prepared.item.turn.traceparent?.slice(3, 35)).not.toBe(parent.slice(3, 35)); + expect(await prepareBackgroundContinuation(c.input, c.store)).toEqual(prepared); +}); diff --git a/apps/worker/tests/room-continuation-fixture.ts b/apps/worker/tests/room-continuation-fixture.ts index 340e964e0..7963fbcb7 100644 --- a/apps/worker/tests/room-continuation-fixture.ts +++ b/apps/worker/tests/room-continuation-fixture.ts @@ -1,7 +1,7 @@ import { buildCodexAppServerTurnWorkflowId, roomAgentSupervisorQueueItemSchema } from "@space/contracts"; import { InMemorySpaceStore } from "@space/runtime"; -export function continuationFixture(actionType: "OPEN_TYPES" | "CONFIGURE_PANE" = "OPEN_TYPES") { +export function continuationFixture(actionType: "OPEN_TYPES" | "CONFIGURE_PANE" = "OPEN_TYPES", traceparent?: string) { const store = new InMemorySpaceStore(); const room = store.createRoom({ name: "Continuation regression", initialPaneCount: 0 }); const pane = store.getOrCreateRoomAgentPane(room.id); @@ -14,7 +14,7 @@ export function continuationFixture(actionType: "OPEN_TYPES" | "CONFIGURE_PANE" modelId: "gpt-5.6-sol", reasoningEffort: "high", agentSessionId: session.sessionId, agentRunId: `agent_run:${suffix}`, agentUserMessageId: `agent_msg:user-${suffix}`, agentAssistantMessageId: `agent_msg:assistant-${suffix}`, roomAgentMissionId: "mission:continuation", - traceId: `trace:continuation-${suffix}`, selectedToolIds: ["room:inspect", "room:open_types", "room:send"] + traceparent, traceId: `trace:continuation-${suffix}`, selectedToolIds: ["room:inspect", "room:open_types", "room:send"] } }); return store.enqueueRoomAgentMission({ requestId: `request:${suffix}`, clientRequestId: `client-${suffix}`, content: prompt, supervisorWorkflowId: `room-supervisor:${room.id}`, childWorkflowId: buildCodexAppServerTurnWorkflowId(item.turn), diff --git a/apps/worker/tests/workflow-tracing.test.ts b/apps/worker/tests/workflow-tracing.test.ts new file mode 100644 index 000000000..02bf9ad1e --- /dev/null +++ b/apps/worker/tests/workflow-tracing.test.ts @@ -0,0 +1,156 @@ +import { createServer } from "node:http"; +import Fastify from "fastify"; +import { fileURLToPath } from "node:url"; +import { Context } from "@temporalio/activity"; +import { Worker, bundleWorkflowCode } from "@temporalio/worker"; +import { TestWorkflowEnvironment } from "@temporalio/testing"; +import { describe, expect, it } from "vitest"; +import { createOtlpSpanExporter, type ServerSpan } from "../../../packages/runtime/src/otlp-span-exporter.js"; +import { createRequestTraceContext, traceContextForRawRequest, TraceContextRegistry, withTurnTraceparent } from "../../api/src/trace-context.js"; +import { createHttpObservability } from "../../api/src/observability.js"; +import { createActivityObservabilityInterceptors } from "../src/activity-observability.js"; +import { workflowTracingOptions } from "../src/workflow-tracing-worker.js"; +import { safeTraceAttributes, traceFromHeaders, tracePayload, traceHeader } from "../src/workflow-trace-context.js"; + +const path = (relative: string) => fileURLToPath(new URL(relative, import.meta.url)); +type CollectedSpan = { traceId: string; spanId: string; parentSpanId?: string; name: string; + attributes: { key: string; value: { stringValue?: string; intValue?: string } }[]; + links?: { traceId: string; spanId: string }[]; status: { code: number } }; +const attribute = (span: CollectedSpan, key: string) => { + const value = span.attributes.find(a => a.key === key)?.value; + return value?.stringValue ?? value?.intValue; +}; + +describe("workflow tracing", () => { + it("is disabled without an exporter and rejects non-allowlisted data", () => { + expect(workflowTracingOptions(null)).toEqual({ workflowModules: [], sinks: {} }); + expect(safeTraceAttributes({ prompt: "secret", "space.workflow_type": "/private/path", + "space.reason_code": "ERROR_CODE", "space.attempt": 2, "space.duration_ms": Infinity })) + .toEqual({ "space.reason_code": "ERROR_CODE", "space.attempt": 2 }); + expect(traceFromHeaders({ [traceHeader]: tracePayload("invalid") })).toBeUndefined(); + }); + + it("exports real Temporal runs, retries, child parents and continuation links, then replays without exporting", async () => { + const received: CollectedSpan[] = []; + const collector = createServer(async (req, res) => { + const parts: Buffer[] = []; + for await (const part of req) parts.push(Buffer.from(part)); + expect(req.url).toBe("/v1/traces"); + const body = JSON.parse(Buffer.concat(parts).toString()); + received.push(...body.resourceSpans.flatMap((r: { scopeSpans: { spans: CollectedSpan[] }[] }) => r.scopeSpans.flatMap(s => s.spans))); + res.writeHead(200).end("{}"); + }); + await new Promise(resolve => collector.listen(0, "127.0.0.1", resolve)); + const address = collector.address() as { port: number }; + const endpoint = `http://127.0.0.1:${address.port}`; + const exporter = createOtlpSpanExporter({ endpoint, serviceName: "space-worker", flushIntervalMs: 60_000 }); + const apiExporter = createOtlpSpanExporter({ endpoint, serviceName: "space-api", flushIntervalMs: 60_000 }); + const app = Fastify(); + const httpObservability = createHttpObservability({ serviceName: "space-api", spanExporter: apiExporter }); + app.addHook("onRequest", async request => httpObservability.onRequest(request)); + app.addHook("onResponse", async (request, reply) => httpObservability.onResponse(request, reply)); + let env: TestWorkflowEnvironment | undefined; + try { + await exporter.flush(); // An idle flush must not permanently block later batches. + // The default is self-contained. A prestarted isolated test server is + // useful on hosts that exceed the SDK's fixed five-second startup limit. + const testAddress = process.env.SPACE_TRACE_TEST_TEMPORAL_ADDRESS; + env = testAddress + ? await TestWorkflowEnvironment.createFromExistingServer({ address: testAddress }) + : await TestWorkflowEnvironment.createTimeSkipping(); + const workflowBundle = await bundleWorkflowCode({ workflowsPath: path("./fixtures/tracing-workflows.ts"), + workflowInterceptorModules: [path("../src/workflow-tracing.ts")] }); + const options = workflowTracingOptions(exporter); + const worker = await Worker.create({ connection: env.nativeConnection, taskQueue: "trace-proof", workflowBundle, + sinks: options.sinks, interceptors: { activity: [createActivityObservabilityInterceptors({ spans: exporter })] }, + activities: { attempt: async () => { + if (Context.current().info.attempt === 1) throw new Error("PRIVATE ERROR /home/private/token SECRET"); + return "PRIVATE OUTPUT"; + } } }); + let api = createRequestTraceContext(undefined); + app.post("/test/turn", async request => { + api = traceContextForRawRequest(request.raw); + const registry = new TraceContextRegistry(); + registry.remember(request.id, api); + const turn = withTurnTraceparent({ traceId: request.id, prompt: "PRIVATE PROMPT https://secret.invalid/token" }, registry); + return env!.client.workflow.execute("tracingRoot", { workflowId: "trace-root", taskQueue: "trace-proof", args: [turn] }); + }); + await worker.runUntil(async () => { + const response = await app.inject({ method: "POST", url: "/test/turn" }); + expect(response.statusCode).toBe(200); + expect(response.body).toBe("complete"); + const background = createRequestTraceContext(undefined); + expect(await env!.client.workflow.execute("tracingBackground", { workflowId: "trace-background", taskQueue: "trace-proof", + args: [{ traceparent: background.traceparent, traceLink: api.traceparent }] })).toBe("background complete"); + }); + await exporter.flush(); + await apiExporter.flush(); + const workflows = received.filter(s => s.name.startsWith("workflow ")); + const roots = workflows.filter(s => s.name === "workflow tracingRoot"); + expect(roots).toHaveLength(2); + const first = roots.find(s => attribute(s, "space.outcome") === "CONTINUED_AS_NEW")!; + const second = roots.find(s => attribute(s, "space.outcome") === "COMPLETED")!; + expect(first).toMatchObject({ traceId: api.w3cTraceId, parentSpanId: api.spanId }); + expect(second.links).toContainEqual({ traceId: first.traceId, spanId: first.spanId }); + const children = workflows.filter(s => s.name === "workflow tracingChild"); + expect(children).toHaveLength(2); + for (const child of children) expect(roots.some(root => root.spanId === child.parentSpanId && root.traceId === child.traceId)).toBe(true); + const background = workflows.find(s => s.name === "workflow tracingBackground")!; + expect(background.traceId).not.toBe(api.w3cTraceId); + expect(background.parentSpanId).toBeUndefined(); + expect(background.links).toContainEqual({ traceId: api.w3cTraceId, spanId: api.spanId }); + const activities = received.filter(s => s.name === "activity attempt"); + expect(activities).toHaveLength(10); + expect(new Set(activities.map(s => s.spanId)).size).toBe(10); + for (const activity of activities) { + expect(workflows.some(w => w.spanId === activity.parentSpanId && w.traceId === activity.traceId)).toBe(true); + const attempt = attribute(activity, "space.attempt"); + expect(attribute(activity, "space.outcome")).toBe(attempt === "1" ? "FAILED" : "COMPLETED"); + if (attempt === "1") expect(attribute(activity, "space.reason_code")).toBe("ACTIVITY_ERROR"); + } + expect(JSON.stringify(received)).not.toMatch(/PRIVATE|secret\.invalid|home\/private|SECRET/); + let replayExports = 0; + const replayExporter = { ...exporter, record: (_span: ServerSpan) => { replayExports++; } }; + for (const workflow of workflows) { + const history = await env.client.workflow.getHandle(attribute(workflow, "space.workflow_id")!, attribute(workflow, "space.workflow_run_id")!).fetchHistory(); + await Worker.runReplayHistory({ workflowBundle, sinks: workflowTracingOptions(replayExporter).sinks }, history); + } + expect(replayExports).toBe(0); + + // Histories recorded before tracing was enabled must still replay. The + // fixture also starts children with SDK-generated ids, guarding the PRNG. + const plainBundle = await bundleWorkflowCode({ workflowsPath: path("./fixtures/tracing-workflows.ts") }); + const plainWorker = await Worker.create({ connection: env.nativeConnection, taskQueue: "trace-disabled", workflowBundle: plainBundle, + activities: { attempt: async () => "ok" } }); + const legacyHistory = await plainWorker.runUntil(async () => { + const handle = await env!.client.workflow.start("tracingRoot", { workflowId: "trace-legacy", taskQueue: "trace-disabled", + args: [{ traceparent: api.traceparent, prompt: "PRIVATE" }] }); + expect(await handle.result()).toBe("complete"); + return env!.client.workflow.getHandle("trace-legacy", handle.firstExecutionRunId).fetchHistory(); + }); + await Worker.runReplayHistory({ workflowBundle, sinks: workflowTracingOptions(replayExporter).sinks }, legacyHistory); + expect(replayExports).toBe(0); + + const unavailable = createOtlpSpanExporter({ endpoint: "http://127.0.0.1:1", serviceName: "outage-test", timeoutMs: 100 }); + try { + const outageWorker = await Worker.create({ connection: env.nativeConnection, taskQueue: "trace-outage", workflowBundle, + sinks: workflowTracingOptions(unavailable).sinks, + interceptors: { activity: [createActivityObservabilityInterceptors({ spans: unavailable })] }, + activities: { attempt: async () => "ok" } }); + await outageWorker.runUntil(async () => { + expect(await env!.client.workflow.execute("tracingBackground", { workflowId: "trace-outage", taskQueue: "trace-outage", + args: [{ traceparent: api.traceparent, traceLink: api.traceparent }] })).toBe("background complete"); + }); + await unavailable.flush(); + expect(unavailable.stats().failed).toBe(2); + } finally { await unavailable.close(); } + } finally { + await app.close(); + httpObservability.close(); + await env?.teardown(); + await exporter.close(); + await apiExporter.close(); + await new Promise((resolve, reject) => collector.close(error => error ? reject(error) : resolve())); + } + }, 180_000); +}); diff --git a/apps/worker/vitest.config.ts b/apps/worker/vitest.config.ts index c21b63776..68e591558 100644 --- a/apps/worker/vitest.config.ts +++ b/apps/worker/vitest.config.ts @@ -8,6 +8,7 @@ const __dirname = path.dirname(fileURLToPath(import.meta.url)); export default defineConfig({ resolve: { alias: { + "@space/contracts/observability": resolve(__dirname, "../../packages/contracts/src/observability.ts"), "@space/db": resolve(__dirname, "../../packages/db/src/index.ts"), "@space/contracts": resolve(__dirname, "../../packages/contracts/src/index.ts"), "@space/opencode-control": resolve(__dirname, "../../packages/opencode-control/src/index.ts"), diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index e432af4c6..2aee41b91 100644 --- a/docs/configuration-reference.md +++ b/docs/configuration-reference.md @@ -52,3 +52,13 @@ background work, and is hidden in the web app. Public installs upgraded from 1.1.x that used Job Copilot, the games, channels, Telegram or the desktop routes set the modules they keep, for example `SPACE_OPTIONAL_MODULES=job-copilot,games`. + +## Optional trace export + +`SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is an optional HTTP(S) collector base URL +(for example `http://127.0.0.1:4318`). The API and all three Temporal workers +append `/v1/traces` and send OTLP JSON. Unset or invalid values disable export; +URLs containing credentials, query strings or fragments are rejected. +No tracing SDK or provider credentials are needed. The worker workflow sink +does not run during replay. See [Temporal OTLP traces](quality/observability.md#temporal-otlp-traces) +for propagation, continuation links, queue limits, failure handling and tests. diff --git a/docs/quality/observability.md b/docs/quality/observability.md index 488fc3036..80dbd794f 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -293,6 +293,60 @@ No credential or transcript export was added. | `apps/worker/tests/activity-observability.test.ts` | Interceptor records, outcomes, categories, metrics, loopback endpoint | | `apps/api/tests/observability-fault-injection.test.ts` | Four injected executor faults and one Temporal start failure, end to end | +## Temporal OTLP traces + +Set `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` to a collector base URL, for example +`http://127.0.0.1:4318`. The shared exporter posts OTLP/HTTP JSON to `/v1/traces`. +The API, main worker, memory-maintenance worker and memory-mutation worker use +the same exporter. With no endpoint, workers install no tracing workflow module +or sink and make no collector requests. + +API SERVER spans parent workflow execution spans. Workflow headers parent each +activity attempt, including retries and local activities, and child workflows. +Activity spans include outcome, attempt, stable reason code/category, identifiers +and duration. Trace attributes are allowlisted; prompts, outputs, paths, URLs and +free-form exception messages are excluded. Failed-result activities and thrown +failures are distinct outcomes. Export never controls activity success. + +Workflow instrumentation uses deterministic run identifiers and Temporal's +workflow clock; it does not consume the workflow PRNG or perform network I/O. +The worker sink is configured with `callDuringReplay: false`. Completed workflow +spans are emitted when execution finishes, fails or continues as new; a running +long-lived supervisor will not yet have a completed span. Workflow-task retries +can deliver a sink call more than once; stable run span IDs let collectors +recognize these duplicates. This is not an exactly-once export guarantee. + +`continueAsNew` carries both the current context and an OTLP span link to the +previous run. Durable background continuations retain their Space-native +`trace:bg-*` identifier, create a fresh W3C trace, and persist `traceLink` to the +source request context. The new workflow exports that link and its activities +parent the new workflow. A supervisor launching a mission from a different API +trace preserves the mission's parent and links back to the supervisor. + +The in-memory queue holds at most 4096 spans, sends batches of up to 256, and +flushes every 5 seconds with a 3-second POST timeout. `stats()` exposes exported, +dropped and failed counts. Failed batches are counted and discarded, with no +retry feedback into workflows. A collector outage does not block task execution. + +Reproducible checks (Node 22, after `npm ci` and the dependency builds in +`npm run check`): + +```sh +npm test -w @space/worker -- tests/workflow-tracing.test.ts +npm test -w @space/runtime -- tests/otlp-span-exporter.test.ts +npm test -w @space/api -- otlp +``` + +The integration test starts a local Temporal test server and local HTTP +collector, runs stub activities with one failed attempt per execution, checks +child and continuation relationships, and replays each recorded history while +asserting zero sink exports. The initial Temporal test-server binary download +requires internet; no paid provider, credentials or Docker service is required. +On a slow host that exceeds the SDK's fixed five-second server startup limit, +start an isolated Temporal test server beforehand and set +`SPACE_TRACE_TEST_TEMPORAL_ADDRESS=127.0.0.1:` for this test. Never point +this test-only override at an operator or production Temporal server. + ## Not covered - **Supervisor failures lose their cause.** When `roomAgentSupervisorWorkflow` @@ -336,8 +390,8 @@ No credential or transcript export was added. the lookup also returns the Space-native agent runs (`space_agent_runs`) of the trace's workflows, joined by workflow id, with their status and error code, and names a failed run's code as the cause when no event carries one. -- **Hosts emit no spans.** Since 2026-10-16 the OTLP exporter sends API SERVER - spans and worker activity spans (the worker and both memory workers); the CLI +- **Hosts emit no spans.** The OTLP exporter sends API SERVER spans and worker + workflow/activity spans (the worker and both memory workers); the CLI and Browser Hosts still join traces through log lines only. - **Recovery counters are partial.** Recovery counters exist for plan model retries, CLI terminal reconnects and worker activity retries only. diff --git a/package-lock.json b/package-lock.json index 1e2aa2a88..5fe71e89d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -192,6 +192,7 @@ "undici": "8.10.2" }, "devDependencies": { + "@temporalio/testing": "1.24.0", "typescript": "^5.9.3", "vitest": "^4.1.12" } @@ -3674,6 +3675,25 @@ "node": ">=12.0.0" } }, + "node_modules/@temporalio/testing": { + "version": "1.24.0", + "resolved": "https://registry.npmjs.org/@temporalio/testing/-/testing-1.24.0.tgz", + "integrity": "sha512-XKCuvgb7smKR3oXy5nDgwYSFX/hmg9zTJ2rZQQsVwJzFkga1TbnWDiuT5YMU0INKjoLhnM6c5YA/uLKlp8sdfg==", + "dev": true, + "license": "MIT", + "dependencies": { + "@temporalio/activity": "1.24.0", + "@temporalio/client": "1.24.0", + "@temporalio/common": "1.24.0", + "@temporalio/core-bridge": "1.24.0", + "@temporalio/proto": "1.24.0", + "@temporalio/worker": "1.24.0", + "@temporalio/workflow": "1.24.0" + }, + "engines": { + "node": ">= 20.3.0" + } + }, "node_modules/@temporalio/worker": { "version": "1.24.0", "resolved": "https://registry.npmjs.org/@temporalio/worker/-/worker-1.24.0.tgz", @@ -11641,12 +11661,12 @@ } }, "packages/run-spaceapp": { - "version": "1.0.35", + "version": "1.1.7", "cpu": [ "x64", "arm64" ], - "license": "Apache-2.0", + "license": "MIT", "os": [ "darwin", "linux", diff --git a/packages/contracts/package.json b/packages/contracts/package.json index 44ac04459..9e2806bb7 100644 --- a/packages/contracts/package.json +++ b/packages/contracts/package.json @@ -13,6 +13,10 @@ "./tool-routing": { "types": "./dist/tool-routing.d.ts", "import": "./dist/tool-routing.js" + }, + "./observability": { + "types": "./dist/observability.d.ts", + "import": "./dist/observability.js" } }, "scripts": { diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index 696eb6d84..b818b910e 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -6807,6 +6807,7 @@ export const dummyTurnInputSchema = z.object({ // Optional W3C trace context of the originating request (additive; see // docs/quality/observability.md). Never part of a workflow id. traceparent: traceparentSchema.optional(), + traceLink: traceparentSchema.optional(), // Space Dots Stage 3 (additive): a gateway-enabled native Chat turn. dotId: dotIdSchema.optional(), toolGateway: z.literal("space-dot-tools-v1").optional(), diff --git a/packages/runtime/src/otlp-span-exporter.ts b/packages/runtime/src/otlp-span-exporter.ts index 86a0edf05..8e5873b3b 100644 --- a/packages/runtime/src/otlp-span-exporter.ts +++ b/packages/runtime/src/otlp-span-exporter.ts @@ -1,11 +1,12 @@ /** - * Optional OTLP/HTTP (JSON) export of API server spans and worker activity spans. Off unless + * Optional OTLP/HTTP (JSON) export of API, workflow and activity spans. Off unless * SPACE_OTEL_EXPORTER_OTLP_ENDPOINT names a collector base URL (spans go to * `/v1/traces`). No dependency, no collector required: Space logs stay * joinable by W3C trace id without it. * - * Spans carry only the method, route template, status code, service name and - * Space request id; never URLs with parameters, headers, bodies or user data. + * Callers allowlist span attributes: route metadata for API requests, and + * identifiers, outcomes and durations for worker executions. Never include + * URLs with parameters, headers, bodies or user data. * Export is best effort: a bounded queue drops (and counts) spans instead of * growing, each POST has a timeout, and a failing collector never affects a * request. @@ -22,6 +23,7 @@ export interface ServerSpan { error: boolean; /** OTLP span kind; SERVER (2) unless given. Worker activities use INTERNAL (1). */ kind?: number; + links?: readonly { traceId: string; spanId: string }[]; } export interface SpanExporterStats { @@ -81,6 +83,7 @@ export function otlpTraceRequest(spans: readonly ServerSpan[], serviceName: stri spanId: span.spanId, ...(span.parentSpanId ? { parentSpanId: span.parentSpanId } : {}), name: span.name, + ...(span.links?.length ? { links: span.links.filter(link => traceIdPattern.test(link.traceId) && spanIdPattern.test(link.spanId)) } : {}), kind: span.kind ?? 2, startTimeUnixNano: span.startTimeUnixNano.toString(), endTimeUnixNano: span.endTimeUnixNano.toString(), @@ -122,13 +125,13 @@ export function createOtlpSpanExporter(options: OtlpSpanExporterOptions): SpanEx }; const flush = (): Promise => { if (!inFlight) { - inFlight = (async () => { + inFlight = Promise.resolve().then(async () => { try { while (queue.length) await flushOnce(); } finally { inFlight = null; } - })(); + }); } return inFlight; }; diff --git a/packages/runtime/tests/otlp-span-exporter.test.ts b/packages/runtime/tests/otlp-span-exporter.test.ts new file mode 100644 index 000000000..eb99e3c38 --- /dev/null +++ b/packages/runtime/tests/otlp-span-exporter.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from "vitest"; +import { createOtlpSpanExporter, otlpTraceRequest, type ServerSpan } from "../src/otlp-span-exporter.js"; + +const span: ServerSpan = { + traceId: "a".repeat(32), spanId: "b".repeat(16), parentSpanId: null, name: "workflow test", + startTimeUnixNano: 1n, endTimeUnixNano: 2n, attributes: {}, error: false, kind: 1 +}; + +describe("shared OTLP exporter", () => { + it("continues exporting after idle and concurrent flushes", async () => { + const bodies: string[] = []; + const exporter = createOtlpSpanExporter({ endpoint: "http://localhost:4318", serviceName: "test", + fetch: async (_url, init) => { bodies.push(init.body); return { ok: true }; } }); + try { + await Promise.all([exporter.flush(), exporter.flush()]); + exporter.record(span); + await exporter.flush(); + await exporter.flush(); + exporter.record(span); + await exporter.flush(); + expect(bodies).toHaveLength(2); + expect(exporter.stats()).toEqual({ exported: 2, dropped: 0, failed: 0 }); + } finally { await exporter.close(); } + }); + + it("serializes valid OTLP links and rejects malformed ids", () => { + const link = { traceId: "c".repeat(32), spanId: "d".repeat(16) }; + const body = otlpTraceRequest([{ ...span, links: [link, { traceId: "secret", spanId: "bad" }] }], "test"); + expect(body.resourceSpans[0]!.scopeSpans[0]!.spans[0]!.links).toEqual([link]); + }); + + it("bounds the queue and counts collector failures without rejecting", async () => { + const exporter = createOtlpSpanExporter({ endpoint: "http://localhost:4318", serviceName: "test", maxQueue: 2, + fetch: async () => { throw new Error("collector unavailable"); } }); + try { + for (let i = 0; i < 5; i++) exporter.record(span); + await expect(exporter.flush()).resolves.toBeUndefined(); + expect(exporter.stats()).toEqual({ exported: 0, dropped: 3, failed: 2 }); + } finally { await exporter.close(); } + }); +}); From 9760eef1001786410f7aa2fb8ee82e1e210d9d16 Mon Sep 17 00:00:00 2001 From: Yusef Atteyih Date: Sat, 24 Oct 2026 22:02:18 +0300 Subject: [PATCH 2/4] docs: align replay guidance with workflow tracing Signed-off-by: Yusef Atteyih --- docs/quality/observability.md | 47 +++++++++++++++++++---------------- 1 file changed, 25 insertions(+), 22 deletions(-) diff --git a/docs/quality/observability.md b/docs/quality/observability.md index 80dbd794f..7d2f95e8d 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -219,9 +219,12 @@ The trace context is local by default. An optional OTLP/HTTP (JSON) exporter sen 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. +`space-worker`, `space-memory-maintenance-worker`, `space-memory-mutation-worker`). +When the exporter is configured, workflow interceptors propagate context through +Temporal headers to workflow executions and activity attempts. Activity spans +include allowlisted identifiers, activity type, outcome, attempt, duration and +stable error code/category. Activity input context remains a fallback. See +[Temporal OTLP traces](#temporal-otlp-traces) for span relationships and limits. - **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`). @@ -233,28 +236,28 @@ that span, with the activity type, outcome and attempt only. - **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. +- **Workflow instrumentation is conditional.** With an exporter configured, + the main worker and both memory workers bundle `workflow-tracing.ts` as a + workflow interceptor module and install its worker-side sink. Without an + exporter, they install neither. API input/memo propagation and the activity + interceptor also participate in correlation. ### Replay compatibility -No file under `apps/worker/src` that the workflow bundle imports was changed -(`workflows.ts`, `room-supervisor-state.ts`, `turn-runtime-policy.ts`, -`dot-goal-loop.ts`). The `agent-turns` workflow bundle was built for the base -commit and for this change and compared after normalizing paths and webpack -module numbers. The only differences are in `@space/contracts`: -- the new `observability` module; -- the optional `traceparent` turn-input field; -- the optional file search and readiness fields. - -A history without `traceparent` parses exactly as before. Activity inputs are -not part of Temporal's determinism check. - -The read-only live replay (`node scripts/temporal-replay-check.mjs --limit 30`) -could not be run from the agent sandbox that made this change. Its network -namespace has no route to the Temporal container (`172.18.0.5:7233`), and -`127.0.0.1:7233` refuses connections. Run it before deploying this change. +The tracing-enabled workflow bundle includes the new interceptor and its +trace-context helpers. Instrumentation uses Temporal's workflow clock and +existing run identifiers, without network I/O or consumption of the workflow +PRNG. The worker sink has `callDuringReplay: false` so replay does not export +duplicate workflow spans. + +`apps/worker/tests/workflow-tracing.test.ts` covers replay of instrumented and +previously uninstrumented fixture histories, as well as disabled tracing. Those +fixtures do not establish compatibility with every production history. + +Live production histories have not been replayed for this change. Before +deploying, run the read-only replay check from an authorized environment with +access to the target Temporal service: +`node scripts/temporal-replay-check.mjs --limit 30`. ## Redaction policy From 90a90c7f7b3297106ce22ffe6c57a3c1440d37e8 Mon Sep 17 00:00:00 2001 From: Yusef Atteyih Date: Sat, 24 Oct 2026 23:04:48 +0300 Subject: [PATCH 3/4] fix(worker): flush maintenance spans when run fails Signed-off-by: Yusef Atteyih --- apps/worker/src/memory-maintenance-worker.ts | 9 ++- .../tests/memory-maintenance-worker.test.ts | 69 +++++++++++++++++++ 2 files changed, 75 insertions(+), 3 deletions(-) create mode 100644 apps/worker/tests/memory-maintenance-worker.test.ts diff --git a/apps/worker/src/memory-maintenance-worker.ts b/apps/worker/src/memory-maintenance-worker.ts index d1996594f..6d4ad9b39 100644 --- a/apps/worker/src/memory-maintenance-worker.ts +++ b/apps/worker/src/memory-maintenance-worker.ts @@ -43,9 +43,12 @@ if (!memoryMaintenanceEnabled(process.env)) { activityConcurrency: memoryMaintenanceWorkerOptions.maxConcurrentActivityTaskExecutions })); const stopReloadWatch = reloadWhenServicesCommitChanges(worker, { log: (message) => console.log(JSON.stringify({ service, status: "reloading", message })) }); - await worker.run(); - stopReloadWatch(); - await spans?.close(); + try { + await worker.run(); + } finally { + stopReloadWatch(); + await spans?.close(); + } } catch (error) { console.error(JSON.stringify({ service, diff --git a/apps/worker/tests/memory-maintenance-worker.test.ts b/apps/worker/tests/memory-maintenance-worker.test.ts new file mode 100644 index 000000000..aa8070148 --- /dev/null +++ b/apps/worker/tests/memory-maintenance-worker.test.ts @@ -0,0 +1,69 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ + run: vi.fn(), + close: vi.fn(), + stopReloadWatch: vi.fn(), + exporter: vi.fn() +})); + +vi.mock("@temporalio/worker", () => ({ + NativeConnection: { connect: vi.fn().mockResolvedValue({}) }, + Worker: { create: vi.fn(async () => ({ run: mocks.run })) } +})); +vi.mock("@space/runtime", () => ({ otlpSpanExporterFromEnv: mocks.exporter })); +vi.mock("../src/memory-maintenance-activities.js", () => ({})); +vi.mock("../src/activity-observability.js", () => ({ createActivityObservabilityInterceptors: vi.fn() })); +vi.mock("../src/workflow-tracing-worker.js", () => ({ workflowTracingOptions: () => ({}) })); +vi.mock("../src/deployed-code-reload.js", () => ({ + reloadWhenServicesCommitChanges: () => mocks.stopReloadWatch +})); + +describe("maintenance worker shutdown", () => { + let originalExitCode: typeof process.exitCode; + + beforeEach(() => { + vi.resetModules(); + vi.resetAllMocks(); + originalExitCode = process.exitCode; + process.exitCode = undefined; + vi.stubEnv("SPACE_MEMORY_GRAPH_ENABLED", "true"); + vi.stubEnv("SPACE_MEMORY_MAINTENANCE_ENABLED", "true"); + vi.spyOn(console, "log").mockImplementation(() => {}); + vi.spyOn(console, "error").mockImplementation(() => {}); + mocks.close.mockResolvedValue(undefined); + mocks.exporter.mockReturnValue({ close: mocks.close }); + }); + + afterEach(() => { + process.exitCode = originalExitCode; + vi.unstubAllEnvs(); + vi.restoreAllMocks(); + }); + + it.each([false, true])("cleans up when worker.run rejects: %s", async (fails) => { + if (fails) mocks.run.mockRejectedValue(new Error("worker failed")); + else mocks.run.mockResolvedValue(undefined); + + await import("../src/memory-maintenance-worker.js"); + + expect(mocks.run).toHaveBeenCalledOnce(); + expect(mocks.stopReloadWatch).toHaveBeenCalledOnce(); + expect(mocks.close).toHaveBeenCalledOnce(); + expect(process.exitCode).toBe(fails ? 1 : undefined); + if (fails) { + expect(console.error).toHaveBeenCalledWith(expect.stringContaining('"error":"worker failed"')); + } + }); + + it("preserves failure handling when tracing is disabled", async () => { + mocks.exporter.mockReturnValue(undefined); + mocks.run.mockRejectedValue(new Error("worker failed")); + + await import("../src/memory-maintenance-worker.js"); + + expect(mocks.stopReloadWatch).toHaveBeenCalledOnce(); + expect(mocks.close).not.toHaveBeenCalled(); + expect(process.exitCode).toBe(1); + }); +}); From 5979fb3abdcc7e404edc41e8118bd3c6c8a25916 Mon Sep 17 00:00:00 2001 From: Yusef Atteyih Date: Sun, 25 Oct 2026 00:00:23 +0300 Subject: [PATCH 4/4] fix(test): resolve observability subpath in API suites Signed-off-by: Yusef Atteyih --- apps/api/vitest.config.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/api/vitest.config.ts b/apps/api/vitest.config.ts index 1917bbed9..9ad30aa7c 100644 --- a/apps/api/vitest.config.ts +++ b/apps/api/vitest.config.ts @@ -8,6 +8,7 @@ const __dirname = path.dirname(fileURLToPath(import.meta.url)); export default defineConfig({ resolve: { alias: { + "@space/contracts/observability": resolve(__dirname, "../../packages/contracts/src/observability.ts"), "@space/cli-host": resolve(__dirname, "../../packages/cli-host/src/index.ts"), "@space/db": resolve(__dirname, "../../packages/db/src/index.ts"), "@space/gateway": resolve(__dirname, "../../packages/gateway/src/index.ts"),