diff --git a/CHANGELOG.md b/CHANGELOG.md index 5703652d9..513e47b4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ Semantic Versioning. ## [Unreleased] +### Observability + +- Optional OTLP export connects API requests to Temporal workflow runs, child workflows and individual activity attempts, + with replay-suppressed workflow export and causal links across continuations. Export remains off by default. +- An empty collector flush no longer prevents the shared exporter from sending later spans. + ## [1.1.9] - 2026-10-25 Source: sanitized Space app @ 25bcec3e77 (5029 files). 1.1.8 was prepared but never published; its changes are part of diff --git a/apps/api/src/dot-goals.ts b/apps/api/src/dot-goals.ts index 30dd69dee..60d038fa6 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 { 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, traceId?: string): 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,13 @@ export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordi try { return await action(new Client({ connection, namespace: this.options.namespace })); } finally { await connection.close(); } } - start(goalId: string) { + start(goalId: string, traceId?: string) { + const traceparent = traceparentForTraceId(traceId); return this.client(async (client) => { try { await client.workflow.start("dotGoalWorkflow", { workflowId: dotGoalWorkflowId(goalId), taskQueue: this.options.taskQueue, - args: [{ goalId }], workflowIdReusePolicy: "REJECT_DUPLICATE" }); + args: [{ goalId, ...(traceparent ? { traceparent } : {}) }], + memo: workflowCorrelationMemo({ traceId, traceparent }), workflowIdReusePolicy: "REJECT_DUPLICATE" }); } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; } }); } @@ -125,7 +128,10 @@ 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) await deps.workflow.start(goal.goalId, request.id); + 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..050b4b93a 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,15 @@ export async function scheduleMemoryConsolidation( workflowId: string, startWorkflow: MemoryConsolidationWorkflowStarter ): Promise { - const input = memoryConsolidationWorkflowInputSchema.parse(rawInput); + const input = withTurnTraceparent(memoryConsolidationWorkflowInputSchema.parse(rawInput)); 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: workflowCorrelationMemo(input) }); return { status: "SCHEDULED", diff --git a/apps/api/src/memory-mutation-coordinator.ts b/apps/api/src/memory-mutation-coordinator.ts index f7d3d48bb..ce56aef28 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,7 +45,7 @@ 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); try { const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, { @@ -51,7 +53,8 @@ export async function scheduleMemoryMutation( taskQueue: MEMORY_MUTATION_TASK_QUEUE, workflowId, workflowIdReusePolicy: "REJECT_DUPLICATE", - workflowIdConflictPolicy: "USE_EXISTING" + workflowIdConflictPolicy: "USE_EXISTING", + memo: workflowCorrelationMemo(input) }); 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..00a736c10 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); 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 }) @@ -225,6 +226,7 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo workflowId, taskQueue: this.options.taskQueue, args: [{ roomId: item.turn.roomId, pending: [], processedCount: 0 }], + memo: workflowCorrelationMemo({ traceId: item.turn.traceId, traceparent: item.turn.traceparent, missionId: item.missionId }), signal: ROOM_AGENT_ENQUEUE_SIGNAL, signalArgs: [item] }); diff --git a/apps/api/src/trace-context.ts b/apps/api/src/trace-context.ts index af9e55efd..cdbee4f8d 100644 --- a/apps/api/src/trace-context.ts +++ b/apps/api/src/trace-context.ts @@ -96,7 +96,9 @@ export const requestTraceContexts = new TraceContextRegistry(); /** traceparent for work started on behalf of this Space request id, if known. */ export function traceparentForTraceId(traceId: string | null | undefined, registry = requestTraceContexts): string | null { - return childTraceparent(registry.lookup(traceId)?.traceparent ?? null); + // The workflow creates and exports its own child span. Handing it a freshly + // minted id here would leave that child's parent absent from the trace. + 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/dot-goals.test.ts b/apps/api/tests/dot-goals.test.ts index 0c4a38e11..3895748f2 100644 --- a/apps/api/tests/dot-goals.test.ts +++ b/apps/api/tests/dot-goals.test.ts @@ -69,6 +69,12 @@ function fixture(overrides: { runStatus?: string; pendingApproval?: boolean; sen const input = { sessionId: "agent_session:1", objective: "Ship the report", milestones: ["Draft", "Review"], criteria: ["Report published"], budget: { maxSteps: 2 } }; describe("Dot Goals", () => { + it("passes the creating HTTP request identity to the workflow coordinator", async () => { + const { service, deps } = fixture(); + const goal = await service.create("user:1", dotId, input, { id: "req:goal-trace" } as never); + expect(deps.workflow!.start).toHaveBeenCalledWith(goal.goalId, "req:goal-trace"); + }); + it("creates one open Goal per conversation for the owner's Dot and starts its workflow", async () => { const { service, deps } = fixture(); const goal = await service.create("user:1", dotId, input, null); diff --git a/apps/api/tests/memory-consolidation-coordinator.test.ts b/apps/api/tests/memory-consolidation-coordinator.test.ts index 3ff07aecb..1a824fd21 100644 --- a/apps/api/tests/memory-consolidation-coordinator.test.ts +++ b/apps/api/tests/memory-consolidation-coordinator.test.ts @@ -4,6 +4,7 @@ import { MEMORY_CONSOLIDATION_WORKFLOW_TYPE, scheduleMemoryConsolidation } from "../src/memory-consolidation-coordinator.js"; +import { createRequestTraceContext, requestTraceContexts } from "../src/trace-context.js"; describe("Memory consolidation coordinator", () => { it("starts a retry-safe workflow on the exclusive maintenance queue", async () => { @@ -24,11 +25,23 @@ describe("Memory consolidation coordinator", () => { taskQueue: MEMORY_CONSOLIDATION_TASK_QUEUE, workflowId, workflowIdReusePolicy: "REJECT_DUPLICATE", - workflowIdConflictPolicy: "USE_EXISTING" + workflowIdConflictPolicy: "USE_EXISTING", + memo: { spaceTraceId: input.traceId } } }]); }); + it("parents consolidation to the API server span without inventing an unrecorded parent", async () => { + const traceId = "req:consolidation-trace"; + const context = createRequestTraceContext(undefined); + requestTraceContexts.remember(traceId, context); + await scheduleMemoryConsolidation({ runId: "memory_consolidation:traced", traceId }, "workflow:traced", async (_type, options) => { + expect(options.args[0].traceparent).toBe(context.traceparent); + expect(options.memo).toMatchObject({ traceparent: context.traceparent, w3cTraceId: context.w3cTraceId }); + return { workflowId: options.workflowId }; + }); + }); + it("maps duplicate starts to a stable idempotent result", async () => { const duplicate = new Error("already started"); duplicate.name = "WorkflowExecutionAlreadyStartedError"; diff --git a/apps/api/tests/memory-mutation-coordinator.test.ts b/apps/api/tests/memory-mutation-coordinator.test.ts index e07db2b56..15bad5b0e 100644 --- a/apps/api/tests/memory-mutation-coordinator.test.ts +++ b/apps/api/tests/memory-mutation-coordinator.test.ts @@ -5,6 +5,7 @@ import { buildMemoryMutationWorkflowId, scheduleMemoryMutation } from "../src/memory-mutation-coordinator.js"; +import { createRequestTraceContext, requestTraceContexts } from "../src/trace-context.js"; describe("Memory mutation coordinator", () => { it("starts the dedicated workflow with deterministic duplicate guards", async () => { @@ -30,12 +31,24 @@ describe("Memory mutation coordinator", () => { taskQueue: MEMORY_MUTATION_TASK_QUEUE, workflowId: result.workflowId, workflowIdReusePolicy: "REJECT_DUPLICATE", - workflowIdConflictPolicy: "USE_EXISTING" + workflowIdConflictPolicy: "USE_EXISTING", + memo: { spaceTraceId: input.traceId } } } ]); }); + it("preserves the actual API span in mutation workflow input and memo", async () => { + const traceId = "req:mutation-trace"; + const context = createRequestTraceContext(undefined); + requestTraceContexts.remember(traceId, context); + await scheduleMemoryMutation({ changeSetId: "memory_change:traced", traceId }, async (_type, options) => { + expect(options.args[0].traceparent).toBe(context.traceparent); + expect(options.memo).toMatchObject({ traceparent: context.traceparent, w3cTraceId: context.w3cTraceId }); + return { workflowId: options.workflowId }; + }); + }); + it("turns duplicate workflow starts into an idempotent response", async () => { const duplicate = new Error("already started"); duplicate.name = "WorkflowExecutionAlreadyStartedError"; diff --git a/apps/api/tests/trace-correlation.test.ts b/apps/api/tests/trace-correlation.test.ts index 538293534..3f7eedec7 100644 --- a/apps/api/tests/trace-correlation.test.ts +++ b/apps/api/tests/trace-correlation.test.ts @@ -12,6 +12,7 @@ import { childTraceparent, createRequestTraceContext, requestTraceContexts, + traceparentForTraceId, withTurnTraceparent, workflowCorrelationMemo } from "../src/trace-context.js"; @@ -50,6 +51,8 @@ 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); + expect(traceparentForTraceId("req:dddddddddddd", registry)).toBe(filled.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" }); @@ -236,15 +239,18 @@ describe("workflow starters", () => { }; await coordinator.enqueue(item); - const signalOptions = (signalWithStart.mock.calls[0] as unknown as [string, { signalArgs: RoomAgentSupervisorQueueItem[] }])[1]; + const signalOptions = (signalWithStart.mock.calls[0] as unknown as [string, { signalArgs: RoomAgentSupervisorQueueItem[]; memo: Record }])[1]; expect(parseTraceparent(signalOptions.signalArgs[0]?.turn.traceparent)?.traceId).toBe(inboundTraceId); + expect(signalOptions.memo.traceparent).toBe(requestTraceContexts.lookup("req:roomagent0001")?.traceparent); requestTraceContexts.remember("req:internal0001", createRequestTraceContext(inbound)); await coordinator.enqueueAction("room_agent_action:1", { roomId: turn.roomId, missionId: "room_agent_mission:trace", agentPaneId: turn.paneId, agentSessionId: "agent_session:1", requestId: "req:roomagent0001", selectedToolIds: [], actions: [] } as never, "req:internal0001"); - const actionOptions = (start.mock.calls[0] as unknown as [string, { memo: Record }])[1]; + const actionOptions = (start.mock.calls[0] as unknown as [string, { args: Array<{ traceparent?: string }>; memo: Record }])[1]; + expect(actionOptions.args[0]?.traceparent).toBe(requestTraceContexts.lookup("req:internal0001")?.traceparent); + expect(actionOptions.memo.traceparent).toBe(actionOptions.args[0]?.traceparent); expect(actionOptions.memo).toMatchObject({ spaceTraceId: "req:internal0001", parentSpaceTraceId: "req:roomagent0001", 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..b880a2c40 100644 --- a/apps/worker/src/activity-observability.ts +++ b/apps/worker/src/activity-observability.ts @@ -13,6 +13,7 @@ import { import type { SpanExporter } from "@space/runtime"; import { logIdentifier, workerLog, type WorkerLogValue } from "./worker-log.js"; import { workerMetrics, type WorkerMetrics } from "./worker-metrics.js"; +import { traceFromHeaders, workflowTraceparent } from "./workflow-tracing-context.js"; /** * One structured start/finish record and metrics per activity execution, with @@ -137,6 +138,8 @@ export interface ObserveActivityDependencies { cancelled?: () => boolean; /** OTLP export (SPACE_OTEL_EXPORTER_OTLP_ENDPOINT): one span per activity whose input carries a traceparent. */ spans?: SpanExporter | null; + /** The workflow interceptor's actual span, carried independently of activity arguments. */ + parentTraceparent?: string; } function activityTraceparent(args: readonly unknown[]) { @@ -194,8 +197,10 @@ export async function observeActivityExecution( queueMs, rssBytes: process.memoryUsage().rss }); - const parent = dependencies.spans ? activityTraceparent(args) : null; + const parent = dependencies.spans ? parseTraceparent(dependencies.parentTraceparent) ?? activityTraceparent(args) : null; if (parent) { + const workflowId = logIdentifier(info.workflowExecution?.workflowId); + const runId = logIdentifier(info.workflowExecution?.runId); dependencies.spans!.record({ traceId: parent.traceId, spanId: randomBytes(8).toString("hex"), @@ -203,7 +208,13 @@ 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: { + "space.activity_type": activityType, "space.outcome": outcome, "space.attempt": info.attempt, + ...(workflowId ? { "space.workflow_id": workflowId } : {}), + ...(runId ? { "space.run_id": runId } : {}), + ...(reasonCode ? { "space.reason_code": reasonCode } : {}), + ...(category ? { "space.category": category } : {}) + }, error: outcome === "FAILED" || outcome === "FAILED_RESULT", kind: 1 }); @@ -227,10 +238,14 @@ export async function observeActivityExecution( export function createActivityObservabilityInterceptors(dependencies: Omit = {}): ActivityInterceptorsFactory { return (context) => ({ inbound: { - execute: (input, next) => observeActivityExecution(context.info, input.args, () => next(input), { - ...dependencies, - cancelled: () => context.cancellationSignal.aborted - }) + execute: (input, next) => { + const parent = dependencies.spans ? traceFromHeaders(input.headers) : null; + return observeActivityExecution(context.info, input.args, () => next(input), { + ...dependencies, + ...(parent ? { parentTraceparent: workflowTraceparent(parent) } : {}), + cancelled: () => context.cancellationSignal.aborted + }); + } } }); } diff --git a/apps/worker/src/memory-maintenance-worker.ts b/apps/worker/src/memory-maintenance-worker.ts index e12046033..179afec33 100644 --- a/apps/worker/src/memory-maintenance-worker.ts +++ b/apps/worker/src/memory-maintenance-worker.ts @@ -9,6 +9,7 @@ import { import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js"; import { otlpSpanExporterFromEnv } from "@space/runtime"; import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { createWorkflowTraceSinks } from "./workflow-tracing-sink.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 })] } + sinks: createWorkflowTraceSinks(spans), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + ...(spans ? { workflowModules: [fileURLToPath(new URL("./workflow-tracing.js", import.meta.url))] } : {}) + } }); console.log(JSON.stringify({ service, @@ -41,9 +46,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/src/memory-mutation-worker.ts b/apps/worker/src/memory-mutation-worker.ts index c2eadf657..a730ade7f 100644 --- a/apps/worker/src/memory-mutation-worker.ts +++ b/apps/worker/src/memory-mutation-worker.ts @@ -5,6 +5,7 @@ import { MEMORY_MUTATION_TASK_QUEUE, memoryMutationsEnabled, memoryMutationWorke import { DEFAULT_TEMPORAL_ADDRESS, DEFAULT_TEMPORAL_NAMESPACE } from "./constants.js"; import { otlpSpanExporterFromEnv } from "@space/runtime"; import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { createWorkflowTraceSinks } from "./workflow-tracing-sink.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 })] } + sinks: createWorkflowTraceSinks(spans), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + ...(spans ? { workflowModules: [fileURLToPath(new URL("./workflow-tracing.js", import.meta.url))] } : {}) + } }); 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..a69b12c5b 100644 --- a/apps/worker/src/room-background-continuation.ts +++ b/apps/worker/src/room-background-continuation.ts @@ -2,6 +2,8 @@ import { buildCodexAppServerTurnWorkflowId, buildRoomAgentSupervisorWorkflowId, roomAgentSupervisorQueueItemSchema, + formatTraceparent, + parseTraceparent, spaceAgentRoomActionRequestSchema, type RoomAgentActionRecord, type RoomAgentSupervisorQueueItem @@ -122,6 +124,7 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis ].join("\n"); const prefix = `${instructions}\n\nOperator request:\n${request}\n\nSettled action results (bounded; inspect for details):\n`; const prompt = prefix + continuationEvidence(actions, Math.max(0, 8000 - prefix.length)); + const sourceTrace = parseTraceparent(source.queueItem.turn.traceparent); const item = roomAgentSupervisorQueueItemSchema.parse({ missionId: input.missionId, turn: { ...source.queueItem.turn, prompt, roomAgentOperatorRequest: request, @@ -130,7 +133,12 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis agentUserMessageId: `agent_msg:bg-context-${input.continuationId}`, agentAssistantMessageId: `agent_msg:bg-response-${input.continuationId}`, // Resolve the current native thread from the durable session when executing. - agentThreadId: null, traceId: `trace:bg-${input.continuationId}` + agentThreadId: null, traceId: `trace:bg-${input.continuationId}`, + // Keep the originating W3C trace and an explicit causal link even though + // the Space correlation id changes for the durable background turn. + ...(sourceTrace ? { + traceLink: formatTraceparent(sourceTrace.traceId, sourceTrace.parentId, sourceTrace.flags) + } : {}) } }); try { const enqueued = await store.enqueueRoomAgentMission({ diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 83d23cbc6..81afc15bb 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -7,6 +7,7 @@ import { TelegramDeliveryWorker, TelegramSecretStore, otlpSpanExporterFromEnv } import { NativeConnection, Runtime, Worker } from "@temporalio/worker"; import * as activities from "./activities.js"; import { createActivityObservabilityInterceptors } from "./activity-observability.js"; +import { createWorkflowTraceSinks } from "./workflow-tracing-sink.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 })] } + sinks: createWorkflowTraceSinks(spans), + interceptors: { + activity: [createActivityObservabilityInterceptors({ spans })], + ...(spans ? { workflowModules: [fileURLToPath(new URL("./workflow-tracing.js", import.meta.url))] } : {}) + } }); // Optional loopback-only Prometheus endpoint for worker metrics. diff --git a/apps/worker/src/workflow-tracing-context.ts b/apps/worker/src/workflow-tracing-context.ts new file mode 100644 index 000000000..5ef4ed60f --- /dev/null +++ b/apps/worker/src/workflow-tracing-context.ts @@ -0,0 +1,101 @@ +import type { Headers, Sinks } from "@temporalio/workflow"; + +/** The same W3C value is used in Temporal headers, arguments and memo. */ +export const TRACE_HEADER = "traceparent"; +export const TRACE_LINK_HEADER = "space-trace-link"; + +export interface TraceContext { traceId: string; spanId: string; flags: string } +export interface WorkflowSpanRecord { + traceId: string; + spanId: string; + parentSpanId: string | null; + startedAtMs: number; + endedAtMs: number; + outcome: "COMPLETED" | "FAILED" | "CANCELLED" | "CONTINUED"; + links?: readonly { traceId: string; spanId: string }[]; + correlation?: Record; +} +export interface WorkflowTracingSinks extends Sinks { + spaceTracing: { record(span: WorkflowSpanRecord): void }; +} + +export function parseWorkflowTraceparent(value: unknown): TraceContext | null { + if (typeof value !== "string") return null; + const match = /^00-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/.exec(value); + if (!match || /^0+$/.test(match[1]!) || /^0+$/.test(match[2]!)) return null; + return { traceId: match[1]!, spanId: match[2]!, flags: match[3]! }; +} + +export function workflowTraceparent(context: TraceContext): string { + return `00-${context.traceId}-${context.spanId}-${context.flags}`; +} + +/** Header values are short ASCII JSON strings; no converter or I/O is needed in the sandbox. */ +export function traceHeader(value: string): Headers[string] { + return { + metadata: { encoding: Uint8Array.from("json/plain", char => char.charCodeAt(0)) }, + data: Uint8Array.from(JSON.stringify(value), char => char.charCodeAt(0)) + }; +} + +export function traceFromHeaders(headers: Headers, key = TRACE_HEADER): TraceContext | null { + const payload = headers[key]; + if (!payload?.data || payload.data.length > 256) return null; + try { + return parseWorkflowTraceparent(JSON.parse(String.fromCharCode(...payload.data))); + } catch { return null; } +} + +const nestedKeys = ["turn", "item", "bridge", "input", "turnInput"] as const; +const correlationKeys = ["roomId", "paneId", "missionId", "roomAgentMissionId", "goalId", "actionId", "turnId", "agentRunId"] as const; +const safeIdentifier = (value: unknown): value is string => typeof value === "string" && /^[A-Za-z0-9:._-]{1,200}$/.test(value); + +export function traceArgument(args: readonly unknown[], key = TRACE_HEADER): TraceContext | null { + const visit = (value: unknown, depth: number): TraceContext | null => { + if (!value || typeof value !== "object" || Array.isArray(value)) return null; + const record = value as Record; + const direct = parseWorkflowTraceparent(record[key]); + if (direct) return direct; + if (!depth) return null; + for (const name of nestedKeys) { + const found = visit(record[name], depth - 1); + if (found) return found; + } + return null; + }; + for (const arg of args) { + const found = visit(arg, 2); + if (found) return found; + } + return null; +} + +export function workflowCorrelation(args: readonly unknown[]): Record { + const result: Record = {}; + const visit = (value: unknown, depth: number) => { + if (!value || typeof value !== "object" || Array.isArray(value)) return; + const record = value as Record; + for (const key of correlationKeys) { + if (safeIdentifier(record[key])) result[key] ??= record[key]; + } + if (depth) for (const key of nestedKeys) visit(record[key], depth - 1); + }; + for (const arg of args) visit(arg, 2); + return result; +} + +/** Stable tracing IDs do not consume the workflow's deterministic random stream. */ +export function workflowSpanId(value: string): string { + let left = 0x811c9dc5; + let right = 0x9e3779b9; + for (let index = 0; index < value.length; index++) { + const code = value.charCodeAt(index); + left = Math.imul(left ^ code, 0x01000193); + right = Math.imul(right ^ code, 0x85ebca6b); + } + return `${(left >>> 0).toString(16).padStart(8, "0")}${(right >>> 0).toString(16).padStart(8, "0")}`.replace(/^0{16}$/, "0000000000000001"); +} + +export function safeWorkflowIdentifier(value: unknown): string | null { + return safeIdentifier(value) ? value : null; +} diff --git a/apps/worker/src/workflow-tracing-sink.ts b/apps/worker/src/workflow-tracing-sink.ts new file mode 100644 index 000000000..ed8f0a4fa --- /dev/null +++ b/apps/worker/src/workflow-tracing-sink.ts @@ -0,0 +1,40 @@ +import type { InjectedSinks } from "@temporalio/worker"; +import type { SpanExporter } from "@space/runtime"; +import { + safeWorkflowIdentifier, + workflowCorrelation, + type WorkflowTracingSinks +} from "./workflow-tracing-context.js"; + +/** A sink only enqueues; HTTP export remains outside the workflow task path. */ +export function createWorkflowTraceSinks(spans: SpanExporter | null | undefined): InjectedSinks | undefined { + if (!spans) return undefined; + return { + spaceTracing: { + record: { + callDuringReplay: false, + fn(info, span) { + if (!["COMPLETED", "FAILED", "CANCELLED", "CONTINUED"].includes(span.outcome)) return; + const workflowType = safeWorkflowIdentifier(info.workflowType) ?? "unknown"; + const attributes: Record = { "space.workflow_type": workflowType, "space.outcome": span.outcome }; + for (const [key, value] of Object.entries({ + workflowId: info.workflowId, runId: info.runId, ...workflowCorrelation([span.correlation]) + })) { + const safe = safeWorkflowIdentifier(value); + if (safe) attributes[`space.${key.replace(/[A-Z]/g, letter => `_${letter.toLowerCase()}`)}`] = safe; + } + if (!Number.isFinite(span.startedAtMs) || !Number.isFinite(span.endedAtMs)) return; + try { + spans.record({ + traceId: span.traceId, spanId: span.spanId, parentSpanId: span.parentSpanId, + name: `workflow ${workflowType}`, kind: 1, + startTimeUnixNano: BigInt(Math.trunc(span.startedAtMs)) * 1_000_000n, + endTimeUnixNano: BigInt(Math.trunc(Math.max(span.startedAtMs, span.endedAtMs))) * 1_000_000n, + attributes, error: span.outcome === "FAILED", links: span.links + }); + } catch { /* Best-effort tracing must not fail a workflow task. */ } + } + } + } + }; +} diff --git a/apps/worker/src/workflow-tracing.ts b/apps/worker/src/workflow-tracing.ts new file mode 100644 index 000000000..ed776407b --- /dev/null +++ b/apps/worker/src/workflow-tracing.ts @@ -0,0 +1,166 @@ +import { + ContinueAsNew, + isCancellation, + proxySinks, + workflowInfo, + type WorkflowInterceptors +} from "@temporalio/workflow"; +import { + TRACE_HEADER, TRACE_LINK_HEADER, parseWorkflowTraceparent, traceArgument, traceFromHeaders, + traceHeader, workflowCorrelation, workflowSpanId, workflowTraceparent, + type TraceContext, type WorkflowSpanRecord, type WorkflowTracingSinks +} from "./workflow-tracing-context.js"; + +/** This module is bundled into the workflow isolate only when export is enabled. */ +export function interceptors(): WorkflowInterceptors { + const { spaceTracing } = proxySinks(); + const info = workflowInfo(); + let root: WorkflowSpanRecord | null = null; + let rootParent: TraceContext | null = null; + let rootArgumentParent: TraceContext | null = null; + // A long-lived supervisor can process requests from several API traces. Do not + // parent every later mission under the request that originally started it. + const missions = new Map(); + const missionIds = new Map(); + const closed = new Set(); + const operations = new Map(); + const childCompleted = new Set(); + const supervisor = info.workflowType === "roomAgentSupervisorWorkflow"; + const finished = (span: WorkflowSpanRecord, outcome: WorkflowSpanRecord["outcome"]) => { + if (closed.has(span)) return; + closed.add(span); + span.endedAtMs = Date.now(); + spaceTracing.record({ ...span, outcome }); + }; + const makeSpan = (parent: TraceContext | null, args: readonly unknown[], link: TraceContext | null = null, discriminator = "run"): WorkflowSpanRecord => ({ + traceId: parent?.traceId ?? `${workflowSpanId(info.runId)}${workflowSpanId(`trace:${info.runId}`)}`, + spanId: workflowSpanId(`${info.runId}:${parent?.traceId ?? "root"}:${parent?.spanId ?? "root"}:${discriminator}`), + parentSpanId: parent?.spanId ?? null, + startedAtMs: Date.now(), endedAtMs: Date.now(), outcome: "COMPLETED", + correlation: workflowCorrelation(args), + ...(link ? { links: [{ traceId: link.traceId, spanId: link.spanId }] } : {}) + }); + const parentFor = (args: readonly unknown[]): WorkflowSpanRecord | null => { + const parent = traceArgument(args); + const correlation = workflowCorrelation(args); + const missionId = correlation.missionId ?? correlation.roomAgentMissionId; + if (!parent) return (missionId ? missionIds.get(missionId) : undefined) ?? root; + if ((!supervisor || !missionId) && [rootParent, rootArgumentParent].some(candidate => candidate && parent.traceId === candidate.traceId && parent.spanId === candidate.spanId)) return root; + const key = `${parent.traceId}:${parent.spanId}:${missionId ?? "request"}:${correlation.agentRunId ?? ""}`; + let span = missions.get(key); + if (!span) { + span = makeSpan(parent, args, traceArgument(args, "traceLink"), key); + // Supervisor histories are continued periodically; also cap instrumentation + // state independently, without exporting any unbounded request data. + if (missions.size >= 256) { + const oldest = missions.keys().next().value!; + const discarded = missions.get(oldest)!; + closed.delete(discarded); + operations.delete(discarded); + childCompleted.delete(discarded); + missions.delete(oldest); + missionIds.clear(); + } + missions.set(key, span); + if (missionId) missionIds.set(missionId, span); + } + return span; + }; + const headersFor = (span: WorkflowSpanRecord) => ({ + [TRACE_HEADER]: traceHeader(workflowTraceparent({ traceId: span.traceId, spanId: span.spanId, flags: "01" })) + }); + return { + inbound: [{ + async execute(input, next) { + rootArgumentParent = traceArgument(input.args); + rootParent = traceFromHeaders(input.headers) ?? rootArgumentParent ?? parseWorkflowTraceparent(info.memo?.traceparent); + const link = traceFromHeaders(input.headers, TRACE_LINK_HEADER) ?? traceArgument(input.args, "traceLink") ?? parseWorkflowTraceparent(info.memo?.traceLink); + root = makeSpan(rootParent, input.args, link); + if (supervisor) { + const first = input.args[0] as { pending?: unknown } | undefined; + if (Array.isArray(first?.pending)) for (const item of first.pending.slice(0, 256)) parentFor([item]); + } + try { + const result = await next(input); + for (const span of missions.values()) finished(span, span.outcome); + finished(root, "COMPLETED"); + return result; + } catch (error) { + const outcome = error instanceof ContinueAsNew ? "CONTINUED" : isCancellation(error) ? "CANCELLED" : "FAILED"; + for (const span of missions.values()) finished(span, span.outcome === "FAILED" ? "FAILED" : outcome); + finished(root, outcome); + throw error; + } + }, + async handleSignal(input, next) { + // Remember traced queue items so subsequent identifier-only bookkeeping + // activities can use that mission's context, without altering its payload. + parentFor(input.args); + return next(input); + } + }], + outbound: [{ + async scheduleActivity(input, next) { + const span = parentFor(input.args); + if (!span) return next(input); + const closesMission = supervisor && span !== root && ["finishRoomAgentSupervisorTurn", "markRoomAgentMissionFinished", "markRoomAgentMissionContinued"].includes(input.activityType); + operations.set(span, (operations.get(span) ?? 0) + 1); + try { + const result = await next({ ...input, headers: { ...input.headers, ...headersFor(span) } }); + if (closesMission) { + const status = (input.args[0] as { status?: unknown } | undefined)?.status; + finished(span, status === "FAILED" ? "FAILED" : status === "INTERRUPTED" ? "CANCELLED" : span.outcome); + } + return result; + } catch (error) { + if (closesMission) finished(span, isCancellation(error) ? "CANCELLED" : "FAILED"); + throw error; + } finally { + operations.set(span, (operations.get(span) ?? 1) - 1); + } + }, + async scheduleLocalActivity(input, next) { + const span = parentFor(input.args); + if (!span) return next(input); + operations.set(span, (operations.get(span) ?? 0) + 1); + try { return await next({ ...input, headers: { ...input.headers, ...headersFor(span) } }); } + finally { operations.set(span, (operations.get(span) ?? 1) - 1); } + }, + async startChildWorkflowExecution(input, next) { + const span = parentFor(input.options.args ?? []); + if (!span) return next(input); + operations.set(span, (operations.get(span) ?? 0) + 1); + let result: [Promise, Promise]; + try { result = await next({ ...input, headers: { ...input.headers, ...headersFor(span) } }); } + catch (error) { operations.set(span, (operations.get(span) ?? 1) - 1); throw error; } + // Preserve both SDK promises: replacing or awaiting the completion here + // would change child start/cancellation semantics. + void result[1].then(result => { + if (result && typeof result === "object" && (result as { status?: unknown }).status === "FAILED") span.outcome = "FAILED"; + }, () => { span.outcome = "FAILED"; }).then(() => { + operations.set(span, (operations.get(span) ?? 1) - 1); + childCompleted.add(span); + }); + return result; + }, + async continueAsNew(input, next) { + if (!root) return next(input); + const continued = workflowTraceparent({ traceId: root.traceId, spanId: root.spanId, flags: "01" }); + const parent = rootParent ? { [TRACE_HEADER]: traceHeader(workflowTraceparent(rootParent)) } : headersFor(root); + return next({ ...input, headers: { ...input.headers, ...parent, [TRACE_LINK_HEADER]: traceHeader(continued) } }); + } + }], + internals: [{ + concludeActivation(input, next) { + const result = next(input); + // A supervisor may go directly from a finished turn to waiting for + // background work or a new signal, with no finish activity. Emit at + // the activation boundary only after any final bookkeeping settled. + for (const span of missions.values()) { + if (childCompleted.has(span) && !operations.get(span)) finished(span, span.outcome); + } + return result; + } + }] + }; +} diff --git a/apps/worker/tests/activity-observability.test.ts b/apps/worker/tests/activity-observability.test.ts index 654bf7962..eba43ae02 100644 --- a/apps/worker/tests/activity-observability.test.ts +++ b/apps/worker/tests/activity-observability.test.ts @@ -1,6 +1,6 @@ import { afterEach, describe, expect, it } from "vitest"; import { request } from "node:http"; -import { MetricsRegistry } from "@space/runtime"; +import { MetricsRegistry, type ServerSpan } from "@space/runtime"; import { activityCorrelation, childTraceparent, @@ -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("activity: [createActivityObservabilityInterceptors({ spans })]"); + expect(source, entry).toContain("sinks: createWorkflowTraceSinks(spans)"); + expect(source, entry).toContain('...(spans ? { workflowModules: [fileURLToPath(new URL("./workflow-tracing.js", import.meta.url))] } : {})'); } }); @@ -203,3 +205,32 @@ it("exports one INTERNAL span per activity under the caller's traceparent", asyn expect(recorded[0]).toMatchObject({ traceId: "4bf92f3577b34da6a3ce929d0e0e4736", parentSpanId: "00f067aa0ba902b7", name: "activity runTurn", kind: 1, error: false, startTimeUnixNano: 1_000_000_000n, endTimeUnixNano: 1_250_000_000n }); }); +it("exports only allowlisted activity attributes even when inputs and failures contain private text", async () => { + const recorded: ServerSpan[] = []; + const spans = { record: (span: ServerSpan) => { recorded.push(span); }, flush: async () => undefined, + close: async () => undefined, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; + const privateText = "private fixture text /home/private/input.csv https://private.invalid"; + const args = [{ traceparent, prompt: privateText, modelOutput: privateText, filePath: privateText, url: privateText }]; + const metrics = createWorkerMetrics(new MetricsRegistry()); + const observed = info({ attempt: 2 }); + await observeActivityExecution(observed, args, async () => ({ status: "COMPLETED", output: privateText }), { metrics, spans }); + await observeActivityExecution(observed, args, async () => { + noteActivityFailureReason("CODEX_APP_SERVER_TURN_TIMEOUT"); + return { status: "FAILED", message: privateText, output: privateText }; + }, { metrics, spans }); + const thrown = Object.assign(new Error(privateText), { code: "TOOL_UNAVAILABLE", path: privateText, url: privateText }); + await expect(observeActivityExecution(observed, args, async () => { throw thrown; }, { metrics, spans })).rejects.toBe(thrown); + expect(recorded).toHaveLength(3); + for (const [index, span] of recorded.entries()) { + const attributes = { + "space.activity_type": "runCodexAppServerTurn", "space.attempt": 2, + "space.workflow_id": observed.workflowExecution.workflowId, "space.run_id": "run-uuid-1", + "space.outcome": ["COMPLETED", "FAILED_RESULT", "FAILED"][index], + ...(index === 1 ? { "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", "space.category": "TIMEOUT" } : {}), + ...(index === 2 ? { "space.reason_code": "TOOL_UNAVAILABLE", "space.category": "TOOL_UNAVAILABLE" } : {}) + }; + expect(span.attributes).toEqual(attributes); + expect(Object.keys(span.attributes).sort()).toEqual(Object.keys(attributes).sort()); + expect(JSON.stringify(span.attributes)).not.toContain(privateText); + } +}); diff --git a/apps/worker/tests/fixtures/otlp-supervisor-workflows.ts b/apps/worker/tests/fixtures/otlp-supervisor-workflows.ts new file mode 100644 index 000000000..283bff986 --- /dev/null +++ b/apps/worker/tests/fixtures/otlp-supervisor-workflows.ts @@ -0,0 +1,2 @@ +// Bundle the production workflow implementations; only activities are stubbed. +export { roomAgentSupervisorWorkflow, codexAppServerTurnWorkflow } from "../../src/workflows.js"; diff --git a/apps/worker/tests/fixtures/otlp-workflows.ts b/apps/worker/tests/fixtures/otlp-workflows.ts new file mode 100644 index 000000000..649b0bc98 --- /dev/null +++ b/apps/worker/tests/fixtures/otlp-workflows.ts @@ -0,0 +1,24 @@ +import { continueAsNew, executeChild, proxyActivities } from "@temporalio/workflow"; +import { dummyTurnWorkflow } from "../../src/workflows.js"; + +export { dummyTurnWorkflow }; + +const { tracedRetry } = proxyActivities<{ tracedRetry(): Promise }>({ + startToCloseTimeout: "10 seconds", retry: { maximumAttempts: 2, initialInterval: "1 second" } +}); + +export async function tracedRoot(input: { traceparent: string; prompt?: string }): Promise { + await tracedRetry(); + // Exercise the actual production dummy-turn workflow as a child. Its two + // activities intentionally receive no business-argument traceparent. + await executeChild(dummyTurnWorkflow, { + workflowId: "otlp-child", args: [{ ...input, traceparent: undefined } as never] + }); + return "complete"; +} + +export async function tracedContinuation(input: { traceparent: string; generation: number }): Promise { + await tracedRetry(); + if (input.generation === 0) return continueAsNew({ ...input, generation: 1 }); + return "continued"; +} diff --git a/apps/worker/tests/otlp-supervisor.integration.test.ts b/apps/worker/tests/otlp-supervisor.integration.test.ts new file mode 100644 index 000000000..56c1f7848 --- /dev/null +++ b/apps/worker/tests/otlp-supervisor.integration.test.ts @@ -0,0 +1,106 @@ +import { randomUUID } from "node:crypto"; +import { fileURLToPath } from "node:url"; +import { Worker } from "@temporalio/worker"; +import { TestWorkflowEnvironment } from "@temporalio/testing"; +import { describe, expect, it } from "vitest"; +import { roomAgentSupervisorInputSchema, type DummyTurnInput, type RoomAgentSupervisorQueueItem } from "@space/contracts"; +import type { ServerSpan, SpanExporter } from "@space/runtime"; +import { createActivityObservabilityInterceptors } from "../src/activity-observability.js"; +import { createWorkflowTraceSinks } from "../src/workflow-tracing-sink.js"; +import { continuationFixture } from "./room-continuation-fixture.js"; + +const workflowsPath = fileURLToPath(new URL("./fixtures/otlp-supervisor-workflows.ts", import.meta.url)); +const workflowModules = [fileURLToPath(new URL("../src/workflow-tracing.ts", import.meta.url))]; +const traceA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; +const traceB = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; +const parentA = "1111111111111111"; +const parentB = "2222222222222222"; + +describe("real supervisor mission tracing", () => { + it("exports two request segments once while the current supervisor remains idle", async () => { + const fixture = continuationFixture("OPEN_TYPES", `00-${traceA}-${parentA}-01`); + const first = fixture.initial.queueItem; + const followUp = fixture.enqueue("follow-up").queueItem; + const second: RoomAgentSupervisorQueueItem = { + ...followUp, + turn: { ...followUp.turn, traceparent: `00-${traceB}-${parentB}-01` } + }; + expect(first.missionId).toBe(second.missionId); + expect(first.turn.agentRunId).not.toBe(second.turn.agentRunId); + const input = roomAgentSupervisorInputSchema.parse({ roomId: fixture.room.id, pending: [first, second] }); + const recorded: ServerSpan[] = []; + const spans: SpanExporter = { + record: span => { recorded.push(span); }, + flush: async () => undefined, + close: async () => undefined, + stats: () => ({ exported: recorded.length, dropped: 0, failed: 0 }) + }; + const finishedRuns: string[] = []; + const executedRuns: string[] = []; + const environment = await TestWorkflowEnvironment.createLocal(); + const taskQueue = `otlp-supervisor-${randomUUID()}`; + try { + const worker = await Worker.create({ + connection: environment.nativeConnection, taskQueue, workflowsPath, + activities: { + checkRoomAgentSupervisorTurn: async () => "READY", + runCodexAppServerTurn: async (turn: DummyTurnInput) => { + executedRuns.push(turn.agentRunId!); + return { + status: "COMPLETED", message: "Synthetic turn completed", + roomAgentOutcome: { status: "VERIFIED", executedActionCount: 1, statusReason: "Synthetic verification" } + }; + }, + finishRoomAgentSupervisorTurn: async ({ item }: { item: RoomAgentSupervisorQueueItem }) => { + finishedRuns.push(item.turn.agentRunId!); + } + }, + sinks: createWorkflowTraceSinks(spans), + interceptors: { workflowModules, activity: [createActivityObservabilityInterceptors({ spans })] } + }); + await worker.runUntil(async () => { + const handle = await environment.client.workflow.start("roomAgentSupervisorWorkflow", { + workflowId: taskQueue, taskQueue, args: [input], + memo: { traceparent: first.turn.traceparent } + }); + try { + const segments = () => recorded.filter(span => span.name === "workflow roomAgentSupervisorWorkflow" && span.attributes["space.agent_run_id"]); + await expect.poll(() => segments().length, { timeout: 30_000 }).toBe(2); + expect(executedRuns).toEqual([first.turn.agentRunId, second.turn.agentRunId]); + // A deliberately has a same-mission successor, so production code + // skips its finish activity. Its segment must still be emitted. + expect(finishedRuns).toEqual([second.turn.agentRunId]); + expect((await handle.describe()).status.name).toBe("RUNNING"); + for (const [item, traceId, parentSpanId] of [[first, traceA, parentA], [second, traceB, parentB]] as const) { + const segment = segments().filter(span => span.attributes["space.agent_run_id"] === item.turn.agentRunId); + expect(segment).toHaveLength(1); + expect(segment[0]).toMatchObject({ traceId, parentSpanId }); + expect(recorded.filter(span => span.spanId === segment[0]!.spanId)).toHaveLength(1); + const children = recorded.filter(span => span.name === "workflow codexAppServerTurnWorkflow" && span.traceId === traceId); + expect(children).toHaveLength(1); + expect(children[0]!.parentSpanId).toBe(segment[0]!.spanId); + const turnActivities = recorded.filter(span => span.name === "activity runCodexAppServerTurn" && span.traceId === traceId); + expect(turnActivities).toHaveLength(1); + expect(turnActivities[0]!.parentSpanId).toBe(children[0]!.spanId); + expect(turnActivities[0]!.attributes["space.attempt"]).toBe(1); + const checks = recorded.filter(span => span.name === "activity checkRoomAgentSupervisorTurn" && span.traceId === traceId); + expect(checks).toHaveLength(1); + expect(checks[0]!.parentSpanId).toBe(segment[0]!.spanId); + } + const finishActivities = recorded.filter(span => span.name === "activity finishRoomAgentSupervisorTurn"); + expect(finishActivities).toHaveLength(1); + expect(finishActivities[0]!.traceId).toBe(traceB); + const secondSegment = segments().find(span => span.traceId === traceB)!; + expect(finishActivities[0]!.parentSpanId).toBe(secondSegment.spanId); + expect(recorded.some(span => Object.keys(span.attributes).some(key => /prompt|message|path|url/i.test(key)))).toBe(false); + } finally { + await handle.cancel().catch(() => undefined); + await handle.result().catch(() => undefined); + } + }); + } finally { + await spans.close(); + await environment.teardown(); + } + }, 120_000); +}); diff --git a/apps/worker/tests/otlp-temporal.integration.test.ts b/apps/worker/tests/otlp-temporal.integration.test.ts new file mode 100644 index 000000000..3c092a996 --- /dev/null +++ b/apps/worker/tests/otlp-temporal.integration.test.ts @@ -0,0 +1,129 @@ +import { createServer } from "node:http"; +import { fileURLToPath } from "node:url"; +import { once } from "node:events"; +import Fastify from "fastify"; +import { Context } from "@temporalio/activity"; +import { Worker } from "@temporalio/worker"; +import { TestWorkflowEnvironment } from "@temporalio/testing"; +import { describe, expect, it } from "vitest"; +import { createOtlpSpanExporter } from "@space/runtime"; +import { createHttpObservability } from "../../api/src/observability.js"; +import { traceContextForRawRequest } from "../../api/src/trace-context.js"; +import { createActivityObservabilityInterceptors } from "../src/activity-observability.js"; +import { createWorkflowTraceSinks } from "../src/workflow-tracing-sink.js"; + +const workflowsPath = fileURLToPath(new URL("./fixtures/otlp-workflows.ts", import.meta.url)); +const workflowModules = [fileURLToPath(new URL("../src/workflow-tracing.ts", import.meta.url))]; +const traceId = "4bf92f3577b34da6a3ce929d0e0e4736"; +const parentId = "00f067aa0ba902b7"; +const traceparent = `00-${traceId}-${parentId}-01`; + +interface CollectedSpan { + traceId: string; spanId: string; parentSpanId?: string; name: string; + attributes: { key: string; value: { stringValue?: string; intValue?: string } }[]; + links?: { traceId: string; spanId: string }[]; +} + +describe("real Temporal OTLP trace structure", () => { + it("connects an HTTP SERVER span, workflow, retries and child; replay emits no spans", async () => { + const collected: CollectedSpan[] = []; + const collector = createServer(async (request, response) => { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const body = JSON.parse(Buffer.concat(chunks).toString()); + for (const resource of body.resourceSpans) for (const scope of resource.scopeSpans) collected.push(...scope.spans); + // Optional local Jaeger proof, using only this test's synthetic data. + const jaegerEndpoint = process.env.SPACE_TEMPORAL_TEST_JAEGER; + if (jaegerEndpoint) { + const target = new URL(jaegerEndpoint); + if (!['127.0.0.1', 'localhost', '[::1]'].includes(target.hostname)) throw new Error("Test collector forwarding must be loopback"); + const forwarded = await fetch(new URL("/v1/traces", target), { + method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), signal: AbortSignal.timeout(3000) + }); + if (!forwarded.ok) throw new Error("Local Jaeger rejected test traces"); + } + response.writeHead(200, { "content-type": "application/json" }); + response.end("{}"); + }); + collector.listen(0, "127.0.0.1"); + await once(collector, "listening"); + const port = (collector.address() as { port: number }).port; + const exporter = createOtlpSpanExporter({ endpoint: `http://127.0.0.1:${port}`, serviceName: "space-test", batchSize: 1 }); + // The collector uses wall time, so keep the local Temporal server on the + // same clock. Time skipping would make trace durations misleading. + const environment = await TestWorkflowEnvironment.createLocal().catch(async (error: unknown) => { + await exporter.close(); + await new Promise(resolve => collector.close(() => resolve())); + throw error; + }); + const workerOptions = { + connection: environment.nativeConnection, taskQueue: "otlp-test", workflowsPath, + activities: { + tracedRetry: async () => { + if (Context.current().info.attempt === 1) throw new Error("PRIVATE-CONTROL error message"); + return "ok"; + }, + recordDummyTurnStarted: async () => undefined, + recordDummyTurnCompleted: async () => ({ status: "COMPLETED" }) + }, + sinks: createWorkflowTraceSinks(exporter), + interceptors: { workflowModules, activity: [createActivityObservabilityInterceptors({ spans: exporter })] } + }; + const observability = createHttpObservability({ serviceName: "space-api", spanExporter: exporter }); + const app = Fastify(); + app.addHook("onRequest", async request => observability.onRequest(request)); + app.addHook("onResponse", async (request, reply) => observability.onResponse(request, reply)); + app.get("/trace-test", async request => { + const context = traceContextForRawRequest(request.raw); + return environment.client.workflow.execute("tracedRoot", { + workflowId: "otlp-root", taskQueue: "otlp-test", + args: [{ traceparent: context.traceparent, prompt: "PRIVATE-CONTROL prompt" }] + }); + }); + try { + const worker = await Worker.create(workerOptions); + await worker.runUntil(async () => { + const result = await app.inject({ method: "GET", url: "/trace-test", headers: { traceparent } }); + expect(result.statusCode).toBe(200); + expect(result.body).toBe("complete"); + }); + await exporter.flush(); + const server = collected.find(span => span.name === "GET /trace-test")!; + const root = collected.find(span => span.name === "workflow tracedRoot")!; + const child = collected.find(span => span.name === "workflow dummyTurnWorkflow")!; + expect(server).toMatchObject({ traceId, parentSpanId: parentId }); + expect(root).toMatchObject({ traceId, parentSpanId: server.spanId }); + expect(child).toMatchObject({ traceId, parentSpanId: root.spanId }); + expect(collected.filter(span => span.name.startsWith("workflow "))).toHaveLength(2); + const retries = collected.filter(span => span.name === "activity tracedRetry"); + expect(retries).toHaveLength(2); + expect(retries.map(span => span.attributes.find(a => a.key === "space.attempt")?.value.intValue).sort()).toEqual(["1", "2"]); + expect(retries.every(span => span.traceId === traceId && span.parentSpanId === root.spanId)).toBe(true); + const childActivities = collected.filter(span => span.name.startsWith("activity recordDummyTurn")); + expect(childActivities).toHaveLength(2); + expect(childActivities.every(span => span.parentSpanId === child.spanId)).toBe(true); + expect(JSON.stringify(collected)).not.toContain("PRIVATE-CONTROL"); + const beforeReplay = collected.length; + const history = await environment.client.workflow.getHandle("otlp-root").fetchHistory(); + await Worker.runReplayHistory({ workflowsPath, interceptors: { workflowModules }, sinks: createWorkflowTraceSinks(exporter) }, history); + await exporter.flush(); + expect(collected).toHaveLength(beforeReplay); + + const continuationWorker = await Worker.create(workerOptions); + await continuationWorker.runUntil(() => environment.client.workflow.execute("tracedContinuation", { + workflowId: "otlp-continuation", taskQueue: "otlp-test", args: [{ traceparent, generation: 0 }] + })); + await exporter.flush(); + const runs = collected.filter(span => span.name === "workflow tracedContinuation"); + expect(runs).toHaveLength(2); + expect(runs[0]!.spanId).not.toBe(runs[1]!.spanId); + expect(runs[1]!.links).toContainEqual({ traceId, spanId: runs[0]!.spanId }); + } finally { + await app.close(); + observability.close(); + await exporter.close(); + await environment.teardown(); + await new Promise(resolve => collector.close(() => resolve())); + } + }, 120_000); +}); diff --git a/apps/worker/tests/room-background-continuation.test.ts b/apps/worker/tests/room-background-continuation.test.ts index 127ccb649..85c414f50 100644 --- a/apps/worker/tests/room-background-continuation.test.ts +++ b/apps/worker/tests/room-background-continuation.test.ts @@ -6,6 +6,17 @@ import { continuationFixture } from "./room-continuation-fixture.js"; import { runCodexAppServerTurn } from "../src/activities.js"; describe("durable background continuation", () => { + it("persists an explicit link to the source trace across a background continuation retry", async () => { + const traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + const c = continuationFixture("OPEN_TYPES", traceparent); + const prepared = await prepareBackgroundContinuation(c.input, c.store); + if (prepared.state !== "READY") throw Error("Missing continuation"); + expect(prepared.item.turn.traceparent).toBe(traceparent); + expect(prepared.item.turn.traceLink).toBe(traceparent); + expect(prepared.item.turn.traceId).not.toBe(c.initial.queueItem.turn.traceId); + expect(await prepareBackgroundContinuation(c.input, c.store)).toEqual(prepared); + }); + it("includes the actual failure reason when resuming after a parsed background action", () => { const c = continuationFixture(); const evidence = JSON.parse(continuationEvidence([{ ...c.action, status: "FAILED", diff --git a/apps/worker/tests/room-continuation-fixture.ts b/apps/worker/tests/room-continuation-fixture.ts index 340e964e0..d8f90f252 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,8 @@ 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"] + traceId: `trace:continuation-${suffix}`, selectedToolIds: ["room:inspect", "room:open_types", "room:send"], + ...(traceparent ? { traceparent } : {}) } }); 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..0ab790a7a --- /dev/null +++ b/apps/worker/tests/workflow-tracing.test.ts @@ -0,0 +1,205 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { WorkflowSpanRecord } from "../src/workflow-tracing-context.js"; +import { + parseWorkflowTraceparent, traceArgument, traceFromHeaders, traceHeader, + workflowCorrelation, workflowSpanId +} from "../src/workflow-tracing-context.js"; +import { createWorkflowTraceSinks } from "../src/workflow-tracing-sink.js"; + +const state = vi.hoisted(() => ({ + info: { workflowId: "workflow:1", runId: "run:1", workflowType: "testWorkflow", memo: {} as Record }, + records: [] as WorkflowSpanRecord[] +})); +vi.mock("@temporalio/workflow", () => ({ + ContinueAsNew: class ContinueAsNew extends Error {}, + isCancellation: (error: unknown) => error instanceof Error && error.name === "CancelledFailure", + proxySinks: () => ({ spaceTracing: { record: (span: WorkflowSpanRecord) => state.records.push(span) } }), + workflowInfo: () => state.info +})); +import { ContinueAsNew } from "@temporalio/workflow"; +import { interceptors } from "../src/workflow-tracing.js"; + +const traceId = "4bf92f3577b34da6a3ce929d0e0e4736"; +const parentId = "00f067aa0ba902b7"; +const traceparent = `00-${traceId}-${parentId}-01`; +const otherTraceparent = "00-11111111111111111111111111111111-2222222222222222-01"; +beforeEach(() => { state.records = []; state.info.memo = {}; state.info.workflowType = "testWorkflow"; }); + +function activity(args: unknown[] = []) { + return { activityType: "testActivity", args, headers: {}, seq: 1, options: { startToCloseTimeout: "1 minute" } } as never; +} + +describe("workflow trace context", () => { + it("roundtrips valid W3C headers and rejects invalid/oversized/zero identifiers", () => { + expect(traceFromHeaders({ traceparent: traceHeader(traceparent) })).toEqual({ traceId, spanId: parentId, flags: "01" }); + expect(traceFromHeaders({ traceparent: { data: new Uint8Array(257) } })).toBeNull(); + expect(traceFromHeaders({ traceparent: { data: Uint8Array.of(1, 2) } })).toBeNull(); + expect(parseWorkflowTraceparent("00-00000000000000000000000000000000-00f067aa0ba902b7-01")).toBeNull(); + expect(parseWorkflowTraceparent("00-4bf92f3577b34da6a3ce929d0e0e4736-0000000000000000-01")).toBeNull(); + expect(parseWorkflowTraceparent("secret prompt")).toBeNull(); + }); + + it("reads only bounded known argument containers and identifier fields", () => { + const args = [{ item: { missionId: "mission:1", turn: { traceparent, prompt: "private text", roomId: "room:1" } } }]; + expect(traceArgument(args)).toMatchObject({ traceId }); + expect(workflowCorrelation(args)).toEqual({ missionId: "mission:1", roomId: "room:1" }); + expect(traceArgument([{ arbitrary: { traceparent } }])).toBeNull(); + expect(workflowCorrelation([{ roomId: "https://private.example/secret", prompt: "private" }])).toEqual({}); + }); + + it("derives stable nonzero span ids without consuming application randomness", () => { + const random = vi.spyOn(Math, "random"); + expect(workflowSpanId("run:1")).toMatch(/^[0-9a-f]{16}$/); + expect(workflowSpanId("run:1")).toBe(workflowSpanId("run:1")); + expect(workflowSpanId("run:1")).not.toBe(workflowSpanId("run:2")); + expect(random).not.toHaveBeenCalled(); + random.mockRestore(); + }); +}); + +describe("workflow interceptor", () => { + it("parents activities and child workflows beneath the actual workflow span", async () => { + const hooks = interceptors(); + const outbound = hooks.outbound![0]!; + const activityNext = vi.fn(async () => "activity result"); + const childNext = vi.fn(async () => [Promise.resolve("child-run"), Promise.resolve("child result")] as [Promise, Promise]); + await hooks.inbound![0]!.execute!({ args: [{ traceparent }], headers: {} }, async () => { + await outbound.scheduleActivity!(activity([{ traceparent }]), activityNext); + await outbound.startChildWorkflowExecution!({ + workflowType: "child", options: { args: [{ traceparent }] }, headers: {}, seq: 2 + } as never, childNext); + return "done"; + }); + const root = state.records.at(-1)!; + expect(state.records).toHaveLength(1); + expect(root).toMatchObject({ traceId, parentSpanId: parentId, outcome: "COMPLETED" }); + expect(traceFromHeaders(activityNext.mock.calls[0]![0].headers)).toMatchObject({ traceId, spanId: root.spanId }); + expect(traceFromHeaders(childNext.mock.calls[0]![0].headers)).toMatchObject({ traceId, spanId: root.spanId }); + }); + + it("prefers a propagated workflow header over an ancestor traceparent left in business arguments", async () => { + const hooks = interceptors(); + const childParent = `00-${traceId}-aaaaaaaaaaaaaaaa-01`; + const next = vi.fn(async () => "ok"); + await hooks.inbound![0]!.execute!({ args: [{ traceparent }], headers: { traceparent: traceHeader(childParent) } }, async () => { + return hooks.outbound![0]!.scheduleActivity!(activity([{ traceparent }]), next); + }); + const root = state.records.at(-1)!; + expect(root.parentSpanId).toBe("aaaaaaaaaaaaaaaa"); + expect(traceFromHeaders(next.mock.calls[0]![0].headers)?.spanId).toBe(root.spanId); + }); + + it("uses memo when arguments and headers have no context", async () => { + state.info.memo = { traceparent }; + const hooks = interceptors(); + await hooks.inbound![0]!.execute!({ args: [], headers: {} }, async () => "ok"); + expect(state.records[0]).toMatchObject({ traceId, parentSpanId: parentId }); + }); + + it("keeps later supervisor mission activities in their own API trace", async () => { + state.info.workflowType = "roomAgentSupervisorWorkflow"; + const hooks = interceptors(); + const next = vi.fn(async () => "ok"); + await hooks.inbound![0]!.execute!({ args: [{ traceparent }], headers: {} }, async () => { + await hooks.inbound![0]!.handleSignal!({ signalName: "enqueue", headers: {}, args: [{ missionId: "mission:2", turn: { traceparent: otherTraceparent } }] }, async () => undefined); + await hooks.outbound![0]!.scheduleActivity!(activity([{ missionId: "mission:2" }]), next); + }); + expect(traceFromHeaders(next.mock.calls[0]![0].headers)?.traceId).toBe("11111111111111111111111111111111"); + expect(state.records.some(span => span.traceId === "11111111111111111111111111111111" && span.parentSpanId === "2222222222222222")).toBe(true); + }); + + it.each(["finishRoomAgentSupervisorTurn", "markRoomAgentMissionFinished"])("exports each supervisor mission once after %s", async (activityType) => { + state.info.workflowType = "roomAgentSupervisorWorkflow"; + const hooks = interceptors(); + const next = vi.fn(async () => "ok"); + await hooks.inbound![0]!.execute!({ args: [{ traceparent }], headers: {} }, async () => { + await hooks.inbound![0]!.handleSignal!({ signalName: "enqueue", headers: {}, args: [{ missionId: "mission:2", turn: { traceparent: otherTraceparent } }] }, async () => undefined); + await hooks.outbound![0]!.scheduleActivity!(activity([{ missionId: "mission:2" }]), next); + expect(state.records).toHaveLength(0); + await hooks.outbound![0]!.scheduleActivity!({ + activityType, args: [{ item: { missionId: "mission:2" }, status: "COMPLETED" }], headers: {}, seq: 2, options: { startToCloseTimeout: "1 minute" } + } as never, next); + expect(state.records).toHaveLength(1); + }); + expect(state.records).toHaveLength(2); // One mission plus the workflow run, each once. + expect(new Set(state.records.map(span => span.spanId)).size).toBe(2); + }); + + it("exports a completed supervisor turn before an indefinite wait, after active bookkeeping settles", async () => { + state.info.workflowType = "roomAgentSupervisorWorkflow"; + const hooks = interceptors(); + let resolveBookkeeping!: () => void; + const bookkeeping = new Promise(resolve => { resolveBookkeeping = resolve; }); + await hooks.inbound![0]!.execute!({ args: [{ traceparent }], headers: {} }, async () => { + const child = await hooks.outbound![0]!.startChildWorkflowExecution!({ + workflowType: "child", options: { args: [{ missionId: "mission:2", traceparent: otherTraceparent }] }, headers: {}, seq: 2 + } as never, async () => [Promise.resolve("child-run"), Promise.resolve("done")]); + await child[1]; + const running = hooks.outbound![0]!.scheduleActivity!(activity([{ missionId: "mission:2" }]), async () => bookkeeping); + await Promise.resolve(); + hooks.internals![0]!.concludeActivation!({ commands: [] }, value => value); + expect(state.records).toHaveLength(0); + resolveBookkeeping(); + await running; + hooks.internals![0]!.concludeActivation!({ commands: [] }, value => value); + expect(state.records).toHaveLength(1); + hooks.internals![0]!.concludeActivation!({ commands: [] }, value => value); + expect(state.records).toHaveLength(1); + }); + expect(state.records).toHaveLength(2); + }); + + it("carries a link to the previous run across continue-as-new", async () => { + const hooks = interceptors(); + const continued = new ContinueAsNew({} as never); + const next = vi.fn(async () => { throw continued; }); + await expect(hooks.inbound![0]!.execute!({ args: [{ traceparent }], headers: {} }, async () => { + return hooks.outbound![0]!.continueAsNew!({ args: [], headers: {}, options: { workflowType: "testWorkflow" } }, next); + })).rejects.toBe(continued); + const root = state.records.at(-1)!; + expect(root.outcome).toBe("CONTINUED"); + const headers = next.mock.calls[0]![0].headers; + expect(traceFromHeaders(headers)).toMatchObject({ traceId, spanId: parentId }); + expect(traceFromHeaders(headers, "space-trace-link")).toMatchObject({ traceId, spanId: root.spanId }); + const resumed = interceptors(); + await resumed.inbound![0]!.execute!({ args: [], headers }, async () => "ok"); + expect(state.records.at(-1)!.links).toEqual([{ traceId, spanId: root.spanId }]); + }); + + it("exports generic failure outcome without error messages or argument content", async () => { + const hooks = interceptors(); + await expect(hooks.inbound![0]!.execute!({ args: [{ traceparent, prompt: "private prompt" }], headers: {} }, async () => { + throw new Error("private exception text"); + })).rejects.toThrow("private exception text"); + expect(state.records.at(-1)!.outcome).toBe("FAILED"); + expect(JSON.stringify(state.records)).not.toMatch(/private/); + }); +}); + +describe("workflow sink", () => { + it("is disabled without an exporter and explicitly suppresses replay calls", () => { + expect(createWorkflowTraceSinks(null)).toBeUndefined(); + const record = vi.fn(); + const sinks = createWorkflowTraceSinks({ record } as never)!; + expect(sinks.spaceTracing.record.callDuringReplay).toBe(false); + }); + + it("reconstructs only allowlisted attributes and isolates exporter failures", () => { + const record = vi.fn(); + const sinks = createWorkflowTraceSinks({ record } as never)!; + const sample = { + traceId, spanId: "aaaaaaaaaaaaaaaa", parentSpanId: parentId, + startedAtMs: 1000, endedAtMs: 1100, outcome: "COMPLETED" as const, + correlation: { roomId: "room:1", prompt: "private", url: "https://private", path: "/secret" }, + unexpected: "private" + }; + sinks.spaceTracing.record.fn(state.info as never, sample); + expect(record.mock.calls[0]![0].attributes).toEqual({ + "space.workflow_type": "testWorkflow", "space.outcome": "COMPLETED", + "space.workflow_id": "workflow:1", "space.run_id": "run:1", "space.room_id": "room:1" + }); + expect(JSON.stringify(record.mock.calls, (_key, value) => typeof value === "bigint" ? String(value) : value)).not.toMatch(/private|secret/); + record.mockImplementation(() => { throw new Error("collector unavailable"); }); + expect(() => sinks.spaceTracing.record.fn(state.info as never, sample)).not.toThrow(); + }); +}); diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index e432af4c6..3e7fa7854 100644 --- a/docs/configuration-reference.md +++ b/docs/configuration-reference.md @@ -39,6 +39,26 @@ capabilities or cross-service compatibility. Authentication/server bootstrap, worker, web build and installer environment fields outside `SpaceApiConfig` are outside this registry. Full cross-service configuration coverage remains open. +## Optional OTLP trace export + +`SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is a collector base URL, for example +`http://127.0.0.1:4318`. Configure it separately in the API, the main worker, +the memory-maintenance worker and the memory-mutation worker. Each enabled +process sends OTLP/HTTP JSON to `/v1/traces`; omit the setting to +disable export. Only plain HTTP(S) URLs without credentials, query strings or +fragments are accepted. No OpenTelemetry SDK or tracing provider account is +required. + +Coverage includes API SERVER spans, Temporal workflow executions and individual +activity attempts. Child workflows inherit their parent's W3C context; +continue-as-new and background turns carry causal span links. The exporter uses +a bounded queue and per-request timeout, and a collector outage does not fail +application work. Exported attributes are identifiers, types, statuses and +stable failure codes/categories, never prompts, output, paths, URLs or error +messages. CLI Host and Browser Host spans remain outside this coverage. See +[observability](quality/observability.md#traces-w3c-trace-context) for replay and +deployment validation limits. + ## Optional modules `SPACE_OPTIONAL_MODULES` turns on modules beyond the coding workspace: diff --git a/docs/quality/observability.md b/docs/quality/observability.md index d5cd7080e..b33ccc05d 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -215,46 +215,44 @@ Without the endpoint, every activity still writes `activity.started` and ## Traces (W3C trace context) -The trace context is local by default. An optional OTLP/HTTP (JSON) exporter sends one -SERVER span per API request when `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` is set (no -OpenTelemetry dependency; `packages/runtime/src/otlp-span-exporter.ts`). Since -2026-10-16 the worker and both memory workers use the same exporter (services -`space-worker`, `space-memory-maintenance-worker`, `space-memory-mutation-worker`): -one INTERNAL span per activity whose input carries a `traceparent`, as a child of -that span, with the activity type, outcome and attempt only. +The trace context is local by default. Set `SPACE_OTEL_EXPORTER_OTLP_ENDPOINT` +to an HTTP(S) collector base URL in the API and each worker process to enable +OTLP/HTTP JSON export to `/v1/traces`. The shared, dependency-free +exporter is `packages/runtime/src/otlp-span-exporter.ts`; service names are +`space-api`, `space-worker`, `space-memory-maintenance-worker` and +`space-memory-mutation-worker`. With the setting absent, no collector requests +are made. Each exporter has a bounded queue, drop/failure counters and a POST +timeout; collector failures do not fail application work. - **API:** a valid inbound `traceparent` is continued (same trace ID, new span). An invalid or missing header gets a new trace ID. Every API log line carries `w3cTraceId` (Fastify `childLoggerFactory`). -- **Propagation into turns:** API turn starters and the Room Agent coordinator add - the request's `traceparent` to the turn input (an optional contract field) and - to the Temporal memo. A bounded in-memory map from request ID to trace context +- **Propagation into workflows:** Turn, Room Agent action/supervisor, Dot Goal + and memory starters carry the actual API SERVER span's `traceparent` in + workflow input and/or Temporal memo. They do not create an unexported + intermediate parent. A bounded in-memory map from request ID to trace context (`apps/api/src/trace-context.ts`) does this without passing the request object through every service. -- **Worker:** worker logs read the trace ID from the activity input. Internal - room-action calls from the worker send a child `traceparent`, so the internal - API request joins the same trace. -- **Workflow code is unchanged.** Correlation is done in the API starter - (input/memo) and in an activity interceptor. The workflow bundle only gains the - optional contract field. +- **Worker:** Workflow spans parent activity-attempt spans and child workflows. + The activity span includes its attempt, outcome and stable failure + code/category. Supervisor missions carry the request context of the queued + turn, including missions signalled to an already running supervisor. Internal + room-action calls retain W3C context. +- **Continuations:** Continue-as-new runs link to the previous workflow run. + Background turns retain their originating W3C context and an explicit + `traceLink` while keeping their own `trace:bg-` Space correlation id. +- **Export boundary:** The workflow sandbox does no collector I/O. Workflow + interceptors pass identifier-only spans through a sink configured with + `callDuringReplay: false`; activity spans are recorded in the worker process. + Trace attributes exclude prompts, model output, paths, URLs and error messages. ### 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 optional contract fields accept old inputs without W3C context. Tracing +must not change activity scheduling or application outcomes. Replay suppresses +workflow sink calls so history replay does not export duplicate workflow spans. +Before deploying to an existing Temporal namespace, run its normal history +replay gate as well as the tracing tests; synthetic test histories do not +certify compatibility with every deployed workflow history. ## Redaction policy @@ -280,6 +278,31 @@ No credential or transcript export was added. ## Tests +`apps/worker/tests/otlp-temporal.integration.test.ts` runs a real Temporal +server with a local HTTP collector and stub activities. It checks the HTTP +SERVER/workflow/child/activity tree, activity retries, continue-as-new links, +and that replay exports no spans. `otlp-supervisor.integration.test.ts` runs the +production supervisor with two different API traces in the same mission and +checks both request segments while it remains idle. Both run by default: + +```sh +ELECTRON_SKIP_BINARY_DOWNLOAD=1 npm ci +# Build internal workspace packages as described in CONTRIBUTING.md, then: +npm test -w @space/worker +``` + +The worker devDependency `@temporalio/testing` is pinned to 1.24.0 in the +manifest and lockfile, matching the Temporal SDK. Its first run may download +Temporal's ephemeral test server. No Docker, database, provider credentials, +or AI calls are needed. The local server uses wall time so workflow and +collector durations use the same clock. + +Worker span attributes use `space.snake_case` consistently, including +`space.workflow_id`, `space.run_id`, and `space.reason_code`. This naming +applies to exported OTLP attributes; existing log and workflow input fields +retain their original names. A Jaeger capture of the synthetic integration +is provided in the PR description rather than committed as a repository asset. + | File | Covers | | --- | --- | | `packages/contracts/tests/observability.test.ts` | Categories, `traceparent` parsing, additive contract fields | @@ -316,7 +339,8 @@ No credential or transcript export was added. continuation from the run it continues (`roomAgentContinuation.sourceRunId`, indexed by migration 0206, at most 50 generations and the row limit) and marks it with `continuationOf`. Its events and logs remain under its own - trace ID. Chat-pane sends from room actions use `:chat`. + Space trace ID; its W3C context and OTLP span link preserve the originating + trace. Chat-pane sends from room actions use `:chat`. - **CLI sessions are linked to their request (since 2026-10-27, migration 0258).** `pane_cli_sessions.trace_id` keeps the request that allocated the session (a rejoin of a running session does not take it over), and the trace lookup returns @@ -341,8 +365,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 96d7fd636..fab1ad7c4 100644 --- a/package-lock.json +++ b/package-lock.json @@ -192,6 +192,7 @@ "undici": "8.12.3" }, "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", diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index 7fb3c217a..af1bdfe28 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -6922,6 +6922,9 @@ 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(), + // A background continuation can start a new W3C trace while linking back to + // its durable predecessor. Identifiers only; never part of a workflow id. + 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(), @@ -7704,7 +7707,8 @@ export const memoryConsolidationDetailSchema = z export const memoryConsolidationWorkflowInputSchema = z .object({ runId: idSchema, - traceId: idSchema + traceId: idSchema, + traceparent: traceparentSchema.optional() }) .strict(); @@ -7916,7 +7920,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 +7993,8 @@ export const memoryMutationRecoveryResultSchema = z export const memoryMutationWorkflowInputSchema = z .object({ changeSetId: idSchema, - traceId: idSchema + traceId: idSchema, + traceparent: traceparentSchema.optional() }) .strict(); diff --git a/packages/contracts/tests/observability.test.ts b/packages/contracts/tests/observability.test.ts index c7717396d..6453d1d93 100644 --- a/packages/contracts/tests/observability.test.ts +++ b/packages/contracts/tests/observability.test.ts @@ -6,6 +6,9 @@ import { fileSearchResponseSchema, formatTraceparent, isSafeErrorCode, + memoryConsolidationWorkflowInputSchema, + memoryMaintenanceInputSchema, + memoryMutationWorkflowInputSchema, parseTraceparent, traceIdFromTraceparent, workerReadinessSchema @@ -90,6 +93,22 @@ describe("additive observability contract fields", () => { expect(dummyTurnInputSchema.parse({ ...turn, traceparent }).traceparent).toBe(traceparent); expect(dummyTurnInputSchema.parse(turn).traceparent).toBeUndefined(); expect(dummyTurnInputSchema.safeParse({ ...turn, traceparent: "not-a-trace" }).success).toBe(false); + expect(dummyTurnInputSchema.parse({ ...turn, traceLink: traceparent }).traceLink).toBe(traceparent); + expect(dummyTurnInputSchema.safeParse({ ...turn, traceLink: "private prompt" }).success).toBe(false); + }); + + it("keeps optional W3C context on memory workflows while accepting their previous inputs", () => { + const traceparent = formatTraceparent("4bf92f3577b34da6a3ce929d0e0e4736", "00f067aa0ba902b7"); + const cases = [ + [memoryConsolidationWorkflowInputSchema, { runId: "memory_consolidation:one", traceId: "trace:one" }], + [memoryMaintenanceInputSchema, { scheduledAt: "2026-10-26T00:00:00.000Z", traceId: "trace:one" }], + [memoryMutationWorkflowInputSchema, { changeSetId: "memory_change:one", traceId: "trace:one" }] + ] as const; + for (const [schema, input] of cases) { + expect(schema.parse(input).traceparent).toBeUndefined(); + expect(schema.parse({ ...input, traceparent }).traceparent).toBe(traceparent); + expect(schema.safeParse({ ...input, traceparent: "not-a-trace" }).success).toBe(false); + } }); it("accepts the old and the new file search response shapes", () => { diff --git a/packages/runtime/src/otlp-span-exporter.ts b/packages/runtime/src/otlp-span-exporter.ts index 86a0edf05..6c69b7481 100644 --- a/packages/runtime/src/otlp-span-exporter.ts +++ b/packages/runtime/src/otlp-span-exporter.ts @@ -22,6 +22,8 @@ export interface ServerSpan { error: boolean; /** OTLP span kind; SERVER (2) unless given. Worker activities use INTERNAL (1). */ kind?: number; + /** Causal predecessors when a continuation starts a separate execution. */ + links?: readonly { traceId: string; spanId: string }[]; } export interface SpanExporterStats { @@ -85,6 +87,12 @@ export function otlpTraceRequest(spans: readonly ServerSpan[], serviceName: stri startTimeUnixNano: span.startTimeUnixNano.toString(), endTimeUnixNano: span.endTimeUnixNano.toString(), attributes: Object.entries(span.attributes).map(([key, value]) => ({ key, value: attributeValue(value) })), + ...(span.links?.length ? { + links: span.links + .filter(link => traceIdPattern.test(link.traceId) && !/^0+$/.test(link.traceId) + && spanIdPattern.test(link.spanId) && !/^0+$/.test(link.spanId)) + .map(link => ({ traceId: link.traceId, spanId: link.spanId })) + } : {}), // STATUS_CODE_ERROR or STATUS_CODE_UNSET status: { code: span.error ? 2 : 0 } })) @@ -122,13 +130,16 @@ export function createOtlpSpanExporter(options: OtlpSpanExporterOptions): SpanEx }; const flush = (): Promise => { if (!inFlight) { - inFlight = (async () => { + // Defer the drain until after inFlight is assigned. An empty synchronous + // drain would otherwise clear it before assignment and retain a resolved + // promise forever, preventing every subsequent batch from being sent. + 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..cb095fb65 --- /dev/null +++ b/packages/runtime/tests/otlp-span-exporter.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "vitest"; +import { createOtlpSpanExporter, otlpTraceRequest, type ServerSpan } from "../src/otlp-span-exporter.js"; + +function span(): ServerSpan { + return { + traceId: "11111111111111111111111111111111", spanId: "2222222222222222", parentSpanId: null, + name: "workflow test", startTimeUnixNano: 1n, endTimeUnixNano: 2n, + attributes: {}, error: false, kind: 1 + }; +} + +describe("shared OTLP continuation export", () => { + it("delivers later spans after an empty flush", async () => { + const sent: string[] = []; + const exporter = createOtlpSpanExporter({ + endpoint: "http://127.0.0.1:4318", serviceName: "test", flushIntervalMs: 60_000, + fetch: async (_url, options) => { sent.push(options.body); return { ok: true }; } + }); + try { + await exporter.flush(); + exporter.record(span()); + await exporter.flush(); + expect(sent).toHaveLength(1); + expect(exporter.stats().exported).toBe(1); + await exporter.flush(); + exporter.record({ ...span(), spanId: "3333333333333333" }); + await exporter.flush(); + expect(sent).toHaveLength(2); + expect(exporter.stats().exported).toBe(2); + } finally { await exporter.close(); } + }); + + it("serializes only valid causal link identifiers", () => { + const valid = { traceId: "33333333333333333333333333333333", spanId: "4444444444444444" }; + const request = otlpTraceRequest([{ ...span(), links: [ + { ...valid, ...{ prompt: "PRIVATE-CONTROL" } }, + { ...valid, spanId: "bad" }, + { ...valid, traceId: "0".repeat(32) }, + { ...valid, spanId: "0".repeat(16) } + ] }], "test"); + expect(request.resourceSpans[0]!.scopeSpans[0]!.spans[0]!.links).toEqual([valid]); + expect(JSON.stringify(request)).not.toContain("PRIVATE-CONTROL"); + }); +});