diff --git a/apps/api/src/dot-goals.ts b/apps/api/src/dot-goals.ts index 30dd69dee..face965f0 100644 --- a/apps/api/src/dot-goals.ts +++ b/apps/api/src/dot-goals.ts @@ -16,6 +16,7 @@ import { } from "@space/contracts"; import { SpaceConflictError, SpaceNotFoundError, type SpaceStore } from "@space/runtime"; import type { DotGoalsRepository, DotsRepository } from "@space/db"; +import { traceContextForRawRequest, traceparentForTraceId, workflowCorrelationMemo } from "./trace-context.js"; /** * Space Dots Stage 4: Goals. @@ -26,7 +27,7 @@ import type { DotGoalsRepository, DotsRepository } from "@space/db"; * blocking question, cancellation, three consecutive failures or the budget. */ export interface DotGoalWorkflowCoordinator { - start(goalId: string): Promise; + start(goalId: string, trace?: { traceId?: string | null; traceparent?: string | null }): Promise; wake(goalId: string): Promise; /** Starts the Goal's workflow when none is running; true when it had to start one. */ ensure?(goalId: string): Promise; @@ -38,11 +39,18 @@ export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordi try { return await action(new Client({ connection, namespace: this.options.namespace })); } finally { await connection.close(); } } - start(goalId: string) { + start(goalId: string, trace?: { traceId?: string | null; traceparent?: string | null }) { + const traceparent = trace?.traceparent ?? traceparentForTraceId(trace?.traceId); + const memo = workflowCorrelationMemo({ traceId: trace?.traceId, traceparent }); return this.client(async (client) => { try { - await client.workflow.start("dotGoalWorkflow", { workflowId: dotGoalWorkflowId(goalId), taskQueue: this.options.taskQueue, - args: [{ goalId }], workflowIdReusePolicy: "REJECT_DUPLICATE" }); + await client.workflow.start("dotGoalWorkflow", { + workflowId: dotGoalWorkflowId(goalId), + taskQueue: this.options.taskQueue, + args: [{ goalId, ...(trace?.traceId ? { traceId: trace.traceId } : {}), ...(traceparent ? { traceparent } : {}) }], + workflowIdReusePolicy: "REJECT_DUPLICATE", + ...(Object.keys(memo).length ? { memo } : {}) + }); } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; } }); } @@ -125,7 +133,15 @@ export function createDotGoalService(deps: DotGoalServiceDeps) { if ((error as { code?: string }).code === "DOT_GOAL_OPEN") throw new DotGoalError(409, "DOT_GOAL_OPEN", "This conversation already has an open Goal."); throw error; } - try { await deps.workflow.start(goal.goalId); } + try { + if (request) { + const traceId = request.id ? String(request.id) : null; + const traceparent = request.raw ? traceContextForRawRequest(request.raw).traceparent : traceparentForTraceId(traceId); + await deps.workflow.start(goal.goalId, { traceId, traceparent }); + } else { + await deps.workflow.start(goal.goalId); + } + } catch { goal = await finish(goal, "FAILED", "The Goal runtime could not start. Try again."); throw new DotGoalError(503, "DOT_GOAL_RUNTIME", "The Goal runtime could not start. Try again."); diff --git a/apps/api/src/memory-consolidation-coordinator.ts b/apps/api/src/memory-consolidation-coordinator.ts index df237c648..a939691c3 100644 --- a/apps/api/src/memory-consolidation-coordinator.ts +++ b/apps/api/src/memory-consolidation-coordinator.ts @@ -6,6 +6,7 @@ import { type MemoryConsolidationWorkflowInput } from "@space/contracts"; import { SpaceFeatureDisabledError } from "@space/runtime"; +import { withTurnTraceparent, workflowCorrelationMemo } from "./trace-context.js"; export { MEMORY_CONSOLIDATION_TASK_QUEUE, MEMORY_CONSOLIDATION_WORKFLOW_TYPE }; @@ -21,6 +22,7 @@ interface WorkflowStartOptions { workflowId: string; workflowIdReusePolicy: "REJECT_DUPLICATE"; workflowIdConflictPolicy: "USE_EXISTING"; + memo?: Record; } interface WorkflowStartHandle { @@ -42,14 +44,16 @@ export async function scheduleMemoryConsolidation( workflowId: string, startWorkflow: MemoryConsolidationWorkflowStarter ): Promise { - const input = memoryConsolidationWorkflowInputSchema.parse(rawInput); + const input = withTurnTraceparent(memoryConsolidationWorkflowInputSchema.parse(rawInput)); + const memo = input.traceparent ? workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) : undefined; try { const handle = await startWorkflow(MEMORY_CONSOLIDATION_WORKFLOW_TYPE, { args: [input], taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE, workflowId, workflowIdReusePolicy: "REJECT_DUPLICATE", - workflowIdConflictPolicy: "USE_EXISTING" + workflowIdConflictPolicy: "USE_EXISTING", + ...(memo && Object.keys(memo).length ? { memo } : {}) }); return { status: "SCHEDULED", diff --git a/apps/api/src/memory-mutation-coordinator.ts b/apps/api/src/memory-mutation-coordinator.ts index f7d3d48bb..dc2a370fc 100644 --- a/apps/api/src/memory-mutation-coordinator.ts +++ b/apps/api/src/memory-mutation-coordinator.ts @@ -7,6 +7,7 @@ import { type MemoryMutationWorkflowInput } from "@space/contracts"; import { SpaceFeatureDisabledError } from "@space/runtime"; +import { withTurnTraceparent, workflowCorrelationMemo } from "./trace-context.js"; export { MEMORY_MUTATION_TASK_QUEUE, MEMORY_MUTATION_WORKFLOW_TYPE, buildMemoryMutationWorkflowId }; @@ -22,6 +23,7 @@ interface WorkflowStartOptions { workflowId: string; workflowIdReusePolicy: "REJECT_DUPLICATE"; workflowIdConflictPolicy: "USE_EXISTING"; + memo?: Record; } interface WorkflowStartHandle { @@ -43,15 +45,17 @@ export async function scheduleMemoryMutation( startWorkflow: MemoryMutationWorkflowStarter, commandHash?: string ): Promise { - const input = memoryMutationWorkflowInputSchema.parse(rawInput); + const input = withTurnTraceparent(memoryMutationWorkflowInputSchema.parse(rawInput)); const workflowId = buildMemoryMutationWorkflowId(input.changeSetId, commandHash); + const memo = input.traceparent ? workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) : undefined; try { const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, { args: [input], taskQueue: MEMORY_MUTATION_TASK_QUEUE, workflowId, workflowIdReusePolicy: "REJECT_DUPLICATE", - workflowIdConflictPolicy: "USE_EXISTING" + workflowIdConflictPolicy: "USE_EXISTING", + ...(memo && Object.keys(memo).length ? { memo } : {}) }); return { status: "SCHEDULED", workflowId: handle.workflowId, runId: handle.firstExecutionRunId ?? null }; } catch (error) { diff --git a/apps/api/src/room-agent.ts b/apps/api/src/room-agent.ts index 5eccd5097..f4f945117 100644 --- a/apps/api/src/room-agent.ts +++ b/apps/api/src/room-agent.ts @@ -188,16 +188,17 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo async enqueueAction(actionId: string, bridge: SpaceAgentRoomActionBridgeRequest, traceId: string): Promise { const lease = await this.connect("roomAgentActionWorkflow"); + const traceparent = traceparentForTraceId(traceId) ?? traceparentForTraceId(bridge.requestId ?? null); try { try { await lease.client.workflow.start("roomAgentActionWorkflow", { workflowId: `space-room-action:${actionId}`, taskQueue: this.options.taskQueue, - args: [{ actionId, bridge: { ...bridge, backgroundExecution: true }, traceId }], + args: [{ actionId, bridge: { ...bridge, backgroundExecution: true }, traceId, ...(traceparent ? { traceparent } : {}) }], workflowIdReusePolicy: "REJECT_DUPLICATE", // traceId is this internal request; bridge.requestId is the originating request. memo: workflowCorrelationMemo({ traceId, - traceparent: traceparentForTraceId(traceId), + traceparent, parentTraceId: bridge.requestId ?? null, missionId: bridge.missionId }) @@ -219,6 +220,11 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo // the child turn workflow and its activities (additive optional field). const item = { ...parsed, turn: withTurnTraceparent(parsed.turn) }; const workflowId = buildRoomAgentSupervisorWorkflowId(item.turn.roomId); + const memo = workflowCorrelationMemo({ + traceId: item.turn.traceId, + traceparent: item.turn.traceparent, + missionId: item.missionId + }); const lease = await this.connect(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE); try { const handle = await lease.client.workflow.signalWithStart(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE, { @@ -226,7 +232,8 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo taskQueue: this.options.taskQueue, args: [{ roomId: item.turn.roomId, pending: [], processedCount: 0 }], signal: ROOM_AGENT_ENQUEUE_SIGNAL, - signalArgs: [item] + signalArgs: [item], + ...(Object.keys(memo).length ? { memo } : {}) }); recordWorkflowStart(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE, "SIGNALED"); return { workflowId, runId: handle.signaledRunId }; diff --git a/apps/api/src/trace-context.ts b/apps/api/src/trace-context.ts index af9e55efd..f45617fc6 100644 --- a/apps/api/src/trace-context.ts +++ b/apps/api/src/trace-context.ts @@ -96,7 +96,7 @@ export const requestTraceContexts = new TraceContextRegistry(); /** traceparent for work started on behalf of this Space request id, if known. */ export function traceparentForTraceId(traceId: string | null | undefined, registry = requestTraceContexts): string | null { - return childTraceparent(registry.lookup(traceId)?.traceparent ?? null); + return registry.lookup(traceId)?.traceparent ?? null; } /** Adds the request's traceparent to a turn input unless it already has a valid one. */ diff --git a/apps/worker/package.json b/apps/worker/package.json index ec58125fa..db05b0164 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -26,6 +26,7 @@ "undici": "8.12.3" }, "devDependencies": { + "@temporalio/testing": "1.24.0", "typescript": "^5.9.3", "vitest": "^4.1.12" }, diff --git a/apps/worker/src/activity-observability.ts b/apps/worker/src/activity-observability.ts index ea4d021d1..409fdd6bf 100644 --- a/apps/worker/src/activity-observability.ts +++ b/apps/worker/src/activity-observability.ts @@ -1,7 +1,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; import { randomBytes } from "node:crypto"; import type { Context } from "@temporalio/activity"; -import type { ActivityInterceptorsFactory } from "@temporalio/worker"; +import { defaultPayloadConverter, type ActivityInterceptorsFactory, type Headers } from "@temporalio/worker"; import { categorizeFailureCode, formatTraceparent, @@ -10,9 +10,11 @@ import { traceIdFromTraceparent, type FailureCategory } from "@space/contracts"; -import type { SpanExporter } from "@space/runtime"; +import { sanitizeSpanAttributes, type SpanExporter } from "@space/runtime"; import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js"; import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js"; +import { lookupActiveWorkflowTrace } from "./workflow-observability.js"; +import { WORKFLOW_TRACE_ID_HEADER, WORKFLOW_TRACEPARENT_HEADER } from "./workflow-observability-interceptors.js"; /** * One structured start/finish record and metrics per activity execution, with @@ -135,8 +137,21 @@ export interface ObserveActivityDependencies { metrics?: WorkerMetrics; now?: () => number; cancelled?: () => boolean; - /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */ + /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input or workflow header carries a traceparent. */ spans?: SpanExporter | null; + headers?: Headers; +} + +function readActivityHeader(headers: Headers | undefined, key: string): string | null { + if (!headers) return null; + const payload = headers[key]; + if (!payload) return null; + try { + const decoded = defaultPayloadConverter.fromPayload(payload); + return typeof decoded === "string" ? decoded : null; + } catch { + return null; + } } function activityTraceparent(args: readonly unknown[]) { @@ -194,8 +209,49 @@ export async function observeActivityExecution( queueMs, rssBytes: process.memoryUsage().rss }); - const parent = dependencies.spans ? activityTraceparent(args) : null; + const headerTp = parseTraceparent(readActivityHeader(dependencies.headers, WORKFLOW_TRACEPARENT_HEADER)); + const activeWf = lookupActiveWorkflowTrace( + info.workflowExecution?.workflowId ?? null, + info.workflowExecution?.runId ?? null + ); + const activeWfTp = parseTraceparent(activeWf?.traceparent); + const parent = dependencies.spans ? (headerTp ?? activeWfTp ?? activityTraceparent(args)) : null; if (parent) { + const corr = activityCorrelation(args); + const rawAttributes: Record = { + ...(activeWf?.correlation ?? {}), + "space.activity_type": activityType, + "space.outcome": outcome, + "space.attempt": info.attempt, + "space.duration_ms": durationMs + }; + if (info.activityId) rawAttributes["space.activity_id"] = String(info.activityId); + if (info.workflowType) rawAttributes["space.workflow_type"] = info.workflowType; + if (info.workflowExecution?.workflowId) rawAttributes["space.workflow_id"] = info.workflowExecution.workflowId; + if (info.workflowExecution?.runId) rawAttributes["space.run_id"] = info.workflowExecution.runId; + if (queueMs !== null) rawAttributes["space.queue_ms"] = queueMs; + if (reasonCode) rawAttributes["space.reason_code"] = reasonCode; + if (category) rawAttributes["space.category"] = category; + const headerTraceId = readActivityHeader(dependencies.headers, WORKFLOW_TRACE_ID_HEADER); + if (headerTraceId) rawAttributes["space.trace_id"] = headerTraceId; + const corrAttrMap: ReadonlyArray = [ + ["traceId", "space.trace_id"], + ["parentTraceId", "space.parent_trace_id"], + ["missionId", "space.mission_id"], + ["roomId", "space.room_id"], + ["paneId", "space.pane_id"], + ["agentRunId", "space.agent_run_id"], + ["agentSessionId", "space.agent_session_id"], + ["actionId", "space.action_id"], + ["goalId", "space.goal_id"], + ["dotId", "space.dot_id"], + ["turnId", "space.turn_id"], + ["cliTaskId", "space.cli_task_id"], + ["revisionId", "space.revision_id"] + ]; + for (const [corrKey, attrKey] of corrAttrMap) { + if (corr[corrKey]) rawAttributes[attrKey] = corr[corrKey]!; + } dependencies.spans!.record({ traceId: parent.traceId, spanId: randomBytes(8).toString("hex"), @@ -203,7 +259,7 @@ export async function observeActivityExecution( 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: sanitizeSpanAttributes(rawAttributes), error: outcome === "FAILED" || outcome === "FAILED_RESULT", kind: 1 }); @@ -224,11 +280,14 @@ export async function observeActivityExecution( } /** Temporal worker interceptor factory: register in Worker.create({ interceptors }). */ -export function createActivityObservabilityInterceptors(dependencies: Omit = {}): ActivityInterceptorsFactory { +export function createActivityObservabilityInterceptors( + dependencies: Omit = {} +): ActivityInterceptorsFactory { return (context) => ({ inbound: { execute: (input, next) => 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..5535be4a9 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 { createWorkflowObservabilitySinks, workflowObservabilityInterceptorPath } from "./workflow-observability.js"; import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js"; const service = "space-memory-maintenance-worker"; @@ -30,7 +31,11 @@ if (!memoryMaintenanceEnabled(process.env)) { workflowsPath, activities, ...memoryMaintenanceWorkerOptions, - interceptors: { activity: [createActivityObservabilityInterceptors({ spans })] } + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + workflowModules: [workflowObservabilityInterceptorPath] + }, + sinks: createWorkflowObservabilitySinks({ spans }) }); 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..5afb333a3 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 { createWorkflowObservabilitySinks, workflowObservabilityInterceptorPath } from "./workflow-observability.js"; import { reloadWhenServicesCommitChanges } from "./deployed-code-reload.js"; const service = "space-memory-mutation-worker"; @@ -23,7 +24,11 @@ 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: [workflowObservabilityInterceptorPath] + }, + sinks: createWorkflowObservabilitySinks({ spans }) }); console.log(JSON.stringify({ service, status: "running", address, namespace, taskQueue: MEMORY_MUTATION_TASK_QUEUE })); const stopReloadWatch = reloadWhenServicesCommitChanges(worker, { log: (message) => console.log(JSON.stringify({ service, status: "reloading", message })) }); diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 83d23cbc6..c7f689fe7 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 { createWorkflowObservabilitySinks, workflowObservabilityInterceptorPath } from "./workflow-observability.js"; import { resolveTemporalCoreMetricsBind, resolveWorkerMetricsListen, startWorkerMetricsServer } from "./worker-metrics.js"; import { workerLog } from "./worker-log.js"; import { sweepCodexCliCompletions } from "./codex-cli-completion.js"; @@ -43,9 +44,11 @@ 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 })] } + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + workflowModules: [workflowObservabilityInterceptorPath] + }, + sinks: createWorkflowObservabilitySinks({ spans }) }); // Optional loopback-only Prometheus endpoint for worker metrics. diff --git a/apps/worker/src/workflow-observability-interceptors.ts b/apps/worker/src/workflow-observability-interceptors.ts new file mode 100644 index 000000000..e7cd83cd6 --- /dev/null +++ b/apps/worker/src/workflow-observability-interceptors.ts @@ -0,0 +1,457 @@ +import { + defaultPayloadConverter, + isCancellation, + proxySinks, + workflowInfo, + type ActivityInput, + type ContinueAsNewInput, + type Headers, + type LocalActivityInput, + type Next, + type SignalInput, + type Sinks, + type StartChildWorkflowExecutionInput, + type WorkflowExecuteInput, + type WorkflowInboundCallsInterceptor, + type WorkflowInterceptorsFactory, + type WorkflowOutboundCallsInterceptor +} from "@temporalio/workflow"; +import { + categorizeFailureCode, + formatTraceparent, + isSafeErrorCode, + parseTraceparent +} from "@space/contracts"; + +export const WORKFLOW_TRACEPARENT_HEADER = "space-traceparent"; +export const WORKFLOW_CONTINUED_TRACEPARENT_HEADER = "space-continued-from-traceparent"; +export const WORKFLOW_TRACE_ID_HEADER = "space-trace-id"; + +export interface WorkflowSpanLinkRecord { + traceId: string; + spanId: string; + attributes?: Record; +} + +export interface WorkflowSpanRecordPayload { + traceId: string; + spanId: string; + parentSpanId: string | null; + workflowType: string; + workflowId: string; + runId: string; + parentWorkflowId: string | null; + parentRunId: string | null; + continuedFromRunId: string | null; + outcome: "COMPLETED" | "FAILED" | "CANCELLED" | "CONTINUED_AS_NEW"; + reasonCode: string | null; + category: string | null; + correlation: Record; + links: WorkflowSpanLinkRecord[]; +} + +export interface WorkflowObservabilitySinks extends Sinks { + workflowObservability: { + workflowStarted(spanId: string, traceparent: string, correlation: Record): void; + workflowFinished(payload: WorkflowSpanRecordPayload): void; + }; +} + +const { workflowObservability } = proxySinks(); + +const safeIdPattern = /^[A-Za-z0-9:._-]{1,240}$/; + +function safeId(value: unknown): string | null { + return typeof value === "string" && safeIdPattern.test(value) ? value : null; +} + +/** + * Pure deterministic 64-bit FNV-1a / SplitMix64 hex generator for workflow isolate. + * Produces 16 hex chars (spanId) or 32 hex chars (traceId) without Node crypto. + */ +function deterministicHexBlock(seed: string, salt: number): string { + let hash = 0xcbf29ce484222325n ^ BigInt(salt & 0xffffffff); + const prime = 0x100000001b3n; + for (let index = 0; index < seed.length; index += 1) { + hash ^= BigInt(seed.charCodeAt(index)); + hash = (hash * prime) & 0xffffffffffffffffn; + } + // SplitMix64 finalizer + hash = ((hash ^ (hash >> 30n)) * 0xbf58476d1ce4e5b9n) & 0xffffffffffffffffn; + hash = ((hash ^ (hash >> 27n)) * 0x94d049bb133111ebn) & 0xffffffffffffffffn; + hash = (hash ^ (hash >> 31n)) & 0xffffffffffffffffn; + if (hash === 0n) hash = 1n; + return hash.toString(16).padStart(16, "0"); +} + +export function workflowSpanId(workflowId: string, runId: string): string { + return deterministicHexBlock(`wf-span:${workflowId}:${runId}`, 0x53504143); +} + +export function deterministicTraceId(seed: string): string { + return `${deterministicHexBlock(`wf-trace-hi:${seed}`, 0x10203040)}${deterministicHexBlock(`wf-trace-lo:${seed}`, 0x50607080)}`; +} + +export function continuationSpanId(sourceRunId: string): string { + return deterministicHexBlock(`bg-continuation:${sourceRunId}`, 0x4247434e); +} + +function readHeaderString(headers: Headers | undefined, key: string): string | null { + if (!headers) return null; + const payload = headers[key]; + if (!payload) return null; + try { + const decoded = defaultPayloadConverter.fromPayload(payload); + return typeof decoded === "string" ? decoded : null; + } catch { + return null; + } +} + +const correlationKeys = [ + "traceId", + "traceparent", + "requestId", + "roomId", + "paneId", + "missionId", + "roomAgentMissionId", + "agentRunId", + "agentSessionId", + "actionId", + "goalId", + "dotId", + "turnId", + "cliTaskId", + "revisionId", + "changeSetId", + "runId" +] as const; +const nestedKeys = ["turn", "item", "bridge", "input", "turnInput"] as const; + +type CorrelationKey = (typeof correlationKeys)[number]; + +interface ExtractedWorkflowInput { + fields: Partial>; + sourceRunId: string | null; +} + +function scanObject(value: unknown, out: ExtractedWorkflowInput, depth: number): void { + if (!value || typeof value !== "object") return; + if (Array.isArray(value)) { + if (depth > 0) { + for (const item of value.slice(0, 4)) scanObject(item, out, depth - 1); + } + return; + } + const record = value as Record; + for (const key of correlationKeys) { + if (out.fields[key] !== undefined) continue; + const id = safeId(record[key]); + if (id !== null) out.fields[key] = id; + } + const continuation = record.roomAgentContinuation; + if (!out.sourceRunId && continuation && typeof continuation === "object" && !Array.isArray(continuation)) { + const sourceId = safeId((continuation as Record).sourceRunId); + if (sourceId) out.sourceRunId = sourceId; + } + if (depth <= 0) return; + for (const key of nestedKeys) { + scanObject(record[key], out, depth - 1); + } + if (Array.isArray(record.pending)) { + scanObject(record.pending[0], out, depth - 1); + } +} + +export function extractWorkflowInputMetadata( + args: readonly unknown[], + memo?: Record +): ExtractedWorkflowInput { + const out: ExtractedWorkflowInput = { fields: {}, sourceRunId: null }; + for (const arg of args) scanObject(arg, out, 3); + if (memo && typeof memo === "object") { + if (!out.fields.traceparent) { + const memoTp = safeId(memo.traceparent); + if (memoTp) out.fields.traceparent = memoTp; + } + if (!out.fields.traceId) { + const memoTraceId = safeId(memo.spaceTraceId); + if (memoTraceId) out.fields.traceId = memoTraceId; + } + if (!out.fields.requestId) { + const parentSpaceTraceId = safeId(memo.parentSpaceTraceId); + if (parentSpaceTraceId) out.fields.requestId = parentSpaceTraceId; + } + if (!out.fields.missionId) { + const memoMissionId = safeId(memo.missionId); + if (memoMissionId) out.fields.missionId = memoMissionId; + } + } + return out; +} + +function buildCorrelationAttributes(fields: Partial>): Record { + const attrs: Record = {}; + if (fields.traceId) attrs["space.trace_id"] = fields.traceId; + if (fields.requestId && fields.requestId !== fields.traceId) attrs["space.parent_trace_id"] = fields.requestId; + const missionId = fields.missionId ?? fields.roomAgentMissionId; + if (missionId) attrs["space.mission_id"] = missionId; + const mapping: ReadonlyArray = [ + ["roomId", "space.room_id"], + ["paneId", "space.pane_id"], + ["agentRunId", "space.agent_run_id"], + ["agentSessionId", "space.agent_session_id"], + ["actionId", "space.action_id"], + ["goalId", "space.goal_id"], + ["dotId", "space.dot_id"], + ["turnId", "space.turn_id"], + ["cliTaskId", "space.cli_task_id"], + ["revisionId", "space.revision_id"], + ["changeSetId", "space.change_set_id"] + ]; + for (const [field, attrKey] of mapping) { + if (fields[field]) attrs[attrKey] = fields[field]!; + } + return attrs; +} + +function workflowErrorCode(error: unknown): string { + let cursor = error as { name?: unknown; type?: unknown; code?: unknown; cause?: unknown } | undefined; + for (let depth = 0; cursor && depth < 8; depth += 1, cursor = cursor.cause as typeof cursor) { + if (isSafeErrorCode(cursor.code)) return cursor.code; + if (isSafeErrorCode(cursor.type)) return cursor.type; + if (cursor.name === "TimeoutFailure") return "WORKFLOW_TIMEOUT"; + } + return "WORKFLOW_ERROR"; +} + +function isSupervisorMissionTerminalActivity(activityType: string): boolean { + return ( + activityType === "finishRoomAgentSupervisorTurn" || + activityType === "markRoomAgentMissionFinished" || + activityType === "markRoomAgentMissionContinued" || + activityType === "discardRoomAgentSupervisorTurn" + ); +} + +export const interceptors: WorkflowInterceptorsFactory = () => { + const info = workflowInfo(); + let spanId = workflowSpanId(info.workflowId, info.runId); + let traceId = deterministicTraceId(`${info.workflowId}:${info.firstExecutionRunId || info.runId}`); + let parentSpanId: string | null = info.parent ? workflowSpanId(info.parent.workflowId, info.parent.runId) : null; + let flags = "01"; + let hasExplicitTrace = false; + let correlation: Record = {}; + const links: WorkflowSpanLinkRecord[] = []; + let spanEmitted = false; + + const currentTraceparent = () => formatTraceparent(traceId, spanId, flags); + + const applyTraceInput = (args: readonly unknown[], headers?: Headers) => { + const headerTp = parseTraceparent(readHeaderString(headers, WORKFLOW_TRACEPARENT_HEADER)); + const continuedTp = parseTraceparent(readHeaderString(headers, WORKFLOW_CONTINUED_TRACEPARENT_HEADER)); + const headerTraceId = safeId(readHeaderString(headers, WORKFLOW_TRACE_ID_HEADER)); + const extracted = extractWorkflowInputMetadata(args, info.memo); + const argTp = parseTraceparent(extracted.fields.traceparent); + const traceIdField = extracted.fields.traceId ?? ""; + // Background continuations keep a new W3C trace (`trace:bg-*`) and point + // at the caller's span with a link. A parent-workflow header still nests + // real child workflows under the parent span. + const isBackgroundContinuation = Boolean(extracted.sourceRunId) || traceIdField.startsWith("trace:bg-"); + + const chosenTp = headerTp ?? (isBackgroundContinuation ? null : argTp); + if (chosenTp) { + traceId = chosenTp.traceId; + parentSpanId = chosenTp.parentId; + flags = chosenTp.flags; + hasExplicitTrace = true; + } else if (info.parent) { + parentSpanId = workflowSpanId(info.parent.workflowId, info.parent.runId); + } + + const nextCorrelation = buildCorrelationAttributes(extracted.fields); + if (!nextCorrelation["space.trace_id"] && headerTraceId) { + nextCorrelation["space.trace_id"] = headerTraceId; + } + correlation = { ...correlation, ...nextCorrelation }; + + // 1. Continue-As-New link + if (info.continuedFromExecutionRunId || continuedTp) { + const prevTraceId = continuedTp?.traceId ?? traceId; + const prevSpanId = + continuedTp?.parentId ?? + (info.continuedFromExecutionRunId ? workflowSpanId(info.workflowId, info.continuedFromExecutionRunId) : null); + if (prevSpanId && !links.some(l => l.spanId === prevSpanId && l.attributes?.["space.link_kind"] === "CONTINUE_AS_NEW")) { + links.push({ + traceId: prevTraceId, + spanId: prevSpanId, + attributes: { + "space.link_kind": "CONTINUE_AS_NEW", + ...(info.continuedFromExecutionRunId ? { "space.continued_from_run_id": info.continuedFromExecutionRunId } : {}) + } + }); + } + } + + // 2. Background continuation link (`trace:bg-...` or `roomAgentContinuation.sourceRunId`). + // Prefer the caller's real span (traceparent) over a synthetic id so the link + // resolves in an OTLP backend. + if (isBackgroundContinuation) { + const sourceRunId = extracted.sourceRunId ?? traceIdField; + const linkedTraceId = argTp?.traceId ?? continuedTp?.traceId ?? traceId; + const linkedSpanId = argTp?.parentId ?? continuedTp?.parentId ?? continuationSpanId(sourceRunId); + if (!links.some(l => l.spanId === linkedSpanId && l.attributes?.["space.link_kind"] === "BACKGROUND_CONTINUATION")) { + links.push({ + traceId: linkedTraceId, + spanId: linkedSpanId, + attributes: { + "space.link_kind": "BACKGROUND_CONTINUATION", + ...(sourceRunId ? { "space.source_run_id": sourceRunId } : {}) + } + }); + } + } + }; + + const emitStart = () => { + if (!info.unsafe.isReplaying) { + workflowObservability.workflowStarted(spanId, currentTraceparent(), correlation); + } + }; + + const emitFinish = ( + outcome: WorkflowSpanRecordPayload["outcome"], + reasonCode: string | null = null, + category: string | null = null + ) => { + if (spanEmitted) return; + spanEmitted = true; + if (!info.unsafe.isReplaying) { + workflowObservability.workflowFinished({ + traceId, + spanId, + parentSpanId, + workflowType: info.workflowType, + workflowId: info.workflowId, + runId: info.runId, + parentWorkflowId: info.parent?.workflowId ?? null, + parentRunId: info.parent?.runId ?? null, + continuedFromRunId: info.continuedFromExecutionRunId ?? null, + outcome, + reasonCode, + category, + correlation, + links + }); + } + }; + + const inbound: WorkflowInboundCallsInterceptor = { + async execute(input: WorkflowExecuteInput, next: Next) { + applyTraceInput(input.args, input.headers); + emitStart(); + try { + const result = await next(input); + const failedObj = + result && typeof result === "object" && (result as { status?: unknown }).status === "FAILED" + ? (result as { reasonCode?: unknown; errorCode?: unknown }) + : null; + if (failedObj) { + const code = isSafeErrorCode(failedObj.reasonCode) + ? failedObj.reasonCode + : isSafeErrorCode(failedObj.errorCode) + ? failedObj.errorCode + : "WORKFLOW_RESULT_FAILED"; + emitFinish("FAILED", code, categorizeFailureCode(code)); + } else { + emitFinish("COMPLETED"); + } + return result; + } catch (error) { + // ContinueAsNew is handled in the outbound continueAsNew interceptor before throwing. + if ((error as { name?: string } | null)?.name === "ContinueAsNew") { + emitFinish("CONTINUED_AS_NEW"); + throw error; + } + if (isCancellation(error)) { + emitFinish("CANCELLED", "WORKFLOW_CANCELLED", "CANCELLED"); + } else { + const code = workflowErrorCode(error); + emitFinish("FAILED", code, categorizeFailureCode(code)); + } + throw error; + } + }, + + async handleSignal(input: SignalInput, next: Next) { + const beforeExplicit = hasExplicitTrace; + applyTraceInput(input.args, input.headers); + if (!beforeExplicit && hasExplicitTrace) { + // A supervisor signaled with a mission turn starts a new mission span cycle if a previous one already emitted. + spanEmitted = false; + emitStart(); + } else if (spanEmitted) { + // Subsequent mission signaled into an existing supervisor workflow run. + spanId = workflowSpanId(info.workflowId, `${info.runId}:${ correlation["space.mission_id"] ?? "next" }`); + spanEmitted = false; + emitStart(); + } + return next(input); + } + }; + + const outbound: WorkflowOutboundCallsInterceptor = { + async scheduleActivity(input: ActivityInput, next: Next) { + input.headers[WORKFLOW_TRACEPARENT_HEADER] = defaultPayloadConverter.toPayload(currentTraceparent()); + const spaceTraceId = correlation["space.trace_id"]; + if (spaceTraceId) { + input.headers[WORKFLOW_TRACE_ID_HEADER] = defaultPayloadConverter.toPayload(spaceTraceId); + } + const result = await next(input); + if (info.workflowType === "roomAgentSupervisorWorkflow" && isSupervisorMissionTerminalActivity(input.activityType)) { + emitFinish("COMPLETED"); + } + return result; + }, + + async scheduleLocalActivity( + input: LocalActivityInput, + next: Next + ) { + input.headers[WORKFLOW_TRACEPARENT_HEADER] = defaultPayloadConverter.toPayload(currentTraceparent()); + const spaceTraceId = correlation["space.trace_id"]; + if (spaceTraceId) { + input.headers[WORKFLOW_TRACE_ID_HEADER] = defaultPayloadConverter.toPayload(spaceTraceId); + } + return next(input); + }, + + async startChildWorkflowExecution( + input: StartChildWorkflowExecutionInput, + next: Next + ) { + input.headers[WORKFLOW_TRACEPARENT_HEADER] = defaultPayloadConverter.toPayload(currentTraceparent()); + const spaceTraceId = correlation["space.trace_id"]; + if (spaceTraceId) { + input.headers[WORKFLOW_TRACE_ID_HEADER] = defaultPayloadConverter.toPayload(spaceTraceId); + } + return next(input); + }, + + async continueAsNew(input: ContinueAsNewInput, next: Next) { + input.headers[WORKFLOW_CONTINUED_TRACEPARENT_HEADER] = defaultPayloadConverter.toPayload(currentTraceparent()); + input.headers[WORKFLOW_TRACEPARENT_HEADER] = defaultPayloadConverter.toPayload( + parentSpanId ? formatTraceparent(traceId, parentSpanId, flags) : currentTraceparent() + ); + const spaceTraceId = correlation["space.trace_id"]; + if (spaceTraceId) { + input.headers[WORKFLOW_TRACE_ID_HEADER] = defaultPayloadConverter.toPayload(spaceTraceId); + } + emitFinish("CONTINUED_AS_NEW"); + return next(input); + } + }; + + return { inbound: [inbound], outbound: [outbound] }; +}; diff --git a/apps/worker/src/workflow-observability.ts b/apps/worker/src/workflow-observability.ts new file mode 100644 index 000000000..753ef8680 --- /dev/null +++ b/apps/worker/src/workflow-observability.ts @@ -0,0 +1,144 @@ +import { existsSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import type { InjectedSinks } from "@temporalio/worker"; +import { sanitizeSpanAttributes, type SpanExporter } from "@space/runtime"; +import { + type WorkflowObservabilitySinks, + type WorkflowSpanRecordPayload +} from "./workflow-observability-interceptors.js"; + +/** Compiled `.js` in production, the `.ts` source when tests bundle the worker. */ +function resolveWorkflowObservabilityInterceptorPath(): string { + const compiled = fileURLToPath(new URL("./workflow-observability-interceptors.js", import.meta.url)); + if (existsSync(compiled)) return compiled; + const source = compiled.replace(/\.js$/, ".ts"); + return existsSync(source) ? source : compiled; +} + +export const workflowObservabilityInterceptorPath = resolveWorkflowObservabilityInterceptorPath(); + +export interface ActiveWorkflowTraceState { + spanId: string; + traceparent: string; + correlation: Record; + startedAtMs: number; +} + +const activeWorkflows = new Map(); +const MAX_ACTIVE_WORKFLOWS = 4096; + +function workflowKey(workflowId: string, runId: string): string { + return `${workflowId}\u0000${runId}`; +} + +export function rememberActiveWorkflowTrace( + workflowId: string, + runId: string, + state: ActiveWorkflowTraceState +): void { + const key = workflowKey(workflowId, runId); + activeWorkflows.delete(key); + activeWorkflows.set(key, state); + while (activeWorkflows.size > MAX_ACTIVE_WORKFLOWS) { + const oldest = activeWorkflows.keys().next().value; + if (oldest === undefined) break; + activeWorkflows.delete(oldest); + } +} + +export function lookupActiveWorkflowTrace( + workflowId: string | null | undefined, + runId: string | null | undefined +): ActiveWorkflowTraceState | null { + if (!workflowId || !runId) return null; + return activeWorkflows.get(workflowKey(workflowId, runId)) ?? null; +} + +export interface WorkflowObservabilityOptions { + spans?: SpanExporter | null; + now?: () => number; +} + +export function recordWorkflowSpan( + payload: WorkflowSpanRecordPayload, + options: WorkflowObservabilityOptions & { startedAtMs?: number } = {} +): void { + const spans = options.spans; + if (!spans) return; + const now = options.now ?? Date.now; + const finishedAtMs = now(); + const key = workflowKey(payload.workflowId, payload.runId); + const existing = activeWorkflows.get(key); + const startedAtMs = options.startedAtMs ?? existing?.startedAtMs ?? finishedAtMs; + const durationMs = Math.max(0, finishedAtMs - startedAtMs); + + const rawAttributes: Record = { + ...payload.correlation, + "space.workflow_type": payload.workflowType, + "space.workflow_id": payload.workflowId, + "space.run_id": payload.runId, + "space.outcome": payload.outcome, + "space.duration_ms": durationMs + }; + if (payload.parentWorkflowId) rawAttributes["space.parent_workflow_id"] = payload.parentWorkflowId; + if (payload.parentRunId) rawAttributes["space.parent_run_id"] = payload.parentRunId; + if (payload.continuedFromRunId) rawAttributes["space.continued_from_run_id"] = payload.continuedFromRunId; + if (payload.reasonCode) rawAttributes["space.reason_code"] = payload.reasonCode; + if (payload.category) rawAttributes["space.category"] = payload.category; + + spans.record({ + traceId: payload.traceId, + spanId: payload.spanId, + parentSpanId: payload.parentSpanId, + name: `workflow ${payload.workflowType}`, + startTimeUnixNano: BigInt(startedAtMs) * 1_000_000n, + endTimeUnixNano: BigInt(startedAtMs + durationMs) * 1_000_000n, + attributes: sanitizeSpanAttributes(rawAttributes), + error: payload.outcome === "FAILED", + kind: 1, + ...(payload.links.length ? { links: payload.links } : {}) + }); +} + +/** + * Temporal worker sinks for deterministic workflow span export. + * Both functions set `callDuringReplay: false` so workflow replay never emits duplicate spans. + */ +export function createWorkflowObservabilitySinks( + options: WorkflowObservabilityOptions = {} +): InjectedSinks { + const now = options.now ?? Date.now; + return { + workflowObservability: { + workflowStarted: { + callDuringReplay: false, + fn(info, spanId, traceparent, correlation) { + const runStartMs = info.runStartTime instanceof Date ? info.runStartTime.getTime() : now(); + rememberActiveWorkflowTrace(info.workflowId, info.runId, { + spanId, + traceparent, + correlation: correlation ?? {}, + startedAtMs: Number.isFinite(runStartMs) ? Math.min(runStartMs, now()) : now() + }); + } + }, + workflowFinished: { + callDuringReplay: false, + fn(info, payload) { + // After a worker restart/cache eviction, workflowStarted was skipped + // during replay (callDuringReplay: false), so the in-memory start + // time is gone. Prefer the remembered start, then Temporal's + // runStartTime, so long-running workflows do not export duration 0. + const existing = lookupActiveWorkflowTrace(info.workflowId, info.runId); + const runStartMs = info.runStartTime instanceof Date ? info.runStartTime.getTime() : Number.NaN; + const startedAtMs = + existing?.startedAtMs ?? (Number.isFinite(runStartMs) ? runStartMs : undefined); + recordWorkflowSpan(payload, { + ...options, + ...(startedAtMs !== undefined ? { startedAtMs } : {}) + }); + } + } + } + }; +} diff --git a/apps/worker/tests/activity-observability.test.ts b/apps/worker/tests/activity-observability.test.ts index 654bf7962..4390fa5f5 100644 --- a/apps/worker/tests/activity-observability.test.ts +++ b/apps/worker/tests/activity-observability.test.ts @@ -182,11 +182,13 @@ describe("worker metrics endpoint", () => { }); }); -it("registers the activity interceptor on every Temporal worker, including both memory workers", async () => { +it("registers the activity and workflow observability interceptors and sinks on every Temporal worker, including both memory workers", async () => { 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\(\{\s*spans\s*\}\)\]/); + expect(source, entry).toMatch(/workflowModules:\s*\[workflowObservabilityInterceptorPath\]/); + expect(source, entry).toMatch(/sinks:\s*createWorkflowObservabilitySinks\(\{\s*spans\s*\}\)/); } }); diff --git a/apps/worker/tests/otlp-probe-workflows.ts b/apps/worker/tests/otlp-probe-workflows.ts new file mode 100644 index 000000000..f2ac504ac --- /dev/null +++ b/apps/worker/tests/otlp-probe-workflows.ts @@ -0,0 +1,46 @@ +import { continueAsNew, executeChild, proxyActivities } from "@temporalio/workflow"; + +export interface OtlpProbeInput { + traceparent: string; + traceId: string; + prompt?: string; +} + +const flaky = proxyActivities<{ otlpFlakyActivity(): Promise }>({ + startToCloseTimeout: "20 seconds", + retry: { maximumAttempts: 2, initialInterval: "1 millisecond", backoffCoefficient: 1 } +}); + +const once = proxyActivities<{ otlpOnceActivity(label: string): Promise }>({ + startToCloseTimeout: "20 seconds", + retry: { maximumAttempts: 1 } +}); + +/** Parent workflow: one retried activity, then a child workflow on the same trace. */ +export async function otlpParentWorkflow(input: OtlpProbeInput): Promise { + await flaky.otlpFlakyActivity(); + return executeChild(otlpChildWorkflow, { + workflowId: `otlp-child:${input.traceId}`, + args: [input] + }); +} + +export async function otlpChildWorkflow(input: OtlpProbeInput): Promise { + return once.otlpOnceActivity("child"); +} + +/** First run continues as new; the next run links back to this one. */ +export async function otlpContinueWorkflow(input: OtlpProbeInput & { remaining: number }): Promise { + if (input.remaining > 0) { + await continueAsNew({ ...input, remaining: input.remaining - 1 }); + } + await once.otlpOnceActivity("continued"); + return "done"; +} + +/** Background continuation: own W3C trace, span link to the caller's traceparent. */ +export async function otlpBackgroundWorkflow( + input: OtlpProbeInput & { roomAgentContinuation: { sourceRunId: string; generation: number } } +): Promise { + return once.otlpOnceActivity("background"); +} diff --git a/apps/worker/tests/workflow-otlp-observability.test.ts b/apps/worker/tests/workflow-otlp-observability.test.ts new file mode 100644 index 000000000..819466e47 --- /dev/null +++ b/apps/worker/tests/workflow-otlp-observability.test.ts @@ -0,0 +1,627 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import { defaultPayloadConverter, Worker } from "@temporalio/worker"; +import { TestWorkflowEnvironment } from "@temporalio/testing"; +import { + formatTraceparent, + parseTraceparent +} from "@space/contracts"; +import { + ALLOWED_SPAN_ATTRIBUTE_KEYS, + createOtlpSpanExporter, + sanitizeSpanAttributes, + type SpanExporter +} from "@space/runtime"; +import { + createActivityObservabilityInterceptors, + noteActivityFailureReason, + observeActivityExecution +} from "../src/activity-observability.js"; +import { + createWorkflowObservabilitySinks, + recordWorkflowSpan, + rememberActiveWorkflowTrace, + workflowObservabilityInterceptorPath +} from "../src/workflow-observability.js"; +import { + continuationSpanId, + extractWorkflowInputMetadata, + WORKFLOW_CONTINUED_TRACEPARENT_HEADER, + WORKFLOW_TRACEPARENT_HEADER, + workflowSpanId +} from "../src/workflow-observability-interceptors.js"; + +interface OtlpJsonAttribute { + key: string; + value: { stringValue?: string; intValue?: string }; +} + +interface OtlpJsonLink { + traceId: string; + spanId: string; + attributes?: OtlpJsonAttribute[]; +} + +interface OtlpJsonSpan { + traceId: string; + spanId: string; + parentSpanId?: string; + name: string; + kind: number; + startTimeUnixNano: string; + endTimeUnixNano: string; + attributes: OtlpJsonAttribute[]; + links?: OtlpJsonLink[]; + status: { code: number }; +} + +function attrMap(attrs: readonly OtlpJsonAttribute[] | undefined): Record { + const map: Record = {}; + for (const item of attrs ?? []) { + if (item.value.intValue !== undefined) map[item.key] = Number(item.value.intValue); + else if (item.value.stringValue !== undefined) map[item.key] = item.value.stringValue; + } + return map; +} + +async function startFakeOtlpCollector() { + const batches: Array<{ serviceName: string; spans: OtlpJsonSpan[] }> = []; + const rawPayloads: string[] = []; + const server = createServer((req: IncomingMessage, res: ServerResponse) => { + if (req.method === "POST" && req.url === "/v1/traces") { + let body = ""; + req.on("data", (chunk) => { body += String(chunk); }); + req.on("end", () => { + rawPayloads.push(body); + const parsed = JSON.parse(body) as { + resourceSpans?: Array<{ + resource?: { attributes?: OtlpJsonAttribute[] }; + scopeSpans?: Array<{ spans?: OtlpJsonSpan[] }>; + }>; + }; + for (const rs of parsed.resourceSpans ?? []) { + const serviceName = String(attrMap(rs.resource?.attributes)["service.name"] ?? ""); + for (const ss of rs.scopeSpans ?? []) { + batches.push({ serviceName, spans: ss.spans ?? [] }); + } + } + res.writeHead(200, { "content-type": "application/json" }); + res.end("{}"); + }); + return; + } + res.writeHead(404); + res.end(); + }); + + await new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve())); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Failed to bind fake OTLP collector"); + const endpoint = `http://127.0.0.1:${address.port}`; + + return { + endpoint, + batches, + rawPayloads, + allSpans: () => batches.flatMap((b) => b.spans), + close: () => new Promise((resolve) => server.close(() => resolve())) + }; +} + +const activeExporters: SpanExporter[] = []; +afterEach(async () => { + await Promise.all(activeExporters.splice(0).map((e) => e.close())); +}); + +describe("OTLP workflow and activity trace tree (#61)", () => { + const w3cTraceId = "4bf92f3577b34da6a3ce929d0e0e4736"; + const apiSpanId = "00f067aa0ba902b7"; + const apiTraceparent = formatTraceparent(w3cTraceId, apiSpanId, "01"); + const sensitivePrompt = "PRIVATE PROMPT sk-live-secret-token-999 /Users/operator/secret.txt https://internal.local/secret?key=1"; + + it("enforces the strict attribute allowlist and strips prompts, paths, URLs, error messages and secrets", () => { + const dirty: Record = { + "space.trace_id": "req:abcdefghijkl", + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.activity_type": "runCodexAppServerTurn", + "space.attempt": 2, + "space.outcome": "FAILED", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT", + "space.duration_ms": 145, + "space.room_id": "room:alpha", + "space.pane_id": "pane:beta", + "space.mission_id": "mission:gamma", + "prompt": sensitivePrompt, + "error.message": "Connection refused to 10.0.0.12 with sk-live-secret", + "file.path": "/Users/operator/secret.txt", + "url.full": "https://internal.local/secret?key=1", + "space.agent_run_id": sensitivePrompt + }; + + const cleaned = sanitizeSpanAttributes(dirty); + expect(Object.keys(cleaned).every((k) => ALLOWED_SPAN_ATTRIBUTE_KEYS.has(k))).toBe(true); + expect(cleaned).toEqual({ + "space.trace_id": "req:abcdefghijkl", + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.activity_type": "runCodexAppServerTurn", + "space.attempt": 2, + "space.outcome": "FAILED", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT", + "space.duration_ms": 145, + "space.room_id": "room:alpha", + "space.pane_id": "pane:beta", + "space.mission_id": "mission:gamma" + }); + expect(JSON.stringify(cleaned)).not.toMatch(/PRIVATE PROMPT|sk-live|secret\.txt|internal\.local/); + }); + + it("connects API SERVER span -> workflow span -> retrying activity attempts -> child workflow span in one OTLP trace", async () => { + const collector = await startFakeOtlpCollector(); + try { + const apiExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-api", + flushIntervalMs: 60_000 + }); + const workerExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + activeExporters.push(apiExporter, workerExporter); + + // 1. API records SERVER span + apiExporter.record({ + traceId: w3cTraceId, + spanId: apiSpanId, + parentSpanId: null, + name: "POST /api/turns", + startTimeUnixNano: 1_700_000_000_000_000_000n, + endTimeUnixNano: 1_700_000_000_050_000_000n, + attributes: { + "http.request.method": "POST", + "http.route": "/api/turns", + "http.response.status_code": 200, + "space.request_id": "req:turn00000001" + }, + error: false, + kind: 2 + }); + + // 2. Parent workflow (`memoryMaintenanceWorkflow`) starts under the API span's traceparent + const parentWfId = "space-memory-maintenance:2026-10-28"; + const parentRunId = "run-parent-1"; + const parentWfSpanId = workflowSpanId(parentWfId, parentRunId); + const parentWfTraceparent = formatTraceparent(w3cTraceId, parentWfSpanId, "01"); + + const sinks = createWorkflowObservabilitySinks({ spans: workerExporter, now: () => 1_700_000_000_100 }); + sinks.workflowObservability.workflowStarted.fn( + { workflowId: parentWfId, runId: parentRunId, runStartTime: new Date(1_700_000_000_050) } as never, + parentWfSpanId, + parentWfTraceparent, + { "space.trace_id": "req:turn00000001", "space.room_id": "room:1" } + ); + + // 3. Activity attempt 1 (fails with TIMEOUT) and attempt 2 (succeeds) run as children of parentWfSpanId + const wfHeaders = { + [WORKFLOW_TRACEPARENT_HEADER]: defaultPayloadConverter.toPayload(parentWfTraceparent) + }; + const actInfo = (attempt: number) => ({ + activityType: "refreshMemoryGraphSnapshot", + activityId: "act-1", + attempt, + workflowType: "memoryMaintenanceWorkflow", + workflowExecution: { workflowId: parentWfId, runId: parentRunId }, + currentAttemptScheduledTimestampMs: 1_700_000_000_060 + }); + + await observeActivityExecution( + actInfo(1), + [{ scheduledAt: "2026-10-28T00:00:00.000Z", traceId: "req:turn00000001", prompt: sensitivePrompt }], + async () => { + noteActivityFailureReason("CODEX_APP_SERVER_TURN_TIMEOUT"); + return { status: "FAILED" }; + }, + { spans: workerExporter, headers: wfHeaders, now: () => 1_700_000_000_080 } + ); + + await observeActivityExecution( + actInfo(2), + [{ scheduledAt: "2026-10-28T00:00:00.000Z", traceId: "req:turn00000001", prompt: sensitivePrompt }], + async () => ({ status: "REFRESHED" }), + { spans: workerExporter, headers: wfHeaders, now: () => 1_700_000_000_095 } + ); + + // 4. Child workflow (`memoryConsolidationWorkflow`) nests under `parentWfSpanId` + const childWfId = "space-memory-consolidation:scheduled:2026-10-28"; + const childRunId = "run-child-1"; + const childWfSpanId = workflowSpanId(childWfId, childRunId); + sinks.workflowObservability.workflowStarted.fn( + { workflowId: childWfId, runId: childRunId, runStartTime: new Date(1_700_000_000_096) } as never, + childWfSpanId, + formatTraceparent(w3cTraceId, childWfSpanId, "01"), + { "space.trace_id": "req:turn00000001" } + ); + sinks.workflowObservability.workflowFinished.fn( + { workflowId: childWfId, runId: childRunId } as never, + { + traceId: w3cTraceId, + spanId: childWfSpanId, + parentSpanId: parentWfSpanId, + workflowType: "memoryConsolidationWorkflow", + workflowId: childWfId, + runId: childRunId, + parentWorkflowId: parentWfId, + parentRunId: parentRunId, + continuedFromRunId: null, + outcome: "COMPLETED", + reasonCode: null, + category: null, + correlation: { "space.trace_id": "req:turn00000001" }, + links: [] + } + ); + + // 5. Parent workflow finishes + sinks.workflowObservability.workflowFinished.fn( + { workflowId: parentWfId, runId: parentRunId } as never, + { + traceId: w3cTraceId, + spanId: parentWfSpanId, + parentSpanId: apiSpanId, + workflowType: "memoryMaintenanceWorkflow", + workflowId: parentWfId, + runId: parentRunId, + parentWorkflowId: null, + parentRunId: null, + continuedFromRunId: null, + outcome: "COMPLETED", + reasonCode: null, + category: null, + correlation: { "space.trace_id": "req:turn00000001", "space.room_id": "room:1" }, + links: [] + } + ); + + await apiExporter.flush(); + await workerExporter.flush(); + + const spans = collector.allSpans(); + expect(spans).toHaveLength(5); + for (const s of spans) { + expect(s.traceId).toBe(w3cTraceId); + } + + const apiSpan = spans.find((s) => s.name === "POST /api/turns")!; + const parentWfSpan = spans.find((s) => s.name === "workflow memoryMaintenanceWorkflow")!; + const childWfSpan = spans.find((s) => s.name === "workflow memoryConsolidationWorkflow")!; + const activitySpans = spans.filter((s) => s.name === "activity refreshMemoryGraphSnapshot"); + + expect(apiSpan.parentSpanId).toBeUndefined(); + expect(parentWfSpan.parentSpanId).toBe(apiSpan.spanId); + expect(childWfSpan.parentSpanId).toBe(parentWfSpan.spanId); + expect(attrMap(childWfSpan.attributes)["space.parent_workflow_id"]).toBe(parentWfId); + + expect(activitySpans).toHaveLength(2); + const attempt1 = activitySpans.find((s) => attrMap(s.attributes)["space.attempt"] === 1)!; + const attempt2 = activitySpans.find((s) => attrMap(s.attributes)["space.attempt"] === 2)!; + + expect(attempt1.parentSpanId).toBe(parentWfSpan.spanId); + expect(attempt1.status.code).toBe(2); + expect(attrMap(attempt1.attributes)).toMatchObject({ + "space.activity_type": "refreshMemoryGraphSnapshot", + "space.attempt": 1, + "space.outcome": "FAILED_RESULT", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT", + "space.trace_id": "req:turn00000001" + }); + + expect(attempt2.parentSpanId).toBe(parentWfSpan.spanId); + expect(attempt2.status.code).toBe(0); + expect(attrMap(attempt2.attributes)).toMatchObject({ + "space.activity_type": "refreshMemoryGraphSnapshot", + "space.attempt": 2, + "space.outcome": "COMPLETED", + "space.trace_id": "req:turn00000001" + }); + + expect(collector.rawPayloads.join("\n")).not.toMatch(/PRIVATE PROMPT|sk-live|secret\.txt|internal\.local/); + } finally { + await collector.close(); + } + }); + + it("attaches OTLP span links for continue-as-new runs and background continuations (trace:bg-*)", async () => { + const collector = await startFakeOtlpCollector(); + try { + const exporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + activeExporters.push(exporter); + + const prevRunId = "run-initial-01"; + const prevSpanId = workflowSpanId("dot-goal:dot_goal:1", prevRunId); + const sourceRunId = "agent_run:source-123"; + const bgSourceSpanId = continuationSpanId(sourceRunId); + + recordWorkflowSpan( + { + traceId: w3cTraceId, + spanId: workflowSpanId("dot-goal:dot_goal:1", "run-continued-02"), + parentSpanId: apiSpanId, + workflowType: "dotGoalWorkflow", + workflowId: "dot-goal:dot_goal:1", + runId: "run-continued-02", + parentWorkflowId: null, + parentRunId: null, + continuedFromRunId: prevRunId, + outcome: "COMPLETED", + reasonCode: null, + category: null, + correlation: { "space.goal_id": "dot_goal:1", "space.trace_id": "trace:bg-cont-01" }, + links: [ + { + traceId: w3cTraceId, + spanId: prevSpanId, + attributes: { + "space.link_kind": "CONTINUE_AS_NEW", + "space.continued_from_run_id": prevRunId + } + }, + { + traceId: w3cTraceId, + spanId: bgSourceSpanId, + attributes: { + "space.link_kind": "BACKGROUND_CONTINUATION", + "space.source_run_id": sourceRunId + } + } + ] + }, + { spans: exporter, startedAtMs: 1_000, now: () => 1_200 } + ); + + await exporter.flush(); + const [exported] = collector.allSpans(); + expect(exported).toBeDefined(); + expect(exported!.links).toHaveLength(2); + expect(exported!.links![0]!.spanId).toBe(prevSpanId); + expect(attrMap(exported!.links![0]!.attributes)).toEqual({ + "space.link_kind": "CONTINUE_AS_NEW", + "space.continued_from_run_id": prevRunId + }); + expect(exported!.links![1]!.spanId).toBe(bgSourceSpanId); + expect(attrMap(exported!.links![1]!.attributes)).toEqual({ + "space.link_kind": "BACKGROUND_CONTINUATION", + "space.source_run_id": sourceRunId + }); + } finally { + await collector.close(); + } + }); + + it("configures workflow sinks with callDuringReplay: false so replaying a workflow history emits zero spans", async () => { + const recorded: unknown[] = []; + const spans: SpanExporter = { + record: (span) => void recorded.push(span), + flush: async () => undefined, + close: async () => undefined, + stats: () => ({ exported: recorded.length, dropped: 0, failed: 0 }) + }; + const sinks = createWorkflowObservabilitySinks({ spans }); + expect(sinks.workflowObservability.workflowStarted.callDuringReplay).toBe(false); + expect(sinks.workflowObservability.workflowFinished.callDuringReplay).toBe(false); + + // Also verify extractWorkflowInputMetadata never leaks free-text prompts + const metadata = extractWorkflowInputMetadata([ + { + roomId: "room:replay", + paneId: "pane:replay", + traceId: "trace:bg-replay1", + traceparent: apiTraceparent, + prompt: sensitivePrompt, + roomAgentContinuation: { sourceRunId: "agent_run:orig1", generation: 1 } + } + ]); + expect(metadata.sourceRunId).toBe("agent_run:orig1"); + expect(metadata.fields.traceId).toBe("trace:bg-replay1"); + expect(JSON.stringify(metadata)).not.toContain("PRIVATE PROMPT"); + }); + + it("uses Temporal runStartTime for workflow span duration when the in-memory start was lost after restart", async () => { + const collector = await startFakeOtlpCollector(); + try { + const exporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + activeExporters.push(exporter); + + const sinks = createWorkflowObservabilitySinks({ spans: exporter, now: () => 1_700_000_000_500 }); + // Simulate a worker that finished a long-running workflow after restart: + // workflowStarted was suppressed during history replay, so no active entry. + sinks.workflowObservability.workflowFinished.fn( + { + workflowId: "dot-goal:dot_goal:restart", + runId: "run-after-restart", + runStartTime: new Date(1_700_000_000_000) + } as never, + { + traceId: w3cTraceId, + spanId: workflowSpanId("dot-goal:dot_goal:restart", "run-after-restart"), + parentSpanId: apiSpanId, + workflowType: "dotGoalWorkflow", + workflowId: "dot-goal:dot_goal:restart", + runId: "run-after-restart", + parentWorkflowId: null, + parentRunId: null, + continuedFromRunId: null, + outcome: "COMPLETED", + reasonCode: null, + category: null, + correlation: { "space.goal_id": "dot_goal:restart" }, + links: [] + } + ); + + await exporter.flush(); + const [exported] = collector.allSpans(); + expect(exported).toBeDefined(); + expect(exported!.name).toBe("workflow dotGoalWorkflow"); + expect(attrMap(exported!.attributes)["space.duration_ms"]).toBe(500); + expect(exported!.startTimeUnixNano).toBe(String(1_700_000_000_000n * 1_000_000n)); + expect(exported!.endTimeUnixNano).toBe(String(1_700_000_000_500n * 1_000_000n)); + } finally { + await collector.close(); + } + }); + + it("runs a real Temporal workflow with stub activities in @temporalio/testing and exports a connected trace to the local HTTP OTLP collector", async () => { + // createLocal uses the Temporal CLI dev server, which has an arm64 build. + // createTimeSkipping's Java test server does not run on Apple Silicon without Rosetta, + // and a skipped server must fail this test rather than pass it. + const testEnv = await TestWorkflowEnvironment.createLocal(); + const collector = await startFakeOtlpCollector(); + const workerExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + activeExporters.push(workerExporter); + + try { + const workflowsPath = fileURLToPath(new URL("./otlp-probe-workflows.ts", import.meta.url)); + const interceptorPath = fileURLToPath(new URL("../src/workflow-observability-interceptors.ts", import.meta.url)); + let flakyAttempts = 0; + const activities = { + otlpFlakyActivity: async () => { + flakyAttempts += 1; + if (flakyAttempts === 1) { + throw Object.assign(new Error("transient failure sk-live-secret"), { code: "UPSTREAM_RETRYABLE_UNAVAILABLE" }); + } + return "ok"; + }, + otlpOnceActivity: async (label: string) => label + }; + + const taskQueue = "otlp-integration-test-queue"; + const worker = await Worker.create({ + connection: testEnv.nativeConnection, + taskQueue, + workflowsPath, + activities, + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans: workerExporter })], + workflowModules: [interceptorPath] + }, + sinks: createWorkflowObservabilitySinks({ spans: workerExporter }) + }); + + const probeInput = { + traceparent: apiTraceparent, + traceId: "req:otlp00000001", + prompt: sensitivePrompt + }; + + const parentHandle = await worker.runUntil(async () => { + const parent = await testEnv.client.workflow.start("otlpParentWorkflow", { + taskQueue, + workflowId: "otlp-parent:req:otlp00000001", + args: [probeInput] + }); + const continued = await testEnv.client.workflow.start("otlpContinueWorkflow", { + taskQueue, + workflowId: "otlp-continue:req:otlp00000001", + args: [{ ...probeInput, remaining: 1 }] + }); + const background = await testEnv.client.workflow.start("otlpBackgroundWorkflow", { + taskQueue, + workflowId: "otlp-background:trace:bg-cont01", + args: [{ + ...probeInput, + traceId: "trace:bg-cont01", + roomAgentContinuation: { sourceRunId: "agent_run:source-123", generation: 1 } + }] + }); + await Promise.all([parent.result(), continued.result(), background.result()]); + return parent; + }); + + await workerExporter.flush(); + const spans = collector.allSpans(); + const parentWf = spans.find((s) => s.name === "workflow otlpParentWorkflow")!; + const childWf = spans.find((s) => s.name === "workflow otlpChildWorkflow")!; + const flakySpans = spans.filter((s) => s.name === "activity otlpFlakyActivity"); + const childActivity = spans.find((s) => s.name === "activity otlpOnceActivity" && attrMap(s.attributes)["space.workflow_type"] === "otlpChildWorkflow"); + + expect(parentWf).toBeDefined(); + expect(parentWf.traceId).toBe(w3cTraceId); + expect(parentWf.parentSpanId).toBe(apiSpanId); + + expect(childWf).toBeDefined(); + expect(childWf.traceId).toBe(w3cTraceId); + expect(childWf.parentSpanId).toBe(parentWf.spanId); + + expect(flakySpans).toHaveLength(2); + const failedAttempt = flakySpans.find((s) => attrMap(s.attributes)["space.attempt"] === 1)!; + const okAttempt = flakySpans.find((s) => attrMap(s.attributes)["space.attempt"] === 2)!; + expect(failedAttempt.parentSpanId).toBe(parentWf.spanId); + expect(failedAttempt.traceId).toBe(w3cTraceId); + expect(attrMap(failedAttempt.attributes)["space.outcome"]).toBe("FAILED"); + expect(attrMap(failedAttempt.attributes)["space.reason_code"]).toBe("UPSTREAM_RETRYABLE_UNAVAILABLE"); + expect(okAttempt.parentSpanId).toBe(parentWf.spanId); + expect(attrMap(okAttempt.attributes)["space.outcome"]).toBe("COMPLETED"); + + expect(childActivity).toBeDefined(); + expect(childActivity!.traceId).toBe(w3cTraceId); + expect(childActivity!.parentSpanId).toBe(childWf.spanId); + + const continueSpans = spans.filter((s) => s.name === "workflow otlpContinueWorkflow"); + expect(continueSpans).toHaveLength(2); + for (const span of continueSpans) { + expect(span.traceId).toBe(w3cTraceId); + expect(span.parentSpanId).toBe(apiSpanId); + } + const continued = continueSpans.find((s) => (s.links ?? []).some((link) => attrMap(link.attributes)["space.link_kind"] === "CONTINUE_AS_NEW")); + expect(continued).toBeDefined(); + const continueLink = continued!.links!.find((link) => attrMap(link.attributes)["space.link_kind"] === "CONTINUE_AS_NEW")!; + expect(continueLink.traceId).toBe(w3cTraceId); + expect(continueSpans.some((s) => s.spanId === continueLink.spanId)).toBe(true); + + const background = spans.find((s) => s.name === "workflow otlpBackgroundWorkflow"); + expect(background).toBeDefined(); + expect(background!.traceId).not.toBe(w3cTraceId); + const bgLink = background!.links?.find((link) => attrMap(link.attributes)["space.link_kind"] === "BACKGROUND_CONTINUATION"); + expect(bgLink).toBeDefined(); + expect(bgLink!.traceId).toBe(w3cTraceId); + expect(bgLink!.spanId).toBe(apiSpanId); + expect(attrMap(bgLink!.attributes)["space.source_run_id"]).toBe("agent_run:source-123"); + const bgActivity = spans.find((s) => s.name === "activity otlpOnceActivity" && s.traceId === background!.traceId); + expect(bgActivity?.parentSpanId).toBe(background!.spanId); + + const countBeforeReplay = collector.allSpans().length; + const history = await parentHandle.fetchHistory(); + await Worker.runReplayHistory( + { + workflowsPath, + interceptors: { workflowModules: [interceptorPath] }, + sinks: createWorkflowObservabilitySinks({ spans: workerExporter }) + }, + history + ); + await workerExporter.flush(); + expect(collector.allSpans().length).toBe(countBeforeReplay); + expect(collector.rawPayloads.join("\n")).not.toMatch(/PRIVATE PROMPT|sk-live|secret/); + } finally { + await collector.close(); + await testEnv.teardown(); + } + }, 180_000); +}); diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index e432af4c6..bec51405c 100644 --- a/docs/configuration-reference.md +++ b/docs/configuration-reference.md @@ -52,3 +52,10 @@ 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`. + +## Observability and OTLP trace export + +`SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` (unset by default) enables optional zero-dependency OTLP/HTTP JSON trace export (`/v1/traces`) across both `space-api` and all Temporal workers (`space-worker`, `space-memory-maintenance-worker`, `space-memory-mutation-worker`). When set to an `http:` or `https:` collector base URL: +- `space-api` exports one `SERVER` span per HTTP request and propagates its W3C `traceparent` into Temporal workflow inputs and memos. +- Every Temporal worker exports one `INTERNAL` workflow span per workflow run, parented to the API request span or to the parent workflow span for a child workflow. Continue-as-new keeps that parent and adds a span link to the previous run. A background continuation (`trace:bg-*`) starts its own W3C trace and links to the caller's span. Export goes through replay-safe Temporal interceptors and sinks (`callDuringReplay: false`). Each activity attempt is an `INTERNAL` child span with allowlisted identifiers, attempt number, outcome, and stable `reasonCode` / `category`. Unset, the workers export nothing. + diff --git a/docs/quality/observability.md b/docs/quality/observability.md index d5cd7080e..d2622f7e4 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -220,8 +220,14 @@ 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. +one INTERNAL workflow span per workflow execution (parented to the propagated API +`traceparent`, or to the parent workflow span for a child workflow). A continue-as-new +run keeps that parent and adds an OTLP span link to the run it continues. A +background continuation (`trace:bg-*`) starts its own W3C trace and adds a span +link to the caller's span. Spans leave the sandbox through Temporal sinks with +`callDuringReplay: false`, so replay emits no duplicates. Each activity attempt is +an INTERNAL child of its workflow span, with the allowlisted correlation ids, +activity type, attempt, outcome, and `reasonCode` / `category`. - **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 +239,24 @@ 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 functions do no export I/O.** Correlation still enters through the API + starter (input/memo). Span export is a workflow interceptor + (`workflow-observability-interceptors.ts`) plus an activity interceptor. The + interceptor only reads workflow info and writes trace headers; spans leave the + sandbox through sinks with `callDuringReplay: false`. Unset + `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` exports nothing. ### 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. +Workflow bodies (`workflows.ts`, `room-supervisor-state.ts`, `turn-runtime-policy.ts`, +`dot-goal-loop.ts`) are not instrumented. The worker loads +`workflow-observability-interceptors.ts` as a workflow module. Span ids are a pure +function of workflow id and run id (no clocks or randomness in the sandbox). Sinks +do not run during replay, so replaying a history emits no spans. +`apps/worker/tests/workflow-otlp-observability.test.ts` runs a workflow in +`@temporalio/testing` and replays its history to prove that. A history without +`traceparent` parses as before. Activity inputs are not part of Temporal's +determinism check. ## Redaction policy @@ -292,6 +294,7 @@ No credential or transcript export was added. | `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/workflow-otlp-observability.test.ts` | Allowlisted workflow and activity spans, parent/child trace, continue-as-new and background links, Temporal test server, replay emits no spans | | `apps/api/tests/observability-fault-injection.test.ts` | Four injected executor faults and one Temporal start failure, end to end | ## Not covered @@ -341,9 +344,10 @@ 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 - and Browser Hosts still join traces through log lines only. +- **CLI and Browser Hosts emit no spans.** With `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` + set, the API exports its SERVER span and every Temporal worker (main, memory + maintenance, memory mutation) exports the workflow span and one span per activity + attempt on that same trace. Host processes 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. - **Worker endpoints are not scraped yet.** Stable's `deploy/space-worker.service` diff --git a/docs/quality/screenshots/otlp-api-to-activity-trace.png b/docs/quality/screenshots/otlp-api-to-activity-trace.png new file mode 100644 index 000000000..62123dd1a Binary files /dev/null and b/docs/quality/screenshots/otlp-api-to-activity-trace.png differ diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index 7fb3c217a..76503b15e 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -7704,7 +7704,8 @@ export const memoryConsolidationDetailSchema = z export const memoryConsolidationWorkflowInputSchema = z .object({ runId: idSchema, - traceId: idSchema + traceId: idSchema, + traceparent: traceparentSchema.optional() }) .strict(); @@ -7916,7 +7917,8 @@ export const listMemoryChangeSetsQuerySchema = paginationRequestSchema.extend({ export const memoryMaintenanceInputSchema = z.object({ scheduledAt: isoDateTimeSchema, - traceId: idSchema + traceId: idSchema, + traceparent: traceparentSchema.optional() }); export const memoryMaintenanceResultSchema = z.object({ @@ -7988,7 +7990,8 @@ export const memoryMutationRecoveryResultSchema = z export const memoryMutationWorkflowInputSchema = z .object({ changeSetId: idSchema, - traceId: idSchema + traceId: idSchema, + traceparent: traceparentSchema.optional() }) .strict(); diff --git a/packages/runtime/src/otlp-span-exporter.ts b/packages/runtime/src/otlp-span-exporter.ts index 86a0edf05..767c89ec1 100644 --- a/packages/runtime/src/otlp-span-exporter.ts +++ b/packages/runtime/src/otlp-span-exporter.ts @@ -11,6 +11,12 @@ * request. */ +export interface SpanLink { + traceId: string; + spanId: string; + attributes?: Readonly>; +} + export interface ServerSpan { traceId: string; spanId: string; @@ -20,8 +26,10 @@ export interface ServerSpan { endTimeUnixNano: bigint; attributes: Readonly>; error: boolean; - /** OTLP span kind; SERVER (2) unless given. Worker activities use INTERNAL (1). */ + /** OTLP span kind; SERVER (2) unless given. Worker activities and workflows use INTERNAL (1). */ kind?: number; + /** Optional OTLP span links (for continue-as-new and background continuations). */ + links?: readonly SpanLink[]; } export interface SpanExporterStats { @@ -52,6 +60,70 @@ export interface OtlpSpanExporterOptions { const hexId = (length: number) => new RegExp(`^[0-9a-f]{${length}}$`); const traceIdPattern = hexId(32); const spanIdPattern = hexId(16); +const safeIdentifierValuePattern = /^[A-Za-z0-9:._-]{1,240}$/; +const safeRouteTemplatePattern = /^[A-Za-z0-9:._/*-]{1,240}$/; +const forbiddenSecretValuePattern = /(?:^|[^A-Za-z0-9])(sk-[A-Za-z0-9_-]{6,}|Bearer\s+)/i; + +export const ALLOWED_SPAN_ATTRIBUTE_KEYS: ReadonlySet = new Set([ + "http.request.method", + "http.route", + "http.response.status_code", + "space.request_id", + "space.trace_id", + "space.parent_trace_id", + "space.workflow_type", + "space.workflow_id", + "space.run_id", + "space.parent_workflow_id", + "space.parent_run_id", + "space.continued_from_run_id", + "space.source_run_id", + "space.activity_type", + "space.activity_id", + "space.attempt", + "space.outcome", + "space.status", + "space.reason_code", + "space.error_code", + "space.category", + "space.failure_category", + "space.duration_ms", + "space.queue_ms", + "space.room_id", + "space.pane_id", + "space.mission_id", + "space.agent_run_id", + "space.agent_session_id", + "space.action_id", + "space.goal_id", + "space.dot_id", + "space.turn_id", + "space.cli_task_id", + "space.revision_id", + "space.change_set_id", + "space.link_kind" +]); + +/** Filters span attributes against the strict key and identifier allowlist. */ +export function sanitizeSpanAttributes( + attributes: Readonly> | undefined +): Record { + if (!attributes) return {}; + const cleaned: Record = {}; + for (const [key, value] of Object.entries(attributes)) { + if (!ALLOWED_SPAN_ATTRIBUTE_KEYS.has(key)) continue; + if (typeof value === "number") { + if (Number.isFinite(value)) cleaned[key] = value; + continue; + } + if (typeof value === "string") { + if (forbiddenSecretValuePattern.test(value)) continue; + const pattern = key === "http.route" ? safeRouteTemplatePattern : safeIdentifierValuePattern; + if (pattern.test(value)) cleaned[key] = value; + } + } + return cleaned; +} /** The collector base URL, or null when export is off or the value is unsafe. */ export function otlpTracesUrl(value: string | undefined): string | null { @@ -76,18 +148,38 @@ export function otlpTraceRequest(spans: readonly ServerSpan[], serviceName: stri resource: { attributes: [{ key: "service.name", value: { stringValue: serviceName } }] }, scopeSpans: [{ scope: { name: serviceName }, - spans: spans.map(span => ({ - traceId: span.traceId, - spanId: span.spanId, - ...(span.parentSpanId ? { parentSpanId: span.parentSpanId } : {}), - name: span.name, - kind: span.kind ?? 2, - startTimeUnixNano: span.startTimeUnixNano.toString(), - endTimeUnixNano: span.endTimeUnixNano.toString(), - attributes: Object.entries(span.attributes).map(([key, value]) => ({ key, value: attributeValue(value) })), - // STATUS_CODE_ERROR or STATUS_CODE_UNSET - status: { code: span.error ? 2 : 0 } - })) + spans: spans.map(span => { + const safeAttrs = sanitizeSpanAttributes(span.attributes); + const validLinks = (span.links ?? []).filter( + link => traceIdPattern.test(link.traceId) && spanIdPattern.test(link.spanId) + ); + return { + traceId: span.traceId, + spanId: span.spanId, + ...(span.parentSpanId && spanIdPattern.test(span.parentSpanId) ? { parentSpanId: span.parentSpanId } : {}), + name: span.name, + kind: span.kind ?? 2, + startTimeUnixNano: span.startTimeUnixNano.toString(), + endTimeUnixNano: span.endTimeUnixNano.toString(), + attributes: Object.entries(safeAttrs).map(([key, value]) => ({ key, value: attributeValue(value) })), + ...(validLinks.length + ? { + links: validLinks.map(link => { + const linkAttrs = sanitizeSpanAttributes(link.attributes); + return { + traceId: link.traceId, + spanId: link.spanId, + ...(Object.keys(linkAttrs).length + ? { attributes: Object.entries(linkAttrs).map(([key, value]) => ({ key, value: attributeValue(value) })) } + : {}) + }; + }) + } + : {}), + // STATUS_CODE_ERROR or STATUS_CODE_UNSET + status: { code: span.error ? 2 : 0 } + }; + }) }] }] }; @@ -142,7 +234,21 @@ export function createOtlpSpanExporter(options: OtlpSpanExporterOptions): SpanEx stats.dropped += 1; return; } - queue.push(span); + queue.push({ + ...span, + attributes: sanitizeSpanAttributes(span.attributes), + ...(span.links + ? { + links: span.links + .filter(link => traceIdPattern.test(link.traceId) && spanIdPattern.test(link.spanId)) + .map(link => ({ + traceId: link.traceId, + spanId: link.spanId, + ...(link.attributes ? { attributes: sanitizeSpanAttributes(link.attributes) } : {}) + })) + } + : {}) + }); if (queue.length >= batchSize) void flush(); }, flush,