diff --git a/apps/api/src/dot-goals.ts b/apps/api/src/dot-goals.ts index 30dd69dee..6ab347311 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,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, traceId?: string) { return this.client(async (client) => { try { - await client.workflow.start("dotGoalWorkflow", { workflowId: dotGoalWorkflowId(goalId), taskQueue: this.options.taskQueue, - args: [{ goalId }], workflowIdReusePolicy: "REJECT_DUPLICATE" }); + const traceparent = traceparentForTraceId(traceId); + const memo = workflowCorrelationMemo({ traceId, traceparent }); + await client.workflow.start("dotGoalWorkflow", { + workflowId: dotGoalWorkflowId(goalId), + taskQueue: this.options.taskQueue, + args: [{ goalId }], + workflowIdReusePolicy: "REJECT_DUPLICATE", + ...(Object.keys(memo).length ? { memo } : {}) + }); } catch (error) { if (!(error instanceof WorkflowExecutionAlreadyStartedError)) throw error; } }); } @@ -65,6 +73,7 @@ export class TemporalDotGoalWorkflowCoordinator implements DotGoalWorkflowCoordi }); } } +/** Creates a TemporalDotGoalWorkflowCoordinator when enabled, otherwise null. */ export function createDotGoalWorkflowCoordinator(options: { enabled: boolean; address: string; namespace: string; taskQueue: string }): DotGoalWorkflowCoordinator | null { return options.enabled ? new TemporalDotGoalWorkflowCoordinator(options) : null; } @@ -93,19 +102,25 @@ export type DotGoalStepState = | { state: "STARTED" | "BUSY"; runId: string; goal: DotGoal } | { state: "PAUSED" | "WAITING" | "DONE"; goal: DotGoal }; +/** Generates a deterministic idempotency key for a specific Goal step number. */ function requestKeyFor(goalId: string, step: number): string { return `goal-${createHash("sha256").update(`${goalId}:${step}`).digest("hex").slice(0, 40)}`; } +/** Factory for DotGoalService managing Goal lifecycle, chat turns, and worker workflow steering. */ export function createDotGoalService(deps: DotGoalServiceDeps) { + /** Retrieves the active SpaceAgentSession corresponding to the Goal's session. */ async function activeSession(goal: DotGoal) { const session = await deps.store.getSpaceAgentSession(goal.sessionId); return session?.isActive && session.dotId === goal.dotId ? session : null; } + + /** Transitions Goal state to terminal or paused and persists status reason. */ async function finish(goal: DotGoal, status: DotGoal["status"], reason: string) { return (await deps.goals.transition(goal.goalId, ["ACTIVE", "PAUSED"], status, reason)) ?? goal; } + /** Validates prerequisites and creates a new Dot Goal, starting its Temporal workflow. */ async function create(owner: string, dotId: string, raw: unknown, request: FastifyRequest | null): Promise { const input = createDotGoalInputSchema.parse(raw); if (!deps.workflow) throw new DotGoalError(503, "DOT_GOALS_UNAVAILABLE", "Goals need the Space worker runtime."); @@ -125,7 +140,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?.id) await deps.workflow.start(goal.goalId, String(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."); @@ -134,6 +152,7 @@ export function createDotGoalService(deps: DotGoalServiceDeps) { return goal; } + /** Pauses, resumes, or cancels an existing Goal owned by the user. */ async function control(owner: string, dotId: string, goalId: string, raw: unknown, request: FastifyRequest | null): Promise { const { action } = dotGoalControlInputSchema.parse(raw); const goal = await deps.goals.getOwned(owner, dotId, goalId); @@ -152,6 +171,7 @@ export function createDotGoalService(deps: DotGoalServiceDeps) { return next; } + /** Appends steering guidance notes to an active Goal. */ async function steer(owner: string, dotId: string, goalId: string, raw: unknown, request: FastifyRequest | null): Promise { const { text } = dotGoalSteerInputSchema.parse(raw); const goal = await deps.goals.getOwned(owner, dotId, goalId); @@ -268,6 +288,7 @@ export function createDotGoalService(deps: DotGoalServiceDeps) { } export type DotGoalService = ReturnType; +/** Registers REST HTTP routes for managing Dot Goals (/api/dots/:dotId/goals). */ export function registerDotGoalRoutes(app: FastifyInstance, service: DotGoalService | null, deps: { goals: DotGoalsRepository } | null, options: { routeOptions: RouteShorthandOptions; authenticatedOwner: (request: FastifyRequest) => Promise; @@ -279,6 +300,7 @@ export function registerDotGoalRoutes(app: FastifyInstance, service: DotGoalServ if (error instanceof z.ZodError) return reply.code(400).send({ error: { code: "DOT_GOAL_INVALID", message: "Goal request is invalid." } }); throw error; }; + /** Authenticated handler wrapper enforcing interactive user session scope for Dot Goal operations. */ const user = (action: (request: FastifyRequest, owner: string, goals: DotGoalService, repo: DotGoalsRepository) => Promise) => async (request: FastifyRequest, reply: import("fastify").FastifyReply) => { if (!request.user) return reply.code(401).send({ error: { code: "UNAUTHENTICATED", message: "Sign in to use Dots." } }); diff --git a/apps/api/src/memory-consolidation-coordinator.ts b/apps/api/src/memory-consolidation-coordinator.ts index df237c648..f1aafe44e 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 { @@ -33,23 +35,28 @@ export type MemoryConsolidationWorkflowStarter = ( options: WorkflowStartOptions ) => Promise; +/** Checks whether the error signifies that the consolidation workflow execution has already started. */ function isAlreadyStarted(error: unknown): boolean { return error instanceof Error && error.name === "WorkflowExecutionAlreadyStartedError"; } +/** Schedules a memory consolidation workflow on Temporal with correlation memo and traceparent. */ export async function scheduleMemoryConsolidation( rawInput: MemoryConsolidationWorkflowInput, workflowId: string, startWorkflow: MemoryConsolidationWorkflowStarter ): Promise { - const input = memoryConsolidationWorkflowInputSchema.parse(rawInput); + // Rollout order: Deploy workers before the API so workers accept the optional + // traceparent field in workflow input arguments before the API starts sending it. + 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", + ...(input.traceparent ? { memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) } : {}) }); return { status: "SCHEDULED", @@ -93,6 +100,7 @@ class TemporalMemoryConsolidationCoordinator implements MemoryConsolidationCoord } } +/** Creates a live or disabled MemoryConsolidationCoordinator based on configuration flags. */ export function createMemoryConsolidationCoordinator(options: { enabled: boolean; address: string; diff --git a/apps/api/src/memory-mutation-coordinator.ts b/apps/api/src/memory-mutation-coordinator.ts index f7d3d48bb..0f5d15655 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 { @@ -34,16 +36,20 @@ export type MemoryMutationWorkflowStarter = ( options: WorkflowStartOptions ) => Promise; +/** Checks whether the error signifies that the mutation workflow execution has already started. */ function isAlreadyStarted(error: unknown): boolean { return error instanceof Error && error.name === "WorkflowExecutionAlreadyStartedError"; } +/** Schedules a memory mutation workflow on Temporal with correlation memo and traceparent. */ export async function scheduleMemoryMutation( rawInput: MemoryMutationWorkflowInput, startWorkflow: MemoryMutationWorkflowStarter, commandHash?: string ): Promise { - const input = memoryMutationWorkflowInputSchema.parse(rawInput); + // Rollout order: Deploy workers before the API so workers accept the optional + // traceparent field in workflow input arguments before the API starts sending it. + const input = withTurnTraceparent(memoryMutationWorkflowInputSchema.parse(rawInput)); const workflowId = buildMemoryMutationWorkflowId(input.changeSetId, commandHash); try { const handle = await startWorkflow(MEMORY_MUTATION_WORKFLOW_TYPE, { @@ -51,7 +57,8 @@ export async function scheduleMemoryMutation( taskQueue: MEMORY_MUTATION_TASK_QUEUE, workflowId, workflowIdReusePolicy: "REJECT_DUPLICATE", - workflowIdConflictPolicy: "USE_EXISTING" + workflowIdConflictPolicy: "USE_EXISTING", + ...(input.traceparent ? { memo: workflowCorrelationMemo({ traceId: input.traceId, traceparent: input.traceparent }) } : {}) }); return { status: "SCHEDULED", workflowId: handle.workflowId, runId: handle.firstExecutionRunId ?? null }; } catch (error) { @@ -91,6 +98,7 @@ class TemporalMemoryMutationCoordinator implements MemoryMutationCoordinator { } } +/** Creates a live or disabled MemoryMutationCoordinator based on configuration flags. */ export function createMemoryMutationCoordinator(options: { enabled: boolean; address: string; diff --git a/apps/api/src/room-agent.ts b/apps/api/src/room-agent.ts index 5eccd5097..ba60b914d 100644 --- a/apps/api/src/room-agent.ts +++ b/apps/api/src/room-agent.ts @@ -190,14 +190,15 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo const lease = await this.connect("roomAgentActionWorkflow"); try { try { + const traceparent = traceparentForTraceId(traceId); 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 }) @@ -226,7 +227,12 @@ 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], + memo: workflowCorrelationMemo({ + traceId: item.turn.traceId, + traceparent: item.turn.traceparent ?? null, + missionId: item.missionId + }) }); recordWorkflowStart(ROOM_AGENT_SUPERVISOR_WORKFLOW_TYPE, "SIGNALED"); return { workflowId, runId: handle.signaledRunId }; @@ -254,6 +260,7 @@ export class TemporalRoomAgentWorkflowCoordinator implements RoomAgentWorkflowCo } } +/** Creates a live TemporalRoomAgentWorkflowCoordinator when enabled, otherwise a DisabledRoomAgentWorkflowCoordinator. */ export function createRoomAgentWorkflowCoordinator(options: { enabled: boolean; address: string; diff --git a/apps/api/src/trace-context.ts b/apps/api/src/trace-context.ts index af9e55efd..eec7b5ca4 100644 --- a/apps/api/src/trace-context.ts +++ b/apps/api/src/trace-context.ts @@ -21,6 +21,7 @@ export interface RequestTraceContext { inbound: boolean; } +/** Generates non-zero random hex bytes of the specified length for trace/span IDs. */ function randomHex(bytes: number): string { for (;;) { const value = randomBytes(bytes).toString("hex"); @@ -28,6 +29,7 @@ function randomHex(bytes: number): string { } } +/** Parses incoming traceparent header or generates fresh trace identifiers for an HTTP request. */ export function createRequestTraceContext(header: unknown, random: (bytes: number) => string = randomHex): RequestTraceContext { const inbound = parseTraceparent(Array.isArray(header) ? header[0] : header); const w3cTraceId = inbound?.traceId ?? random(16); @@ -96,7 +98,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/api/tests/otlp-span-exporter.test.ts b/apps/api/tests/otlp-span-exporter.test.ts index d94ec6539..446202ee1 100644 --- a/apps/api/tests/otlp-span-exporter.test.ts +++ b/apps/api/tests/otlp-span-exporter.test.ts @@ -50,6 +50,15 @@ describe("OTLP span export configuration", () => { expect(otlpSpanExporterFromEnv({}, "space-api")).toBeNull(); expect(otlpSpanExporterFromEnv({ SPACE_OTEL_EXPORTER_OTLP_ENDPOINT: "javascript:alert(1)" }, "space-api")).toBeNull(); }); + + it("rejects non-positive and non-finite batchSize values", () => { + expect(() => createOtlpSpanExporter({ endpoint: "http://127.0.0.1:4318/", serviceName: "space-api", batchSize: 0 })) + .toThrow("batchSize must be a positive finite number."); + expect(() => createOtlpSpanExporter({ endpoint: "http://127.0.0.1:4318/", serviceName: "space-api", batchSize: -1 })) + .toThrow("batchSize must be a positive finite number."); + expect(() => createOtlpSpanExporter({ endpoint: "http://127.0.0.1:4318/", serviceName: "space-api", batchSize: Number.NaN })) + .toThrow("batchSize must be a positive finite number."); + }); }); describe("OTLP span exporter", () => { 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..53a666628 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 type { ActivityInterceptorsFactory, Headers } from "@temporalio/worker"; import { categorizeFailureCode, formatTraceparent, @@ -10,9 +10,14 @@ 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 { + decodeSpaceTraceHeader, + extractArgsTraceContext, + mergeTraceContexts +} from "./workflow-observability-interceptors.js"; /** * One structured start/finish record and metrics per activity execution, with @@ -55,13 +60,19 @@ export function failureReasonScope(execute: () => Promise): { promise: Pro return { promise: observations.run(observation, execute), reasonCode: () => observation.reasonCode }; } +/** Generates non-zero random hex string of the given byte length. */ +function randomNonZeroHex(bytes: number): string { + for (;;) { + const value = randomBytes(bytes).toString("hex"); + if (!/^0+$/.test(value)) return value; + } +} + /** traceparent for an outbound call: same W3C trace, new span id; null if none. */ export function childTraceparent(traceparent: string | null | undefined): string | null { const parsed = parseTraceparent(traceparent); if (!parsed) return null; - let spanId = randomBytes(8).toString("hex"); - while (/^0+$/.test(spanId)) spanId = randomBytes(8).toString("hex"); - return formatTraceparent(parsed.traceId, spanId, parsed.flags); + return formatTraceparent(parsed.traceId, randomNonZeroHex(8), parsed.flags); } // Keys whose values are identifiers. Prompts, content and messages are never read. @@ -86,6 +97,7 @@ const nestedKeys = ["turn", "item", "bridge", "input", "turnInput"] as const; type CorrelationKey = (typeof correlationKeys)[number]; +/** Traverses nested arguments up to the specified depth to collect allowlisted correlation identifiers. */ function readCorrelation(value: unknown, into: Partial>, depth: number): void { if (!value || typeof value !== "object" || Array.isArray(value)) return; const record = value as Record; @@ -115,6 +127,7 @@ export function activityCorrelation(args: readonly unknown[]): Record 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; + /** Inbound Temporal headers from the workflow outbound interceptor. */ + headers?: Headers; } +/** Extracts and parses traceparent from activity arguments if present. */ function activityTraceparent(args: readonly unknown[]) { const found: Partial> = {}; for (const arg of args) readCorrelation(arg, found, 2); return parseTraceparent(found.traceparent); } +/** Wraps activity execution with metrics, structured logging, failure reason propagation, and OTLP spans. */ export async function observeActivityExecution( info: ActivityInfo, args: readonly unknown[], @@ -158,6 +176,11 @@ export async function observeActivityExecution( const queueMs = Number.isFinite(info.currentAttemptScheduledTimestampMs) ? Math.max(0, startedAt - info.currentAttemptScheduledTimestampMs) : null; + const headerCtx = decodeSpaceTraceHeader(dependencies.headers); + const argsCtx = extractArgsTraceContext(args, info.workflowType); + const mergedCtx = mergeTraceContexts(headerCtx, argsCtx); + const argCorr = activityCorrelation(args); + const w3cTraceId = traceIdFromTraceparent(mergedCtx.traceparent) ?? argCorr.w3cTraceId; const base: Record = { activityType, activityId: info.activityId, @@ -165,7 +188,18 @@ export async function observeActivityExecution( workflowId: info.workflowExecution?.workflowId ?? null, temporalRunId: info.workflowExecution?.runId ?? null, attempt: info.attempt, - ...activityCorrelation(args) + ...(mergedCtx.traceId ? { traceId: mergedCtx.traceId } : {}), + ...(w3cTraceId ? { w3cTraceId } : {}), + ...(mergedCtx.parentTraceId ? { parentTraceId: mergedCtx.parentTraceId } : {}), + ...(mergedCtx.roomId ? { roomId: mergedCtx.roomId } : {}), + ...(mergedCtx.paneId ? { paneId: mergedCtx.paneId } : {}), + ...(mergedCtx.missionId ? { missionId: mergedCtx.missionId } : {}), + ...(mergedCtx.agentRunId ? { agentRunId: mergedCtx.agentRunId } : {}), + ...(mergedCtx.actionId ? { actionId: mergedCtx.actionId } : {}), + ...(mergedCtx.goalId ? { goalId: mergedCtx.goalId } : {}), + ...(mergedCtx.dotId ? { dotId: mergedCtx.dotId } : {}), + ...(mergedCtx.turnId ? { turnId: mergedCtx.turnId } : {}), + ...argCorr }; const labels = { activity_type: activityType }; if (info.attempt > 1) metrics.activityRetries.inc(labels); @@ -174,6 +208,7 @@ export async function observeActivityExecution( workerLog("info", "activity.started", { ...base, queueMs }); const observation: ActivityObservation = { reasonCode: null }; + /** Completes activity observation: logs finished event, records metrics, and exports activity span. */ const finish = (outcome: ActivityOutcome, reasonCode: string | null, errorName: string | null) => { const durationMs = Math.max(0, now() - startedAt); metrics.activitiesInFlight.inc(labels, -1); @@ -194,16 +229,46 @@ export async function observeActivityExecution( queueMs, rssBytes: process.memoryUsage().rss }); - const parent = dependencies.spans ? activityTraceparent(args) : null; + const parent = dependencies.spans + ? (parseTraceparent(headerCtx?.traceparent) ?? activityTraceparent(args)) + : null; if (parent) { + const rawAttributes: Record = { + "space.activity_type": activityType, + "space.outcome": outcome, + "space.attempt": info.attempt, + "space.duration_ms": durationMs, + ...(queueMs !== null ? { "space.queue_ms": queueMs } : {}), + ...(info.workflowType ? { "space.workflow_type": info.workflowType } : {}), + ...(info.workflowExecution?.workflowId ? { "space.workflow_id": info.workflowExecution.workflowId } : {}), + ...(info.workflowExecution?.runId ? { "space.run_id": info.workflowExecution.runId } : {}), + ...(mergedCtx.traceId ? { "space.trace_id": mergedCtx.traceId } : {}), + ...(mergedCtx.parentTraceId ? { "space.parent_trace_id": mergedCtx.parentTraceId } : {}), + ...(mergedCtx.roomId ? { "space.room_id": mergedCtx.roomId } : {}), + ...(mergedCtx.paneId ? { "space.pane_id": mergedCtx.paneId } : {}), + ...(mergedCtx.missionId ? { "space.mission_id": mergedCtx.missionId } : {}), + ...(mergedCtx.agentRunId ? { "space.agent_run_id": mergedCtx.agentRunId } : {}), + ...(mergedCtx.continuationOfRunId ? { "space.continuation_of_run_id": mergedCtx.continuationOfRunId } : {}), + ...(mergedCtx.continuationGeneration !== undefined ? { "space.continuation_generation": mergedCtx.continuationGeneration } : {}), + ...(mergedCtx.actionId ? { "space.action_id": mergedCtx.actionId } : {}), + ...(mergedCtx.goalId ? { "space.goal_id": mergedCtx.goalId } : {}), + ...(mergedCtx.dotId ? { "space.dot_id": mergedCtx.dotId } : {}), + ...(mergedCtx.turnId ? { "space.turn_id": mergedCtx.turnId } : {}), + ...(mergedCtx.changeSetId ? { "space.change_set_id": mergedCtx.changeSetId } : {}), + ...(mergedCtx.consolidationRunId ? { "space.consolidation_run_id": mergedCtx.consolidationRunId } : {}), + ...(reasonCode && isSafeErrorCode(reasonCode) + ? { "space.reason_code": reasonCode, "space.error_code": reasonCode } + : {}), + ...(category ? { "space.category": category } : {}) + }; dependencies.spans!.record({ traceId: parent.traceId, - spanId: randomBytes(8).toString("hex"), + spanId: randomNonZeroHex(8), parentSpanId: parent.parentId, name: `activity ${activityType}`, startTimeUnixNano: BigInt(startedAt) * 1_000_000n, endTimeUnixNano: BigInt(startedAt + durationMs) * 1_000_000n, - attributes: { "space.activity_type": activityType, "space.outcome": outcome, "space.attempt": info.attempt }, + attributes: sanitizeSpanAttributes(rawAttributes), error: outcome === "FAILED" || outcome === "FAILED_RESULT", kind: 1 }); @@ -224,13 +289,15 @@ 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..a1296deef 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, workflowObservabilityInterceptorsPath } 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: { + workflowModules: [workflowObservabilityInterceptorsPath], + activity: [createActivityObservabilityInterceptors({ spans })] + }, + 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..8b4399c7b 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, workflowObservabilityInterceptorsPath } 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: { + workflowModules: [workflowObservabilityInterceptorsPath], + activity: [createActivityObservabilityInterceptors({ spans })] + }, + 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/room-background-continuation.ts b/apps/worker/src/room-background-continuation.ts index e15c29a21..b8076aa57 100644 --- a/apps/worker/src/room-background-continuation.ts +++ b/apps/worker/src/room-background-continuation.ts @@ -1,6 +1,8 @@ import { buildCodexAppServerTurnWorkflowId, buildRoomAgentSupervisorWorkflowId, + deterministicTraceparent, + parseTraceparent, roomAgentSupervisorQueueItemSchema, spaceAgentRoomActionRequestSchema, type RoomAgentActionRecord, @@ -10,6 +12,7 @@ import { SpaceConflictError, redactMemoryText, type SpaceStore } from "@space/ru import { roomActionSummary } from "./room-action-summary.js"; import { DEFAULT_ROOM_AGENT_MAX_CONTINUATIONS, DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS, ROOM_AGENT_CONTINUATION_LIMIT, ROOM_AGENT_TOKEN_LIMIT } from "./room-agent-limits.js"; +/** Summarizes settled action results and status counts into a bounded JSON evidence string. */ export function continuationEvidence(actions: RoomAgentActionRecord[], budget: number): string { // Count the entire mission, including synchronous actions. Background roots // are only a scheduling distinction, not the record of completed work. @@ -45,6 +48,7 @@ export type BackgroundContinuationResult = interface MissionTokenUsage { usedTokens: number; countedThroughRunId: string } +/** Extracts recorded token usage and run checkpoint from mission execution state. */ function recordedTokenUsage(executionState: Record): MissionTokenUsage | null { const value = executionState.tokenUsage as Partial | undefined; return value && Number.isSafeInteger(value.usedTokens) && (value.usedTokens as number) >= 0 && typeof value.countedThroughRunId === "string" @@ -52,6 +56,7 @@ function recordedTokenUsage(executionState: Record): MissionTok : null; } +/** Prepares a continuation turn item for background execution with bounded token/continuation limits. */ export async function prepareBackgroundContinuation(input: { roomId: string; missionId: string; continuationId: string; }, store: SpaceStore, maxContinuations = DEFAULT_ROOM_AGENT_MAX_CONTINUATIONS, @@ -122,15 +127,24 @@ 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 nextTraceId = `trace:bg-${input.continuationId}`; + const sourceTraceparent = parseTraceparent(source.queueItem.turn.traceparent) ? source.queueItem.turn.traceparent : undefined; const item = roomAgentSupervisorQueueItemSchema.parse({ missionId: input.missionId, turn: { ...source.queueItem.turn, prompt, roomAgentOperatorRequest: request, - roomAgentContinuation: { sourceRunId: source.run.runId, generation }, + roomAgentContinuation: { + sourceRunId: source.run.runId, + generation, + ...(sourceTraceparent ? { sourceTraceId: source.queueItem.turn.traceId, sourceTraceparent } : {}) + }, artifactIds: [], agentRunId: runId, 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: nextTraceId, + ...(sourceTraceparent + ? { traceparent: deterministicTraceparent(nextTraceId, `${nextTraceId}:${runId}`, parseTraceparent(sourceTraceparent)?.flags ?? "01") } + : {}) } }); try { const enqueued = await store.enqueueRoomAgentMission({ @@ -149,7 +163,7 @@ maxMissionTokens: number | null = DEFAULT_ROOM_AGENT_MAX_MISSION_TOKENS): Promis } } -// Without an errorCode the turn was superseded or stopped; with one it failed and keeps that cause. +/** Marks an unstarted or superseded supervisor turn as INTERRUPTED or FAILED. */ export async function discardSupervisorTurn(item: RoomAgentSupervisorQueueItem, reason: string, store: SpaceStore, errorCode?: string) { const record = await store.getRoomAgentTurn(item.missionId, item.turn.agentRunId); if (!record || !["QUEUED", "RUNNING"].includes(record.run.status)) return; @@ -161,6 +175,7 @@ export async function discardSupervisorTurn(item: RoomAgentSupervisorQueueItem, await store.updateSpaceAgentMessage(record.run.responseMessageId, { status, content: reason }); } +/** Determines whether the supervisor queue item is ready to execute, paused, or should be skipped. */ export async function supervisorTurnState(item: RoomAgentSupervisorQueueItem, store: SpaceStore): Promise<"READY" | "PAUSED" | "SKIP"> { const mission = await store.getRoomAgentMission(item.turn.roomId, item.missionId); const record = await store.getRoomAgentTurn(item.missionId, item.turn.agentRunId); @@ -181,6 +196,7 @@ export async function supervisorTurnState(item: RoomAgentSupervisorQueueItem, st return "READY"; } +/** Finalizes a supervisor turn, recording terminal state and checkpointing mission status. */ export async function finishSupervisorTurn(input: { item: RoomAgentSupervisorQueueItem; status: "COMPLETED" | "FAILED" | "INTERRUPTED"; statusReason: string; errorCode?: string; }, store: SpaceStore) { diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 83d23cbc6..1a67209e5 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, workflowObservabilityInterceptorsPath } 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"; @@ -22,6 +23,7 @@ const namespace = process.env.SPACE_TEMPORAL_NAMESPACE ?? DEFAULT_TEMPORAL_NAMES const taskQueue = process.env.SPACE_TEMPORAL_TASK_QUEUE ?? DEFAULT_TASK_QUEUE; const workflowsPath = fileURLToPath(new URL("./workflows.js", import.meta.url)); +/** Parses and clamps an interval in milliseconds between 1,000ms and 60,000ms. */ function boundedInterval(value: string | undefined): number { const parsed = Number.parseInt(value ?? "2000", 10); return Number.isFinite(parsed) ? Math.max(1_000, Math.min(parsed, 60_000)) : 2_000; @@ -43,9 +45,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: { + workflowModules: [workflowObservabilityInterceptorsPath], + activity: [createActivityObservabilityInterceptors({ spans })] + }, + 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..c844e4b8f --- /dev/null +++ b/apps/worker/src/workflow-observability-interceptors.ts @@ -0,0 +1,805 @@ +import { + ContinueAsNew, + defaultPayloadConverter, + isCancellation, + proxySinks, + rootCause, + workflowInfo, + type ActivityInput, + type ConcludeActivationInput, + type ConcludeActivationOutput, + type ContinueAsNewInput, + type Headers, + type LocalActivityInput, + type Next, + type SignalInput, + type Sinks, + type StartChildWorkflowExecutionInput, + type WorkflowExecuteInput, + type WorkflowInboundCallsInterceptor, + type WorkflowInternalsInterceptor, + type WorkflowInterceptorsFactory, + type WorkflowOutboundCallsInterceptor +} from "@temporalio/workflow"; +import { + categorizeFailureCode, + deterministicSpanId, + formatTraceparent, + isSafeErrorCode, + parseTraceparent +} from "@space/contracts"; + +export const SPACE_TRACE_HEADER = "space-trace-context"; + +export interface SpaceTraceLinkPayload { + traceId: string; + spanId: string; + attributes?: Record; +} + +export interface SpaceTraceHeaderPayload { + traceparent?: string; + traceId?: string; + parentTraceId?: string; + roomId?: string; + paneId?: string; + missionId?: string; + agentRunId?: string; + continuationOfRunId?: string; + continuationGeneration?: number; + actionId?: string; + goalId?: string; + dotId?: string; + turnId?: string; + changeSetId?: string; + consolidationRunId?: string; + continuedFromSpanId?: string; + continuedFromRunId?: string; + links?: SpaceTraceLinkPayload[]; +} + +export interface WorkflowSpanRecordPayload { + traceId: string; + spanId: string; + parentSpanId: string | null; + name: string; + startTimeMs: number; + endTimeMs: number; + error: boolean; + attributes: Record; + links?: SpaceTraceLinkPayload[]; +} + +export interface WorkflowObservabilitySinks extends Sinks { + observability: { + recordWorkflowSpan(payload: WorkflowSpanRecordPayload): void; + }; +} + +const { observability } = proxySinks(); + +const idPattern = /^[A-Za-z0-9][A-Za-z0-9:._-]{0,239}$/; +const forbiddenTokenPattern = /^(?:https?:|file:|ws:|wss:|ftp:|sk-[A-Za-z0-9_-]{6,})/i; + +/** Validates and returns a safe identifier matching idPattern, or undefined if unsafe. */ +function safeId(value: unknown): string | undefined { + return typeof value === "string" && idPattern.test(value) && !forbiddenTokenPattern.test(value) ? value : undefined; +} + +/** Converts an unknown value to a non-array record object or null. */ +function asRecord(value: unknown): Record | null { + return value && typeof value === "object" && !Array.isArray(value) ? (value as Record) : null; +} + +/** Decodes and validates the space-trace-context header payload from inbound Temporal headers. */ +export function decodeSpaceTraceHeader(headers: Headers | undefined): SpaceTraceHeaderPayload | null { + const raw = headers?.[SPACE_TRACE_HEADER]; + if (!raw) return null; + try { + const decoded = defaultPayloadConverter.fromPayload(raw); + const record = asRecord(decoded); + if (!record) return null; + const links: SpaceTraceLinkPayload[] = []; + if (Array.isArray(record.links)) { + for (const entry of record.links) { + const link = asRecord(entry); + const parsed = link && typeof link.traceId === "string" && typeof link.spanId === "string" + ? parseTraceparent(`00-${link.traceId}-${link.spanId}-01`) + : null; + if (!parsed) continue; + const attrsRecord = asRecord(link?.attributes); + const attributes: Record = {}; + if (attrsRecord) { + for (const [k, v] of Object.entries(attrsRecord)) { + if (typeof v === "string" || typeof v === "number") attributes[k] = v; + } + } + links.push({ + traceId: parsed.traceId, + spanId: parsed.parentId, + ...(Object.keys(attributes).length ? { attributes } : {}) + }); + } + } + const parsedTraceparent = parseTraceparent(record.traceparent); + const generation = typeof record.continuationGeneration === "number" && Number.isInteger(record.continuationGeneration) && record.continuationGeneration > 0 + ? record.continuationGeneration + : undefined; + return { + ...(parsedTraceparent ? { traceparent: formatTraceparent(parsedTraceparent.traceId, parsedTraceparent.parentId, parsedTraceparent.flags) } : {}), + ...(safeId(record.traceId) ? { traceId: safeId(record.traceId)! } : {}), + ...(safeId(record.parentTraceId) ? { parentTraceId: safeId(record.parentTraceId)! } : {}), + ...(safeId(record.roomId) ? { roomId: safeId(record.roomId)! } : {}), + ...(safeId(record.paneId) ? { paneId: safeId(record.paneId)! } : {}), + ...(safeId(record.missionId) ? { missionId: safeId(record.missionId)! } : {}), + ...(safeId(record.agentRunId) ? { agentRunId: safeId(record.agentRunId)! } : {}), + ...(safeId(record.continuationOfRunId) ? { continuationOfRunId: safeId(record.continuationOfRunId)! } : {}), + ...(generation !== undefined ? { continuationGeneration: generation } : {}), + ...(safeId(record.actionId) ? { actionId: safeId(record.actionId)! } : {}), + ...(safeId(record.goalId) ? { goalId: safeId(record.goalId)! } : {}), + ...(safeId(record.dotId) ? { dotId: safeId(record.dotId)! } : {}), + ...(safeId(record.turnId) ? { turnId: safeId(record.turnId)! } : {}), + ...(safeId(record.changeSetId) ? { changeSetId: safeId(record.changeSetId)! } : {}), + ...(safeId(record.consolidationRunId) ? { consolidationRunId: safeId(record.consolidationRunId)! } : {}), + ...(safeId(record.continuedFromSpanId) ? { continuedFromSpanId: safeId(record.continuedFromSpanId)! } : {}), + ...(safeId(record.continuedFromRunId) ? { continuedFromRunId: safeId(record.continuedFromRunId)! } : {}), + ...(links.length ? { links } : {}) + }; + } catch { + return null; + } +} + +/** Encodes and attaches the space-trace-context payload into outbound Temporal headers. */ +export function encodeSpaceTraceHeader(headers: Headers, payload: SpaceTraceHeaderPayload): void { + const encoded = defaultPayloadConverter.toPayload(payload); + if (encoded) headers[SPACE_TRACE_HEADER] = encoded; +} + +/** Traverses workflow or activity input arguments to extract correlation IDs, traceparent, and links. */ +export function extractArgsTraceContext(args: readonly unknown[], workflowType?: string): SpaceTraceHeaderPayload { + const out: SpaceTraceHeaderPayload = {}; + const links: SpaceTraceLinkPayload[] = []; + /** Inspects a candidate argument object to extract trace correlation identifiers and span links. */ + const visit = (candidate: Record | null) => { + if (!candidate) return; + const tp = parseTraceparent(candidate.traceparent); + if (tp && !out.traceparent) out.traceparent = formatTraceparent(tp.traceId, tp.parentId, tp.flags); + out.traceId ??= safeId(candidate.traceId); + out.roomId ??= safeId(candidate.roomId); + out.paneId ??= safeId(candidate.paneId) ?? safeId(candidate.agentPaneId); + out.missionId ??= safeId(candidate.missionId) ?? safeId(candidate.roomAgentMissionId); + out.agentRunId ??= safeId(candidate.agentRunId); + out.actionId ??= safeId(candidate.actionId); + out.goalId ??= safeId(candidate.goalId); + out.dotId ??= safeId(candidate.dotId); + out.turnId ??= safeId(candidate.turnId); + out.changeSetId ??= safeId(candidate.changeSetId); + if (workflowType === "memoryConsolidationWorkflow" || (typeof candidate.runId === "string" && candidate.runId.startsWith("memory_consolidation:"))) { + out.consolidationRunId ??= safeId(candidate.runId); + } + const continuation = asRecord(candidate.roomAgentContinuation); + if (continuation) { + const sourceRunId = safeId(continuation.sourceRunId); + const sourceTraceId = safeId(continuation.sourceTraceId); + const generation = typeof continuation.generation === "number" && Number.isInteger(continuation.generation) && continuation.generation > 0 + ? continuation.generation + : undefined; + if (sourceRunId) out.continuationOfRunId ??= sourceRunId; + if (generation !== undefined) out.continuationGeneration ??= generation; + const sourceTp = parseTraceparent(continuation.sourceTraceparent); + if (sourceTp) { + links.push({ + traceId: sourceTp.traceId, + spanId: sourceTp.parentId, + attributes: { + "space.link_type": "background_continuation", + ...(sourceRunId ? { "space.source_run_id": sourceRunId } : {}), + ...(sourceTraceId ? { "space.source_trace_id": sourceTraceId } : {}), + ...(generation !== undefined ? { "space.continuation_generation": generation } : {}) + } + }); + } + } + }; + + for (const arg of args) { + if (typeof arg === "string" && /^dot_goal:[A-Za-z0-9_-]{8,64}$/.test(arg)) { + out.goalId ??= arg; + continue; + } + const record = asRecord(arg); + if (!record) continue; + visit(record); + visit(asRecord(record.turn)); + const item = asRecord(record.item); + visit(item); + visit(asRecord(item?.turn)); + const queueItem = asRecord(record.queueItem); + visit(queueItem); + visit(asRecord(queueItem?.turn)); + const bridge = asRecord(record.bridge); + if (bridge) { + out.parentTraceId ??= safeId(bridge.requestId); + visit(bridge); + } + } + if (links.length) out.links = links; + return out; +} + +/** Extracts W3C traceparent and correlation IDs from Temporal workflow memo. */ +function extractMemoTraceContext(memo: Record | undefined): SpaceTraceHeaderPayload { + if (!memo) return {}; + /** Decodes a memo field value supporting raw strings and Temporal Payload objects. */ + const readField = (key: string): unknown => { + const raw = memo[key]; + const record = asRecord(raw); + if (record && "metadata" in record && "data" in record) { + try { + return defaultPayloadConverter.fromPayload(raw as never); + } catch { + return undefined; + } + } + return raw; + }; + const tp = parseTraceparent(readField("traceparent")); + const traceId = safeId(readField("spaceTraceId")); + const parentTraceId = safeId(readField("parentSpaceTraceId")); + const missionId = safeId(readField("missionId")); + return { + ...(tp ? { traceparent: formatTraceparent(tp.traceId, tp.parentId, tp.flags) } : {}), + ...(traceId ? { traceId } : {}), + ...(parentTraceId ? { parentTraceId } : {}), + ...(missionId ? { missionId } : {}) + }; +} + +/** Merges and deduplicates two lists of OTLP span links by trace ID, span ID, and link type. */ +function mergeLinks( + first: readonly SpaceTraceLinkPayload[] | undefined, + second: readonly SpaceTraceLinkPayload[] | undefined +): SpaceTraceLinkPayload[] | undefined { + const combined = [...(first ?? []), ...(second ?? [])]; + if (!combined.length) return undefined; + const seen = new Set(); + const deduped: SpaceTraceLinkPayload[] = []; + for (const link of combined) { + const key = `${link.traceId}:${link.spanId}:${link.attributes?.["space.link_type"] ?? ""}`; + if (seen.has(key)) continue; + seen.add(key); + deduped.push(link); + } + return deduped; +} + +/** Merges multiple trace context payloads with earlier non-empty values taking precedence. */ +export function mergeTraceContexts(...contexts: Array): SpaceTraceHeaderPayload { + const merged: SpaceTraceHeaderPayload = {}; + let links: SpaceTraceLinkPayload[] | undefined; + for (const ctx of contexts) { + if (!ctx) continue; + merged.traceparent ??= ctx.traceparent; + merged.traceId ??= ctx.traceId; + merged.parentTraceId ??= ctx.parentTraceId; + merged.roomId ??= ctx.roomId; + merged.paneId ??= ctx.paneId; + merged.missionId ??= ctx.missionId; + merged.agentRunId ??= ctx.agentRunId; + merged.continuationOfRunId ??= ctx.continuationOfRunId; + merged.continuationGeneration ??= ctx.continuationGeneration; + merged.actionId ??= ctx.actionId; + merged.goalId ??= ctx.goalId; + merged.dotId ??= ctx.dotId; + merged.turnId ??= ctx.turnId; + merged.changeSetId ??= ctx.changeSetId; + merged.consolidationRunId ??= ctx.consolidationRunId; + merged.continuedFromSpanId ??= ctx.continuedFromSpanId; + merged.continuedFromRunId ??= ctx.continuedFromRunId; + links = mergeLinks(links, ctx.links); + } + if (links?.length) merged.links = links; + return merged; +} + +/** Generates OTLP span links to the previous run when continuing as new. */ +function continueAsNewLinks(ctx: SpaceTraceHeaderPayload, continuedFromRunId?: string): SpaceTraceLinkPayload[] | undefined { + const tp = parseTraceparent(ctx.traceparent); + const continuedSpanId = ctx.continuedFromSpanId; + const runId = ctx.continuedFromRunId ?? continuedFromRunId; + const extra: SpaceTraceLinkPayload[] = []; + if (tp && continuedSpanId && /^[0-9a-f]{16}$/.test(continuedSpanId)) { + extra.push({ + traceId: tp.traceId, + spanId: continuedSpanId, + attributes: { + "space.link_type": "continue_as_new", + ...(runId && safeId(runId) ? { "space.source_run_id": runId } : {}) + } + }); + } + return mergeLinks(ctx.links, extra); +} + +/** Constructs allowlisted OTLP span attributes for a workflow execution. */ +function buildWorkflowSpanAttributes( + ctx: SpaceTraceHeaderPayload, + outcome: string, + durationMs: number, + reasonCode: string | null +): Record { + const info = workflowInfo(); + const attributes: Record = { + "space.workflow_type": info.workflowType, + "space.workflow_id": info.workflowId, + "space.run_id": info.runId, + "space.task_queue": info.taskQueue, + "space.attempt": info.attempt, + "space.outcome": outcome, + "space.duration_ms": Math.max(0, Math.round(durationMs)) + }; + if (info.parent) { + if (safeId(info.parent.workflowId)) attributes["space.parent_workflow_id"] = info.parent.workflowId; + if (safeId(info.parent.runId)) attributes["space.parent_run_id"] = info.parent.runId; + } + const continuedRunId = ctx.continuedFromRunId ?? info.continuedFromExecutionRunId; + if (continuedRunId && safeId(continuedRunId)) { + attributes["space.continued_from_run_id"] = continuedRunId; + } + if (ctx.traceId) attributes["space.trace_id"] = ctx.traceId; + if (ctx.parentTraceId) attributes["space.parent_trace_id"] = ctx.parentTraceId; + if (ctx.roomId) attributes["space.room_id"] = ctx.roomId; + if (ctx.paneId) attributes["space.pane_id"] = ctx.paneId; + if (ctx.missionId) attributes["space.mission_id"] = ctx.missionId; + if (ctx.agentRunId) attributes["space.agent_run_id"] = ctx.agentRunId; + if (ctx.continuationOfRunId) attributes["space.continuation_of_run_id"] = ctx.continuationOfRunId; + if (ctx.continuationGeneration !== undefined) attributes["space.continuation_generation"] = ctx.continuationGeneration; + if (ctx.actionId) attributes["space.action_id"] = ctx.actionId; + if (ctx.goalId) attributes["space.goal_id"] = ctx.goalId; + if (ctx.dotId) attributes["space.dot_id"] = ctx.dotId; + if (ctx.turnId) attributes["space.turn_id"] = ctx.turnId; + if (ctx.changeSetId) attributes["space.change_set_id"] = ctx.changeSetId; + if (ctx.consolidationRunId) attributes["space.consolidation_run_id"] = ctx.consolidationRunId; + if (reasonCode && isSafeErrorCode(reasonCode)) { + attributes["space.reason_code"] = reasonCode; + attributes["space.error_code"] = reasonCode; + attributes["space.category"] = categorizeFailureCode(reasonCode); + } + return attributes; +} + +/** Classifies the outcome, error flag, and reason code from a completed workflow return value. */ +function classifyWorkflowResult(result: unknown): { outcome: string; error: boolean; reasonCode: string | null } { + const record = asRecord(result); + if (record && (record.status === "FAILED" || record.status === "OPERATOR_REQUIRED")) { + const code = [record.reasonCode, record.errorCode, record.code].find(isSafeErrorCode) ?? null; + return { outcome: "FAILED_RESULT", error: true, reasonCode: code }; + } + return { outcome: "COMPLETED", error: false, reasonCode: null }; +} + +/** Classifies the outcome, error flag, and stable reason code from a workflow error or rejection. */ +function classifyWorkflowError(err: unknown): { outcome: string; error: boolean; reasonCode: string | null } { + if (err instanceof ContinueAsNew || (err as { name?: string } | null)?.name === "ContinueAsNew") { + return { outcome: "CONTINUED_AS_NEW", error: false, reasonCode: null }; + } + if (isCancellation(err)) { + return { outcome: "CANCELLED", error: false, reasonCode: "WORKFLOW_CANCELLED" }; + } + const cause = rootCause(err); + for (const candidate of [err, cause]) { + const rec = asRecord(candidate); + if (!rec) continue; + for (const field of [rec.code, rec.type, rec.reasonCode, rec.errorCode]) { + if (isSafeErrorCode(field)) return { outcome: "FAILED", error: true, reasonCode: field }; + } + } + return { outcome: "FAILED", error: true, reasonCode: "WORKFLOW_ERROR" }; +} + +interface ActiveSupervisorTurnSpan { + turnKey: string; + spanId: string; + context: SpaceTraceHeaderPayload; + startTimeMs: number; + inFlightActivities: number; + inFlightChildren: number; + childCompleted: boolean; + discardCompleted: boolean; + finishCompleted: boolean; + outcome: string; + error: boolean; + reasonCode: string | null; + emitted: boolean; +} + +const supervisorTurnTerminalActivities = new Set([ + "finishRoomAgentSupervisorTurn", + "markRoomAgentMissionFinished", + "markRoomAgentMissionContinued" +]); + +/** + * Temporal workflow interceptors factory providing inbound, outbound, and internal interceptors + * for W3C traceparent propagation, workflow spans, child workflow spans, and activity spans. + */ +export const interceptors: WorkflowInterceptorsFactory = () => { + let workflowContext: SpaceTraceHeaderPayload = {}; + let workflowSpanId = ""; + let workflowStartTimeMs = 0; + let continueAsNewCarryIncluded = false; + + // Room Agent Supervisor per-mission-turn span state. + const missionContexts = new Map(); + let activeTurn: ActiveSupervisorTurnSpan | null = null; + + const emitSpan = ( + ctx: SpaceTraceHeaderPayload, + spanId: string, + startTimeMs: number, + endTimeMs: number, + outcome: string, + error: boolean, + reasonCode: string | null, + links: SpaceTraceLinkPayload[] | undefined + ) => { + const parsed = parseTraceparent(ctx.traceparent); + if (!parsed) return; + observability.recordWorkflowSpan({ + traceId: parsed.traceId, + spanId, + parentSpanId: parsed.parentId, + name: `workflow ${workflowInfo().workflowType}`, + startTimeMs, + endTimeMs: Math.max(startTimeMs, endTimeMs), + error, + attributes: buildWorkflowSpanAttributes(ctx, outcome, Math.max(0, endTimeMs - startTimeMs), reasonCode), + ...(links?.length ? { links } : {}) + }); + }; + + /** Emits the active supervisor turn workflow span and links when the turn completes or transitions. */ + const flushActiveSupervisorTurn = () => { + if (!activeTurn || activeTurn.emitted) return; + activeTurn.emitted = true; + const turn = activeTurn; + activeTurn = null; + const links = !continueAsNewCarryIncluded + ? continueAsNewLinks( + mergeTraceContexts(turn.context, { + continuedFromSpanId: workflowContext.continuedFromSpanId, + continuedFromRunId: workflowContext.continuedFromRunId + }), + workflowInfo().continuedFromExecutionRunId + ) + : turn.context.links; + continueAsNewCarryIncluded = true; + emitSpan(turn.context, turn.spanId, turn.startTimeMs, Date.now(), turn.outcome, turn.error, turn.reasonCode, links); + }; + + /** Merges turn call context and stored mission context, only inheriting workflowContext trace attributes when matching the same request trace. */ + const mergeSupervisorTurnContext = ( + callCtx: SpaceTraceHeaderPayload, + missionCtx?: SpaceTraceHeaderPayload + ): SpaceTraceHeaderPayload => { + const current = mergeTraceContexts(callCtx, missionCtx); + const sameRequest = Boolean( + current.traceId && + workflowContext.traceId && + current.traceId === workflowContext.traceId + ); + return mergeTraceContexts( + current, + sameRequest + ? workflowContext + : { ...workflowContext, traceparent: undefined, traceId: undefined } + ); + }; + + /** Obtains or lazily initializes the active turn span tracking execution for a supervisor mission turn. */ + const ensureSupervisorTurnSpan = (callCtx: SpaceTraceHeaderPayload): ActiveSupervisorTurnSpan | null => { + const info = workflowInfo(); + const missionId = + callCtx.missionId ?? + (activeTurn && (!callCtx.traceId || callCtx.traceId === activeTurn.context.traceId) + ? activeTurn.context.missionId + : undefined); + const merged = mergeSupervisorTurnContext( + callCtx, + missionId ? missionContexts.get(missionId) : undefined + ); + if (!parseTraceparent(merged.traceparent)) return null; + const turnKey = merged.agentRunId ?? `${merged.missionId ?? "mission"}:${merged.traceId ?? "trace"}`; + if (activeTurn && activeTurn.turnKey !== turnKey) { + flushActiveSupervisorTurn(); + } + if (!activeTurn) { + const spanId = deterministicSpanId(`workflow:${info.workflowId}:${info.runId}:${turnKey}`); + activeTurn = { + turnKey, + spanId, + context: merged, + startTimeMs: Date.now(), + inFlightActivities: 0, + inFlightChildren: 0, + childCompleted: false, + discardCompleted: false, + finishCompleted: false, + outcome: "COMPLETED", + error: false, + reasonCode: null, + emitted: false + }; + } else { + activeTurn.context = mergeTraceContexts(activeTurn.context, merged); + } + if (activeTurn.context.missionId) { + missionContexts.set(activeTurn.context.missionId, activeTurn.context); + } + return activeTurn; + }; + + /** Rewrites traceparent in the header payload so child activities and workflows point to parentSpanId. */ + const childContextHeader = (baseCtx: SpaceTraceHeaderPayload, parentSpanId: string): SpaceTraceHeaderPayload | null => { + const parsed = parseTraceparent(baseCtx.traceparent); + if (!parsed) return null; + return { + ...baseCtx, + traceparent: formatTraceparent(parsed.traceId, parentSpanId, parsed.flags) + }; + }; + + /** Caches mission trace contexts from pending or interrupted queue items in supervisor workflow args. */ + const rememberSupervisorItemsFromArgs = (args: readonly unknown[]) => { + for (const arg of args) { + const rec = asRecord(arg); + if (!rec) continue; + const items: unknown[] = [ + rec, + ...(Array.isArray(rec.pending) ? rec.pending : []), + ...(Array.isArray(rec.interrupted) ? rec.interrupted.map((entry) => asRecord(entry)?.item) : []) + ]; + for (const item of items) { + const itemCtx = extractArgsTraceContext([item], "roomAgentSupervisorWorkflow"); + if (itemCtx.missionId && parseTraceparent(itemCtx.traceparent)) { + missionContexts.set(itemCtx.missionId, mergeTraceContexts(itemCtx, workflowContext)); + } + } + } + }; + + const inbound: WorkflowInboundCallsInterceptor = { + async execute(input: WorkflowExecuteInput, next: Next): Promise { + const info = workflowInfo(); + workflowStartTimeMs = Date.now(); + workflowSpanId = deterministicSpanId(`workflow:${info.workflowId}:${info.runId}`); + const headerCtx = decodeSpaceTraceHeader(input.headers); + const argsCtx = extractArgsTraceContext(input.args, info.workflowType); + const memoCtx = extractMemoTraceContext(info.memo); + workflowContext = mergeTraceContexts(headerCtx, argsCtx, memoCtx); + + if (info.workflowType === "roomAgentSupervisorWorkflow") { + rememberSupervisorItemsFromArgs(input.args); + } + + try { + const result = await next(input); + if (info.workflowType === "roomAgentSupervisorWorkflow") { + flushActiveSupervisorTurn(); + } else { + const classified = classifyWorkflowResult(result); + emitSpan( + workflowContext, + workflowSpanId, + workflowStartTimeMs, + Date.now(), + classified.outcome, + classified.error, + classified.reasonCode, + continueAsNewLinks(workflowContext, info.continuedFromExecutionRunId) + ); + } + return result; + } catch (err) { + const classified = classifyWorkflowError(err); + if (info.workflowType === "roomAgentSupervisorWorkflow") { + if (activeTurn) { + if (classified.error || classified.outcome === "CANCELLED") { + activeTurn.outcome = classified.outcome; + activeTurn.error = classified.error; + activeTurn.reasonCode = classified.reasonCode; + } + flushActiveSupervisorTurn(); + } + } else { + emitSpan( + workflowContext, + workflowSpanId, + workflowStartTimeMs, + Date.now(), + classified.outcome, + classified.error, + classified.reasonCode, + continueAsNewLinks(workflowContext, info.continuedFromExecutionRunId) + ); + } + throw err; + } + }, + + async handleSignal(input: SignalInput, next: Next): Promise { + if (workflowInfo().workflowType === "roomAgentSupervisorWorkflow") { + const headerCtx = decodeSpaceTraceHeader(input.headers); + const signalCtx = mergeTraceContexts(headerCtx, extractArgsTraceContext(input.args, "roomAgentSupervisorWorkflow")); + if (signalCtx.missionId && parseTraceparent(signalCtx.traceparent)) { + missionContexts.set(signalCtx.missionId, mergeTraceContexts(signalCtx, workflowContext)); + } + } + return next(input); + } + }; + + const outbound: WorkflowOutboundCallsInterceptor = { + async scheduleActivity(input: ActivityInput, next: Next): Promise { + const info = workflowInfo(); + const callCtx = extractArgsTraceContext(input.args, info.workflowType); + + if (info.workflowType === "roomAgentSupervisorWorkflow") { + if (input.activityType === "prepareRoomAgentBackgroundContinuation" || input.activityType === "settleRoomAgentBackgroundMission") { + const stored = callCtx.missionId ? missionContexts.get(callCtx.missionId) : undefined; + const base = mergeSupervisorTurnContext(callCtx, stored); + const parentSpan = activeTurn?.spanId ?? (callCtx.missionId ? deterministicSpanId(`workflow:${info.workflowId}:${info.runId}:${callCtx.missionId}:${base.traceId ?? "trace"}`) : workflowSpanId); + const header = childContextHeader(base, parentSpan); + if (header) encodeSpaceTraceHeader(input.headers, header); + const res = await next(input); + const resRec = asRecord(res); + if (resRec?.state === "READY" && resRec.item) { + const readyCtx = extractArgsTraceContext([resRec.item], "roomAgentSupervisorWorkflow"); + if (readyCtx.missionId && parseTraceparent(readyCtx.traceparent)) { + missionContexts.set(readyCtx.missionId, readyCtx); + } + } + return res; + } + + const turnSpan = ensureSupervisorTurnSpan(callCtx); + if (turnSpan) { + const header = childContextHeader(turnSpan.context, turnSpan.spanId); + if (header) encodeSpaceTraceHeader(input.headers, header); + turnSpan.inFlightActivities += 1; + try { + const res = await next(input); + if (input.activityType === "discardRoomAgentSupervisorTurn") { + turnSpan.discardCompleted = true; + } + if (supervisorTurnTerminalActivities.has(input.activityType)) { + const statusArg = asRecord(input.args[0])?.status; + if (statusArg === "FAILED") { + turnSpan.outcome = "FAILED_RESULT"; + turnSpan.error = true; + } else if (statusArg === "INTERRUPTED") { + turnSpan.outcome = "CANCELLED"; + turnSpan.reasonCode = "ROOM_AGENT_STOPPED"; + } + turnSpan.finishCompleted = true; + } + return res; + } catch (err) { + const classified = classifyWorkflowError(err); + turnSpan.outcome = classified.outcome; + turnSpan.error = classified.error; + turnSpan.reasonCode = classified.reasonCode; + throw err; + } finally { + turnSpan.inFlightActivities = Math.max(0, turnSpan.inFlightActivities - 1); + if (turnSpan.finishCompleted && turnSpan.inFlightActivities === 0 && turnSpan.inFlightChildren === 0) { + flushActiveSupervisorTurn(); + } + } + } + } + + workflowContext = mergeTraceContexts(workflowContext, callCtx); + const header = childContextHeader(mergeTraceContexts(callCtx, workflowContext), workflowSpanId); + if (header) encodeSpaceTraceHeader(input.headers, header); + return next(input); + }, + + async scheduleLocalActivity(input: LocalActivityInput, next: Next): Promise { + const info = workflowInfo(); + const callCtx = extractArgsTraceContext(input.args, info.workflowType); + workflowContext = mergeTraceContexts(workflowContext, callCtx); + const header = childContextHeader(mergeTraceContexts(callCtx, workflowContext), workflowSpanId); + if (header) encodeSpaceTraceHeader(input.headers, header); + return next(input); + }, + + async startChildWorkflowExecution( + input: StartChildWorkflowExecutionInput, + next: Next + ): Promise<[Promise, Promise]> { + const info = workflowInfo(); + const callCtx = extractArgsTraceContext(input.options.args ?? [], input.workflowType); + + if (info.workflowType === "roomAgentSupervisorWorkflow") { + const turnSpan = ensureSupervisorTurnSpan(callCtx); + if (turnSpan) { + const header = childContextHeader(turnSpan.context, turnSpan.spanId); + if (header) encodeSpaceTraceHeader(input.headers, header); + turnSpan.inFlightChildren += 1; + const [started, completed] = await next(input); + const trackedCompleted = completed.then( + (res) => { + turnSpan.inFlightChildren = Math.max(0, turnSpan.inFlightChildren - 1); + turnSpan.childCompleted = true; + const classified = classifyWorkflowResult(res); + if (classified.error) { + turnSpan.outcome = classified.outcome; + turnSpan.error = true; + turnSpan.reasonCode = classified.reasonCode; + } + return res; + }, + (err) => { + turnSpan.inFlightChildren = Math.max(0, turnSpan.inFlightChildren - 1); + turnSpan.childCompleted = true; + const classified = classifyWorkflowError(err); + turnSpan.outcome = classified.outcome; + turnSpan.error = classified.error; + turnSpan.reasonCode = classified.reasonCode; + throw err; + } + ); + return [started, trackedCompleted]; + } + } + + workflowContext = mergeTraceContexts(workflowContext, callCtx); + const header = childContextHeader(mergeTraceContexts(callCtx, workflowContext), workflowSpanId); + if (header) encodeSpaceTraceHeader(input.headers, header); + return next(input); + }, + + async continueAsNew(input: ContinueAsNewInput, next: Next): Promise { + const info = workflowInfo(); + const callCtx = extractArgsTraceContext(input.args, input.options.workflowType); + const latestMissionCtx = missionContexts.size ? Array.from(missionContexts.values()).at(-1) : undefined; + const continuedFromSpanId = activeTurn?.spanId ?? workflowSpanId; + if (info.workflowType === "roomAgentSupervisorWorkflow") { + flushActiveSupervisorTurn(); + } + const nextCtx = mergeTraceContexts(callCtx, workflowContext, latestMissionCtx); + if (parseTraceparent(nextCtx.traceparent)) { + encodeSpaceTraceHeader(input.headers, { + ...nextCtx, + continuedFromSpanId, + continuedFromRunId: info.runId + }); + } + return next(input); + } + }; + + const internals: WorkflowInternalsInterceptor = { + concludeActivation(input: ConcludeActivationInput, next: Next): ConcludeActivationOutput { + const output = next(input); + if ( + activeTurn && + (activeTurn.childCompleted || activeTurn.discardCompleted || activeTurn.finishCompleted) && + activeTurn.inFlightActivities === 0 && + activeTurn.inFlightChildren === 0 + ) { + flushActiveSupervisorTurn(); + } + return output; + } + }; + + return { + inbound: [inbound], + outbound: [outbound], + internals: [internals] + }; +}; + diff --git a/apps/worker/src/workflow-observability.ts b/apps/worker/src/workflow-observability.ts new file mode 100644 index 000000000..d3b6540aa --- /dev/null +++ b/apps/worker/src/workflow-observability.ts @@ -0,0 +1,63 @@ +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, + WorkflowSpanRecordPayload +} from "./workflow-observability-interceptors.js"; + +const candidateJs = fileURLToPath(new URL("./workflow-observability-interceptors.js", import.meta.url)); + +/** + * Path to the workflow interceptor module loaded into the Temporal workflow + * bundle. Resolves to `.ts` when running directly from source in tests and + * `.js` in compiled builds. + */ +export const workflowObservabilityInterceptorsPath = existsSync(candidateJs) + ? candidateJs + : candidateJs.replace(/\.js$/, ".ts"); + +/** Translates a workflow sink span payload into a ServerSpan and passes it to the exporter. */ +export function recordWorkflowSpanFromSink( + payload: WorkflowSpanRecordPayload, + spans: SpanExporter | null | undefined +): void { + if (!spans) return; + const startNs = BigInt(Math.max(0, Math.round(payload.startTimeMs * 1_000_000))); + const endNs = BigInt(Math.max(Number(startNs), Math.round(payload.endTimeMs * 1_000_000))); + spans.record({ + traceId: payload.traceId, + spanId: payload.spanId, + parentSpanId: payload.parentSpanId, + name: payload.name, + kind: 1, + startTimeUnixNano: startNs, + endTimeUnixNano: endNs, + attributes: sanitizeSpanAttributes(payload.attributes), + ...(payload.links?.length ? { links: payload.links } : {}), + error: payload.error + }); +} + +/** + * Temporal workflow sinks for OTLP workflow span export. + * `callDuringReplay: false` guarantees replaying workflow histories never + * re-exports duplicate spans or performs non-deterministic I/O inside the + * workflow sandbox. + */ +export function createWorkflowObservabilitySinks( + options: { spans?: SpanExporter | null } = {} +): InjectedSinks { + return { + observability: { + recordWorkflowSpan: { + fn(_info, payload) { + recordWorkflowSpanFromSink(payload, options.spans); + }, + callDuringReplay: false + } + } + }; +} + diff --git a/apps/worker/tests/activity-observability.test.ts b/apps/worker/tests/activity-observability.test.ts index 654bf7962..bd9388223 100644 --- a/apps/worker/tests/activity-observability.test.ts +++ b/apps/worker/tests/activity-observability.test.ts @@ -16,10 +16,12 @@ import { resolveWorkerMetricsListen, startWorkerMetricsServer } from "../src/worker-metrics.js"; +import { encodeSpaceTraceHeader } from "../src/workflow-observability-interceptors.js"; const traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; const secretPrompt = "PRIVATE PROMPT: deploy with token sk-live-worker-secret-123"; +/** Generates mock ActivityInfo with customizable defaults for testing. */ const info = (overrides: Partial[0]> = {}) => ({ activityType: "runCodexAppServerTurn", activityId: "3", @@ -30,6 +32,7 @@ const info = (overrides: Partial[0]> ...overrides }); +/** Intercepts worker log records into memory for test assertions and returns a cleanup function. */ function capture() { const records: Array> = []; const lines: string[] = []; @@ -162,6 +165,7 @@ describe("worker metrics endpoint", () => { const port = 20_000 + Math.floor(Math.random() * 20_000); const server = await startWorkerMetricsServer({ host: "127.0.0.1", port }); expect(server).not.toBeNull(); + /** Performs a test HTTP GET request against the local Prometheus metrics port. */ const get = (path: string) => new Promise<{ status: number; body: string }>((resolve, reject) => { request({ host: "127.0.0.1", port, path }, (response) => { let body = ""; @@ -182,11 +186,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 interceptors 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(/workflowModules:\s*\[workflowObservabilityInterceptorsPath\]/); + expect(source, entry).toMatch(/activity:\s*\[createActivityObservabilityInterceptors\(\{ spans \}\)\]/); + expect(source, entry).toMatch(/sinks:\s*createWorkflowObservabilitySinks\(\{ spans \}\)/); } }); @@ -203,3 +209,21 @@ it("exports one INTERNAL span per activity under the caller's traceparent", asyn expect(recorded[0]).toMatchObject({ traceId: "4bf92f3577b34da6a3ce929d0e0e4736", parentSpanId: "00f067aa0ba902b7", name: "activity runTurn", kind: 1, error: false, startTimeUnixNano: 1_000_000_000n, endTimeUnixNano: 1_250_000_000n }); }); + +it("exports an activity span under the parent from inbound workflow headers", async () => { + const recorded: Array> = []; + const spans = { record: (span: Record) => recorded.push(span), flush: async () => undefined, close: async () => undefined, stats: () => ({ exported: 0, dropped: 0, failed: 0 }) }; + const metrics = createWorkerMetrics(new MetricsRegistry()); + const headerInfo = { activityType: "runHeaderTurn", activityId: "1", attempt: 1, currentAttemptScheduledTimestampMs: 1_000, workflowExecution: { workflowId: "w", runId: "r" } } as never; + const parentTraceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"; + const headers: Record = {}; + encodeSpaceTraceHeader(headers, { traceId: "4bf92f3577b34da6a3ce929d0e0e4736", traceparent: parentTraceparent }); + await observeActivityExecution(headerInfo, [{ noTraceparent: true }], async () => "ok", { metrics, headers, spans: spans as never }); + expect(recorded).toHaveLength(1); + expect(recorded[0]).toMatchObject({ + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + parentSpanId: "00f067aa0ba902b7", + name: "activity runHeaderTurn" + }); +}); + diff --git a/apps/worker/tests/otlp-trace-continuity.test.ts b/apps/worker/tests/otlp-trace-continuity.test.ts new file mode 100644 index 000000000..3e940ea5b --- /dev/null +++ b/apps/worker/tests/otlp-trace-continuity.test.ts @@ -0,0 +1,1070 @@ +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import type { AddressInfo } from "node:net"; +import Fastify from "fastify"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + deterministicSpanId, + formatTraceparent, + parseTraceparent, + type DummyTurnInput, + type RoomAgentSupervisorQueueItem +} from "@space/contracts"; +import { + ALLOWED_SPAN_ATTRIBUTE_KEYS, + ALLOWED_SPAN_LINK_ATTRIBUTE_KEYS, + createOtlpSpanExporter, + otlpTraceRequest, + sanitizeSpanAttributes, + type SpanExporter +} from "@space/runtime"; +import { + createActivityObservabilityInterceptors, + noteActivityFailureReason +} from "../src/activity-observability.js"; +import { prepareBackgroundContinuation } from "../src/room-background-continuation.js"; +import { + createWorkflowObservabilitySinks, + recordWorkflowSpanFromSink, + workflowObservabilityInterceptorsPath +} from "../src/workflow-observability.js"; +import { + ContinueAsNew, + type Headers, + type WorkflowInfo +} from "@temporalio/workflow"; +import { continuationFixture } from "./room-continuation-fixture.js"; + +const workflowState = vi.hoisted(() => ({ + info: null as WorkflowInfo | null, + sinks: null as ReturnType | null, + nowMs: 1_700_000_000_000 +})); + +vi.mock("@temporalio/workflow", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + workflowInfo: () => { + if (!workflowState.info) throw new Error("workflowInfo() called outside simulated workflow context"); + return workflowState.info; + }, + proxySinks: () => ({ + observability: { + recordWorkflowSpan(payload: Parameters[0]) { + if (workflowState.info && workflowState.sinks) { + void workflowState.sinks.observability.recordWorkflowSpan.fn(workflowState.info, payload); + } + } + } + }) + }; +}); + +import { + decodeSpaceTraceHeader, + encodeSpaceTraceHeader, + interceptors as createWorkflowInterceptors +} from "../src/workflow-observability-interceptors.js"; + +interface OtlpExportedAttribute { + key: string; + value: { stringValue?: string; intValue?: string }; +} + +interface OtlpExportedLink { + traceId: string; + spanId: string; + attributes?: OtlpExportedAttribute[]; +} + +interface OtlpExportedSpan { + serviceName: string; + traceId: string; + spanId: string; + parentSpanId?: string; + name: string; + kind: number; + startTimeUnixNano: string; + endTimeUnixNano: string; + attributes: Record; + links?: Array<{ + traceId: string; + spanId: string; + attributes: Record; + }>; + status: { code: number }; +} + +/** Decodes an array of OTLP attribute key/value pairs into a flat record of strings and numbers. */ +function decodeAttributes(entries: OtlpExportedAttribute[] | undefined): Record { + const out: Record = {}; + for (const entry of entries ?? []) { + if (entry.value.intValue !== undefined) out[entry.key] = Number(entry.value.intValue); + else if (entry.value.stringValue !== undefined) out[entry.key] = entry.value.stringValue; + } + return out; +} + +/** Spins up an ephemeral local HTTP server implementing the OTLP/HTTP traces JSON receiver contract. */ +async function startFakeOtlpCollector() { + const payloads: unknown[] = []; + const spans: OtlpExportedSpan[] = []; + const server = createServer((req: IncomingMessage, res: ServerResponse) => { + if (req.method !== "POST" || req.url !== "/v1/traces") { + res.writeHead(404).end(); + return; + } + let body = ""; + req.on("data", (chunk) => { body += String(chunk); }); + req.on("end", () => { + const parsed = JSON.parse(body); + payloads.push(parsed); + for (const rs of parsed.resourceSpans ?? []) { + const serviceAttr = (rs.resource?.attributes ?? []).find((a: OtlpExportedAttribute) => a.key === "service.name"); + const serviceName = serviceAttr?.value?.stringValue ?? "unknown"; + for (const ss of rs.scopeSpans ?? []) { + for (const rawSpan of ss.spans ?? []) { + spans.push({ + serviceName, + traceId: rawSpan.traceId, + spanId: rawSpan.spanId, + ...(rawSpan.parentSpanId ? { parentSpanId: rawSpan.parentSpanId } : {}), + name: rawSpan.name, + kind: rawSpan.kind, + startTimeUnixNano: rawSpan.startTimeUnixNano, + endTimeUnixNano: rawSpan.endTimeUnixNano, + attributes: decodeAttributes(rawSpan.attributes), + ...(rawSpan.links?.length + ? { + links: rawSpan.links.map((link: OtlpExportedLink) => ({ + traceId: link.traceId, + spanId: link.spanId, + attributes: decodeAttributes(link.attributes) + })) + } + : {}), + status: rawSpan.status + }); + } + } + } + res.writeHead(200, { "content-type": "application/json" }).end("{}"); + }); + }); + + await new Promise((resolve) => server.listen(0, "127.0.0.1", () => resolve())); + const { port } = server.address() as AddressInfo; + return { + endpoint: `http://127.0.0.1:${port}`, + payloads, + spans, + close: () => new Promise((resolve) => server.close((_err) => resolve())) + }; +} + +/** Synthesizes mock WorkflowInfo with customizable execution and identity parameters. */ +function makeWorkflowInfo(overrides: Partial & Pick): WorkflowInfo { + return { + taskQueue: "space-default", + namespace: "default", + firstExecutionRunId: overrides.runId, + originalExecutionRunId: overrides.runId, + startTime: new Date(workflowState.nowMs), + runStartTime: new Date(workflowState.nowMs), + taskTimeoutMs: 10_000, + attempt: 1, + historyLength: 4, + historySize: 512, + continueAsNewSuggested: false, + targetWorkerDeploymentVersionChanged: false, + searchAttributes: {}, + typedSearchAttributes: {} as never, + unsafe: { now: () => Date.now(), isReplaying: false }, + ...overrides + }; +} + +/** + * Runs a workflow body through the real workflow-observability-interceptors and + * activity-observability-interceptors, supporting nested child workflows, + * signals, concludeActivation, and activity retries. + */ +async function executeSimulatedWorkflow(options: { + info: WorkflowInfo; + args: unknown[]; + headers?: Headers; + spans: SpanExporter; + run: (runtime: { + scheduleActivity: ( + activityType: string, + activityArgs: unknown[], + fn: (attempt: number, headers: Headers) => Promise, + maxAttempts?: number + ) => Promise; + executeChild: ( + childInfo: WorkflowInfo, + childArgs: unknown[], + fn: (childHeaders: Headers, childInterceptors: ReturnType) => Promise + ) => Promise; + deliverSignal: (signalName: string, signalArgs: unknown[], headers?: Headers) => Promise; + concludeActivation: () => void; + continueAsNew: (nextArgs: unknown[], workflowType?: string) => Promise<{ headers: Headers; error: unknown }>; + }) => Promise; +}): Promise { + const previousInfo = workflowState.info; + const previousSinks = workflowState.sinks; + workflowState.info = options.info; + workflowState.sinks = createWorkflowObservabilitySinks({ spans: options.spans }); + const wfInterceptors = createWorkflowInterceptors(); + const inbound = wfInterceptors.inbound![0]!; + const outbound = wfInterceptors.outbound![0]!; + const internals = wfInterceptors.internals![0]!; + const actInterceptorFactory = createActivityObservabilityInterceptors({ spans: options.spans }); + + const runtime = { + async scheduleActivity( + activityType: string, + activityArgs: unknown[], + fn: (attempt: number, headers: Headers) => Promise, + maxAttempts = 1 + ): Promise { + const headers: Headers = {}; + return (await outbound.scheduleActivity!( + { + activityType, + args: activityArgs, + options: { startToCloseTimeout: "1 minute" }, + headers, + seq: 1 + }, + async (actInput) => { + let lastError: unknown; + for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + const actInterceptor = actInterceptorFactory({ + info: { + activityType, + activityId: `${attempt}`, + attempt, + workflowType: options.info.workflowType, + workflowExecution: { workflowId: options.info.workflowId, runId: options.info.runId }, + currentAttemptScheduledTimestampMs: Date.now() - 5 + }, + cancellationSignal: new AbortController().signal + } as never); + try { + return (await actInterceptor.inbound!.execute!( + { args: actInput.args, headers: actInput.headers }, + async () => fn(attempt, actInput.headers) + )) as R; + } catch (err) { + lastError = err; + if (attempt === maxAttempts) throw err; + } + } + throw lastError; + } + )) as R; + }, + + async executeChild( + childInfo: WorkflowInfo, + childArgs: unknown[], + fn: (childHeaders: Headers, childInterceptors: ReturnType) => Promise + ): Promise { + const childHeaders: Headers = {}; + const [, completed] = await outbound.startChildWorkflowExecution!( + { + workflowType: childInfo.workflowType, + options: { + workflowId: childInfo.workflowId, + args: childArgs, + taskQueue: childInfo.taskQueue, + cancellationType: 0 as never, + parentClosePolicy: 0 as never, + workflowIdReusePolicy: 0 as never + }, + headers: childHeaders, + seq: 1 + }, + async (childInput) => { + const savedInfo = workflowState.info; + try { + const result = await fn(childInput.headers, wfInterceptors); + return [Promise.resolve(childInfo.runId), Promise.resolve(result)]; + } finally { + workflowState.info = savedInfo; + } + } + ); + return (await completed) as R; + }, + + async deliverSignal(signalName: string, signalArgs: unknown[], headers: Headers = {}): Promise { + await inbound.handleSignal!({ signalName, args: signalArgs, headers }, async () => undefined); + }, + + concludeActivation(): void { + internals.concludeActivation!({ commands: [] }, (c) => c); + }, + + async continueAsNew(nextArgs: unknown[], workflowType = options.info.workflowType): Promise<{ headers: Headers; error: unknown }> { + const headers: Headers = {}; + try { + await outbound.continueAsNew!( + { + args: nextArgs, + headers, + options: { workflowType, taskQueue: options.info.taskQueue } + }, + async () => { + throw new ContinueAsNew({ + workflowType, + arguments: [], + headers, + taskQueue: options.info.taskQueue + }); + } + ); + throw new Error("unreachable"); + } catch (error) { + return { headers, error }; + } + } + }; + + try { + return (await inbound.execute!( + { args: options.args, headers: options.headers ?? {} }, + async () => options.run(runtime) + )) as T; + } finally { + workflowState.info = previousInfo; + workflowState.sinks = previousSinks; + } +} + +const collectors: Array<{ close: () => Promise }> = []; +const exporters: SpanExporter[] = []; + +afterEach(async () => { + await Promise.all(exporters.splice(0).map((e) => e.close())); + await Promise.all(collectors.splice(0).map((c) => c.close())); +}); + +describe("End-to-end OTLP trace continuity across API, workflows, child workflows, and activities", () => { + const secretPrompt = "PRIVATE PROMPT: summarize /home/operator/secret-plan.md using https://internal.example/api?token=sk-live-secret999"; + + it("exports a single W3C trace tree from API HTTP request -> workflow -> retried & failed activities", async () => { + const collector = await startFakeOtlpCollector(); + collectors.push(collector); + + const apiExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-api", + flushIntervalMs: 60_000 + }); + const workerExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + exporters.push(apiExporter, workerExporter); + + const { createHttpObservability } = await import("../../api/src/observability.js"); + const { + requestTraceContexts, + traceContextForRawRequest, + withTurnTraceparent, + workflowCorrelationMemo + } = await import("../../api/src/trace-context.js"); + + const httpObs = createHttpObservability({ serviceName: "space-api", spanExporter: apiExporter }); + const app = Fastify({ genReqId: () => "req:turn00000001" }); + app.addHook("onRequest", async (req) => { + requestTraceContexts.remember(String(req.id), traceContextForRawRequest(req.raw)); + httpObs.onRequest(req); + }); + app.addHook("onResponse", async (req, reply) => httpObs.onResponse(req, reply)); + + let capturedTurnInput: DummyTurnInput | null = null; + let capturedMemo: Record = {}; + app.post("/api/rooms/:roomId/turns", async (req) => { + const turn: DummyTurnInput = withTurnTraceparent({ + roomId: "room:alpha", + paneId: "pane:alpha", + prompt: secretPrompt, + artifactIds: [], + permissionMode: null, + collaborationMode: "default", + traceId: String(req.id), + agentRunId: "agent_run:turn-1", + dotId: "dot:researcher" + }); + capturedTurnInput = turn; + capturedMemo = workflowCorrelationMemo({ traceId: turn.traceId, traceparent: turn.traceparent ?? null }); + return { ok: true, traceId: turn.traceId }; + }); + + const inboundTraceId = "4bf92f3577b34da6a3ce929d0e0e4736"; + const inboundCallerSpanId = "00f067aa0ba902b7"; + try { + const res = await app.inject({ + method: "POST", + url: "/api/rooms/room:alpha/turns?secret=sk-live-query", + headers: { traceparent: `00-${inboundTraceId}-${inboundCallerSpanId}-01` } + }); + expect(res.statusCode).toBe(200); + } finally { + await app.close(); + httpObs.close(); + } + + expect(capturedTurnInput).not.toBeNull(); + const wfInfo = makeWorkflowInfo({ + workflowId: "workflow:codex-app-server:room:alpha:pane:alpha:req:turn00000001", + runId: "run:turn-wf-1", + workflowType: "codexAppServerTurnWorkflow", + memo: capturedMemo + }); + + await executeSimulatedWorkflow({ + info: wfInfo, + args: [capturedTurnInput!], + spans: workerExporter, + run: async ({ scheduleActivity }) => + scheduleActivity( + "runCodexAppServerTurn", + [capturedTurnInput!], + async (attempt) => { + if (attempt === 1) { + throw Object.assign(new Error(`transient failure for ${secretPrompt}`), { + code: "UPSTREAM_RETRYABLE_UNAVAILABLE" + }); + } + noteActivityFailureReason("CODEX_APP_SERVER_TURN_TIMEOUT"); + return { status: "FAILED", reasonCode: "CODEX_APP_SERVER_TURN_TIMEOUT", message: secretPrompt }; + }, + 2 + ) + }); + + await apiExporter.flush(); + await workerExporter.flush(); + + expect(collector.spans).toHaveLength(4); + const apiSpan = collector.spans.find((s) => s.serviceName === "space-api" && s.name === "POST /api/rooms/:roomId/turns")!; + const wfSpan = collector.spans.find((s) => s.serviceName === "space-worker" && s.name === "workflow codexAppServerTurnWorkflow")!; + const attempt1Span = collector.spans.find((s) => s.name === "activity runCodexAppServerTurn" && s.attributes["space.attempt"] === 1)!; + const attempt2Span = collector.spans.find((s) => s.name === "activity runCodexAppServerTurn" && s.attributes["space.attempt"] === 2)!; + + expect(apiSpan).toBeDefined(); + expect(wfSpan).toBeDefined(); + expect(attempt1Span).toBeDefined(); + expect(attempt2Span).toBeDefined(); + + // All 4 spans belong to the exact same W3C trace. + for (const s of [apiSpan, wfSpan, attempt1Span, attempt2Span]) { + expect(s.traceId).toBe(inboundTraceId); + } + + // Exact parent-child hierarchy: Inbound caller -> API request -> Workflow -> Activity attempts. + expect(apiSpan.parentSpanId).toBe(inboundCallerSpanId); + expect(wfSpan.parentSpanId).toBe(apiSpan.spanId); + expect(attempt1Span.parentSpanId).toBe(wfSpan.spanId); + expect(attempt2Span.parentSpanId).toBe(wfSpan.spanId); + expect(attempt1Span.spanId).not.toBe(attempt2Span.spanId); + + // Correlation & failure categorization on workflow and activity spans. + expect(wfSpan.attributes).toMatchObject({ + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.workflow_id": wfInfo.workflowId, + "space.run_id": wfInfo.runId, + "space.trace_id": "req:turn00000001", + "space.room_id": "room:alpha", + "space.pane_id": "pane:alpha", + "space.agent_run_id": "agent_run:turn-1", + "space.dot_id": "dot:researcher", + "space.outcome": "FAILED_RESULT", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT" + }); + expect(wfSpan.status.code).toBe(2); + + expect(attempt1Span.attributes).toMatchObject({ + "space.activity_type": "runCodexAppServerTurn", + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.trace_id": "req:turn00000001", + "space.attempt": 1, + "space.outcome": "FAILED", + "space.reason_code": "UPSTREAM_RETRYABLE_UNAVAILABLE", + "space.category": "UPSTREAM_UNAVAILABLE" + }); + expect(attempt1Span.status.code).toBe(2); + + expect(attempt2Span.attributes).toMatchObject({ + "space.activity_type": "runCodexAppServerTurn", + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.trace_id": "req:turn00000001", + "space.attempt": 2, + "space.outcome": "FAILED_RESULT", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT" + }); + expect(attempt2Span.status.code).toBe(2); + + // Zero sensitive data anywhere in exported OTLP JSON payloads. + const serialized = JSON.stringify(collector.payloads); + expect(serialized).not.toMatch(/PRIVATE PROMPT|secret-plan|internal\.example|sk-live/); + }); + + it("links API request -> roomAgentSupervisorWorkflow -> child codexAppServerTurnWorkflow -> activities and links background continuations (trace:bg-)", async () => { + const collector = await startFakeOtlpCollector(); + collectors.push(collector); + + const workerExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + exporters.push(workerExporter); + + const fixture = continuationFixture(); + const initialTraceId = "4bf92f3577b34da6a3ce929d0e0e4736"; + const apiSpanId = "a1b2c3d4e5f60718"; + const initialTraceparent = formatTraceparent(initialTraceId, apiSpanId, "01"); + + // Enqueue initial mission turn with W3C traceparent from the API request. + const initialTurn: DummyTurnInput = { + ...fixture.initial.queueItem.turn, + agentRunId: "agent_run:traced-1", + traceparent: initialTraceparent, + prompt: secretPrompt + }; + const initialItem: RoomAgentSupervisorQueueItem = { + missionId: fixture.initial.mission.id, + turn: initialTurn + }; + // Update the store's initial turn so prepareBackgroundContinuation reads the traced turn. + fixture.store.enqueueRoomAgentMission({ + requestId: "room_agent_request:traced-1", + clientRequestId: "traced-1", + content: secretPrompt, + supervisorWorkflowId: "room-supervisor:room:continuation", + childWorkflowId: "workflow:codex-app-server:room:continuation:pane:agent:req:traced1", + promptMessageId: "agent_msg:traced-user-1", + responseMessageId: "agent_msg:traced-assistant-1", + runId: "agent_run:traced-1", + queueItem: { + missionId: fixture.initial.mission.id, + turn: { + ...initialTurn, + agentRunId: "agent_run:traced-1", + agentUserMessageId: "agent_msg:traced-user-1", + agentAssistantMessageId: "agent_msg:traced-assistant-1" + } + } + }, initialTurn.traceId); + fixture.store.updateSpaceAgentRun("agent_run:traced-1", { status: "COMPLETED" }); + + const supervisorInfo = makeWorkflowInfo({ + workflowId: `room-supervisor:${fixture.room.id}`, + runId: "run:supervisor-1", + workflowType: "roomAgentSupervisorWorkflow", + memo: { + spaceTraceId: initialTurn.traceId, + traceparent: initialTraceparent, + w3cTraceId: initialTraceId, + missionId: fixture.initial.mission.id + } + }); + + const child1Info = makeWorkflowInfo({ + workflowId: `workflow:codex-app-server:${fixture.room.id}:${initialTurn.paneId}:${initialTurn.traceId}`, + runId: "run:child-turn-1", + workflowType: "codexAppServerTurnWorkflow", + parent: { workflowId: supervisorInfo.workflowId, runId: supervisorInfo.runId, namespace: "default" } + }); + + let preparedContinuation: RoomAgentSupervisorQueueItem | null = null; + + await executeSimulatedWorkflow({ + info: supervisorInfo, + args: [{ roomId: fixture.room.id, pending: [], monitoringMissionIds: [], interrupted: [], processedCount: 0 }], + spans: workerExporter, + run: async ({ deliverSignal, scheduleActivity, executeChild, concludeActivation }) => { + // 1. Signal initial mission turn into supervisor. + await deliverSignal("enqueueRoomAgentMission", [ + { + missionId: fixture.initial.mission.id, + turn: { ...initialTurn, agentRunId: "agent_run:traced-1" } + } + ]); + + // 2. Supervisor checks turn state, runs child codexAppServerTurnWorkflow, and enters monitoring. + await scheduleActivity("checkRoomAgentSupervisorTurn", [initialItem], async () => "READY"); + await executeChild( + child1Info, + [{ ...initialTurn, agentRunId: "agent_run:traced-1" }], + async (childHeaders) => + executeSimulatedWorkflow({ + info: child1Info, + args: [{ ...initialTurn, agentRunId: "agent_run:traced-1" }], + headers: childHeaders, + spans: workerExporter, + run: async (childRuntime) => + childRuntime.scheduleActivity( + "runCodexAppServerTurn", + [{ ...initialTurn, agentRunId: "agent_run:traced-1" }], + async () => ({ + status: "COMPLETED" as const, + message: "Allocated panes", + roomAgentOutcome: { + status: "VERIFIED" as const, + executedActionCount: 1, + pendingActionIds: [fixture.action.actionId], + statusReason: "Background action running" + } + }) + ) + }) + ); + // Activation concludes while mission enters monitoring (flushes turn 1 supervisor span). + concludeActivation(); + + // 3. Supervisor polls prepareRoomAgentBackgroundContinuation. + const prep = await scheduleActivity( + "prepareRoomAgentBackgroundContinuation", + [{ roomId: fixture.room.id, missionId: fixture.initial.mission.id, continuationId: "cont0001" }], + async () => prepareBackgroundContinuation( + { roomId: fixture.room.id, missionId: fixture.initial.mission.id, continuationId: "cont0001" }, + fixture.store + ) + ); + if (prep.state !== "READY") throw new Error("Expected READY continuation"); + preparedContinuation = prep.item; + + // 4. Supervisor executes the prepared background continuation turn (trace:bg-cont0001). + const child2Info = makeWorkflowInfo({ + workflowId: `workflow:codex-app-server:${fixture.room.id}:${prep.item.turn.paneId}:${prep.item.turn.traceId}`, + runId: "run:child-turn-bg-1", + workflowType: "codexAppServerTurnWorkflow", + parent: { workflowId: supervisorInfo.workflowId, runId: supervisorInfo.runId, namespace: "default" } + }); + + await scheduleActivity("checkRoomAgentSupervisorTurn", [prep.item], async () => "READY"); + await executeChild( + child2Info, + [prep.item.turn], + async (childHeaders) => + executeSimulatedWorkflow({ + info: child2Info, + args: [prep.item.turn], + headers: childHeaders, + spans: workerExporter, + run: async (childRuntime) => + childRuntime.scheduleActivity( + "runCodexAppServerTurn", + [prep.item.turn], + async () => ({ + status: "COMPLETED" as const, + message: "Verified", + roomAgentOutcome: { + status: "VERIFIED" as const, + executedActionCount: 1, + statusReason: "All done" + } + }) + ) + }) + ); + await scheduleActivity( + "finishRoomAgentSupervisorTurn", + [{ item: prep.item, status: "COMPLETED", statusReason: "All done" }], + async () => undefined + ); + } + }); + + await workerExporter.flush(); + + // Check initial mission turn hierarchy in initialTraceId: + const trace1Spans = collector.spans.filter((s) => s.traceId === initialTraceId); + const supSpan1 = trace1Spans.find((s) => s.name === "workflow roomAgentSupervisorWorkflow")!; + const childWfSpan1 = trace1Spans.find((s) => s.name === "workflow codexAppServerTurnWorkflow")!; + const checkSpan1 = trace1Spans.find((s) => s.name === "activity checkRoomAgentSupervisorTurn")!; + const turnActSpan1 = trace1Spans.find((s) => s.name === "activity runCodexAppServerTurn")!; + const prepSpan = trace1Spans.find((s) => s.name === "activity prepareRoomAgentBackgroundContinuation")!; + + expect(supSpan1).toBeDefined(); + expect(childWfSpan1).toBeDefined(); + expect(checkSpan1).toBeDefined(); + expect(turnActSpan1).toBeDefined(); + expect(prepSpan).toBeDefined(); + + // API span -> Supervisor workflow span -> Child turn workflow span -> Turn activity span. + expect(supSpan1.parentSpanId).toBe(apiSpanId); + expect(checkSpan1.parentSpanId).toBe(supSpan1.spanId); + expect(childWfSpan1.parentSpanId).toBe(supSpan1.spanId); + expect(childWfSpan1.attributes["space.parent_workflow_id"]).toBe(supervisorInfo.workflowId); + expect(turnActSpan1.parentSpanId).toBe(childWfSpan1.spanId); + + // Check background continuation trace (trace:bg-cont0001): + expect(preparedContinuation).not.toBeNull(); + expect(preparedContinuation!.turn.traceId).toBe("trace:bg-cont0001"); + const bgW3cTraceId = parseTraceparent(preparedContinuation!.turn.traceparent)?.traceId; + expect(bgW3cTraceId).toMatch(/^[0-9a-f]{32}$/); + expect(bgW3cTraceId).not.toBe(initialTraceId); + + const bgSpans = collector.spans.filter((s) => s.traceId === bgW3cTraceId); + const supSpanBg = bgSpans.find((s) => s.name === "workflow roomAgentSupervisorWorkflow")!; + const childWfSpanBg = bgSpans.find((s) => s.name === "workflow codexAppServerTurnWorkflow")!; + const turnActSpanBg = bgSpans.find((s) => s.name === "activity runCodexAppServerTurn")!; + const finishSpanBg = bgSpans.find((s) => s.name === "activity finishRoomAgentSupervisorTurn")!; + + expect(supSpanBg).toBeDefined(); + expect(childWfSpanBg).toBeDefined(); + expect(turnActSpanBg).toBeDefined(); + expect(finishSpanBg).toBeDefined(); + + expect(childWfSpanBg.parentSpanId).toBe(supSpanBg.spanId); + expect(turnActSpanBg.parentSpanId).toBe(childWfSpanBg.spanId); + expect(finishSpanBg.parentSpanId).toBe(supSpanBg.spanId); + + // Both supervisor and child workflow spans in the background continuation carry OTLP span links to the source trace! + expect(supSpanBg.links).toEqual([ + { + traceId: initialTraceId, + spanId: apiSpanId, + attributes: { + "space.link_type": "background_continuation", + "space.source_run_id": "agent_run:traced-1", + "space.source_trace_id": initialTurn.traceId, + "space.continuation_generation": 1 + } + } + ]); + expect(supSpanBg.attributes).toMatchObject({ + "space.trace_id": "trace:bg-cont0001", + "space.mission_id": fixture.initial.mission.id, + "space.agent_run_id": "agent_run:bg-cont0001", + "space.continuation_of_run_id": "agent_run:traced-1", + "space.continuation_generation": 1 + }); + }); + + it("propagates traceparent and OTLP span links across continueAsNew for long-running workflows (dotGoalWorkflow)", async () => { + const collector = await startFakeOtlpCollector(); + collectors.push(collector); + const workerExporter = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + exporters.push(workerExporter); + + const w3cTraceId = "11112222333344445555666677778888"; + const apiSpanId = "9999aaaabbbbcccc"; + const traceparent = formatTraceparent(w3cTraceId, apiSpanId, "01"); + const goalInput = { goalId: "dot_goal:goal12345", traceId: "req:goalcreate01", traceparent }; + + const run1Info = makeWorkflowInfo({ + workflowId: "dot-goal:dot_goal:goal12345", + runId: "run:goal-1", + workflowType: "dotGoalWorkflow" + }); + + let continuedHeaders: Headers = {}; + await expect( + executeSimulatedWorkflow({ + info: run1Info, + args: [goalInput], + spans: workerExporter, + run: async ({ scheduleActivity, continueAsNew }) => { + // Positional string arg (`dotGoalStep(goalId)`) receives traceparent via workflow outbound header! + await scheduleActivity("dotGoalStep", [goalInput.goalId], async () => ({ state: "STARTED", runId: "agent_run:step1" })); + const continued = await continueAsNew([goalInput]); + continuedHeaders = continued.headers; + throw continued.error; + } + }) + ).rejects.toBeInstanceOf(ContinueAsNew); + + const decodedCarry = decodeSpaceTraceHeader(continuedHeaders); + expect(decodedCarry).toMatchObject({ + traceId: "req:goalcreate01", + goalId: "dot_goal:goal12345", + continuedFromRunId: "run:goal-1", + continuedFromSpanId: deterministicSpanId(`workflow:${run1Info.workflowId}:${run1Info.runId}`) + }); + + // Run 2 starts after ContinueAsNew with the headers propagated by Temporal. + const run2Info = makeWorkflowInfo({ + workflowId: "dot-goal:dot_goal:goal12345", + runId: "run:goal-2", + workflowType: "dotGoalWorkflow", + continuedFromExecutionRunId: "run:goal-1" + }); + + await executeSimulatedWorkflow({ + info: run2Info, + args: [goalInput], + headers: continuedHeaders, + spans: workerExporter, + run: async ({ scheduleActivity }) => { + await scheduleActivity("dotGoalObserve", [goalInput.goalId, "agent_run:step1"], async () => ({ goalStatus: "COMPLETED" })); + return { goalStatus: "COMPLETED" }; + } + }); + + await workerExporter.flush(); + + const run1WfSpan = collector.spans.find((s) => s.name === "workflow dotGoalWorkflow" && s.attributes["space.run_id"] === "run:goal-1")!; + const stepActSpan = collector.spans.find((s) => s.name === "activity dotGoalStep")!; + const run2WfSpan = collector.spans.find((s) => s.name === "workflow dotGoalWorkflow" && s.attributes["space.run_id"] === "run:goal-2")!; + const observeActSpan = collector.spans.find((s) => s.name === "activity dotGoalObserve")!; + + expect(run1WfSpan.attributes["space.outcome"]).toBe("CONTINUED_AS_NEW"); + expect(run1WfSpan.status.code).toBe(0); + expect(run1WfSpan.parentSpanId).toBe(apiSpanId); + expect(stepActSpan.parentSpanId).toBe(run1WfSpan.spanId); + expect(stepActSpan.attributes["space.goal_id"]).toBe("dot_goal:goal12345"); + + expect(run2WfSpan.parentSpanId).toBe(apiSpanId); + expect(run2WfSpan.attributes["space.continued_from_run_id"]).toBe("run:goal-1"); + expect(run2WfSpan.links).toEqual([ + { + traceId: w3cTraceId, + spanId: run1WfSpan.spanId, + attributes: { + "space.link_type": "continue_as_new", + "space.source_run_id": "run:goal-1" + } + } + ]); + expect(observeActSpan.parentSpanId).toBe(run2WfSpan.spanId); + }); + + it("enforces the strict attribute allowlist and strips prompts, paths, URLs, error messages, and unknown keys", () => { + const dirty = { + "space.trace_id": "req:validtrace01", + "space.workflow_id": "workflow:valid", + "space.run_id": "run:valid", + "space.room_id": "room:valid", + "space.pane_id": "/home/operator/secret/file.ts", + "space.mission_id": "https://evil.example/callback?secret=1", + "space.agent_run_id": "sk-live-supersecretkey123456", + "space.action_id": "has spaces and user text", + "space.goal_id": "C:\\Users\\Admin\\secret.txt", + "space.dot_id": "dot:valid", + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.activity_type": "runCodexAppServerTurn", + "space.attempt": 1, + "space.outcome": "FAILED", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT", + "prompt": secretPrompt, + "error.message": "Connection refused to 10.0.0.8", + "http.url": "https://space.local/api/rooms?token=abc" + }; + + const sanitized = sanitizeSpanAttributes(dirty); + expect(Object.keys(sanitized).every((k) => ALLOWED_SPAN_ATTRIBUTE_KEYS.includes(k))).toBe(true); + expect(sanitized).toEqual({ + "space.trace_id": "req:validtrace01", + "space.workflow_id": "workflow:valid", + "space.run_id": "run:valid", + "space.room_id": "room:valid", + "space.dot_id": "dot:valid", + "space.workflow_type": "codexAppServerTurnWorkflow", + "space.activity_type": "runCodexAppServerTurn", + "space.attempt": 1, + "space.outcome": "FAILED", + "space.reason_code": "CODEX_APP_SERVER_TURN_TIMEOUT", + "space.category": "TIMEOUT" + }); + + const wire = JSON.stringify( + otlpTraceRequest( + [ + { + traceId: "4bf92f3577b34da6a3ce929d0e0e4736", + spanId: "00f067aa0ba902b7", + parentSpanId: null, + name: "workflow codexAppServerTurnWorkflow", + startTimeUnixNano: 1000n, + endTimeUnixNano: 2000n, + attributes: dirty as never, + error: true, + kind: 1, + links: [ + { + traceId: "11112222333344445555666677778888", + spanId: "9999aaaabbbbcccc", + attributes: { + "space.link_type": "background_continuation", + "space.source_run_id": "agent_run:1", + "prompt": secretPrompt + } as never + } + ] + } + ], + "space-worker" + ) + ); + expect(wire).not.toMatch(/PRIVATE PROMPT|secret|evil\.example|sk-live|10\.0\.0\.8|space\.local/); + expect(ALLOWED_SPAN_LINK_ATTRIBUTE_KEYS).toEqual([ + "space.link_type", + "space.source_run_id", + "space.source_trace_id", + "space.continuation_generation" + ]); + }); + + it("UNDETERMINISTIC REPLAY PROOF: sink callDuringReplay is false and workflow interceptor bundles cleanly in Temporal", async () => { + const recorded: unknown[] = []; + const spans: SpanExporter = { + record: (span) => recorded.push(span), + flush: async () => undefined, + close: async () => undefined, + stats: () => ({ exported: recorded.length, dropped: 0, failed: 0 }) + }; + const sinks = createWorkflowObservabilitySinks({ spans }); + expect(sinks.observability.recordWorkflowSpan.callDuringReplay).toBe(false); + + // Bundle workflow-observability-interceptors.ts + workflows.ts with Temporal's real WorkflowCodeBundler + // to prove zero disallowed Node built-ins or non-deterministic imports enter the V8 isolate. + const { bundleWorkflowCode } = await import("@temporalio/worker"); + const { existsSync } = await import("node:fs"); + const { fileURLToPath } = await import("node:url"); + const wfJs = fileURLToPath(new URL("../src/workflows.js", import.meta.url)); + const workflowsPath = existsSync(wfJs) ? wfJs : wfJs.replace(/\.js$/, ".ts"); + + const bundle = await bundleWorkflowCode({ + workflowsPath, + workflowInterceptorModules: [workflowObservabilityInterceptorsPath], + logger: { + trace: () => undefined, + debug: () => undefined, + info: () => undefined, + warn: () => undefined, + error: () => undefined, + log: () => undefined + } + }); + expect(bundle.code.length).toBeGreaterThan(1000); + }, 60_000); + + it("INTEGRATION TEST: runs a real workflow in TestWorkflowEnvironment.createTimeSkipping with stub activities, asserting trace ID continuity, parentage, retries, and OTLP export", async () => { + const { TestWorkflowEnvironment } = await import("@temporalio/testing"); + const { Worker, bundleWorkflowCode, DefaultLogger } = await import("@temporalio/worker"); + const { fileURLToPath } = await import("node:url"); + const { existsSync } = await import("node:fs"); + + const collector = await startFakeOtlpCollector(); + collectors.push(collector); + const spans = createOtlpSpanExporter({ + endpoint: collector.endpoint, + serviceName: "space-worker", + flushIntervalMs: 60_000 + }); + exporters.push(spans); + + const sinks = createWorkflowObservabilitySinks({ spans }); + const activityInterceptor = createActivityObservabilityInterceptors({ spans }); + + const traceId = "4bf92f3577b34da6a3ce929d0e0e4736"; + const parentSpanId = "00f067aa0ba902b7"; + const traceparent = `00-${traceId}-${parentSpanId}-01`; + + const env = await TestWorkflowEnvironment.createTimeSkipping(); + try { + const wfJs = fileURLToPath(new URL("../src/workflows.js", import.meta.url)); + const workflowsPath = existsSync(wfJs) ? wfJs : wfJs.replace(/\.js$/, ".ts"); + + const workflowBundle = await bundleWorkflowCode({ + workflowsPath, + workflowInterceptorModules: [workflowObservabilityInterceptorsPath], + logger: new DefaultLogger("ERROR") + }); + + let startedAttempts = 0; + const activities = { + recordDummyTurnStarted: async () => { + startedAttempts++; + if (startedAttempts === 1) { + throw new Error("Temporary failure for retry test"); + } + return { status: "STARTED" }; + }, + recordDummyTurnCompleted: async () => { + return { response: "done" }; + } + }; + + const worker = await Worker.create({ + connection: env.nativeConnection, + taskQueue: "otlp-integration-queue", + workflowBundle, + activities, + sinks, + interceptors: { + activity: [activityInterceptor] + } + }); + + const { Client } = await import("@temporalio/client"); + const client = new Client({ + connection: env.connection, + interceptors: { + workflow: [{ + startWithDetails: async (input, next) => { + encodeSpaceTraceHeader(input.headers, { traceId, traceparent }); + return next(input); + } + }] + } + }); + + await worker.runUntil(async () => { + const handle = await client.workflow.start("dummyTurnWorkflow", { + taskQueue: "otlp-integration-queue", + workflowId: "test-integration-workflow-1", + args: [{ providerId: "opencode", prompt: "hello" }] + }); + const res = await handle.result(); + expect(res.response).toBe("done"); + }); + + await spans.flush(); + } finally { + await env.teardown(); + } + + expect(collector.spans.length).toBe(4); + expect(collector.spans.every((s) => s.traceId === traceId)).toBe(true); + + const wfSpan = collector.spans.find((s) => s.name === "workflow dummyTurnWorkflow"); + expect(wfSpan).toBeDefined(); + expect(wfSpan!.parentSpanId).toBe(parentSpanId); + expect(wfSpan!.status.code).toBe(0); + + const act1Attempts = collector.spans.filter((s) => s.name === "activity recordDummyTurnStarted"); + expect(act1Attempts.length).toBe(2); + expect(act1Attempts[0]!.parentSpanId).toBe(wfSpan!.spanId); + expect(act1Attempts[1]!.parentSpanId).toBe(wfSpan!.spanId); + expect(act1Attempts[0]!.attributes["space.attempt"]).toBe(1); + expect(act1Attempts[0]!.attributes["space.outcome"]).toBe("FAILED"); + expect(act1Attempts[0]!.status.code).toBe(2); + expect(act1Attempts[1]!.attributes["space.attempt"]).toBe(2); + expect(act1Attempts[1]!.attributes["space.outcome"]).toBe("COMPLETED"); + expect(act1Attempts[1]!.status.code).toBe(0); + + const act2 = collector.spans.find((s) => s.name === "activity recordDummyTurnCompleted"); + expect(act2).toBeDefined(); + expect(act2!.parentSpanId).toBe(wfSpan!.spanId); + expect(act2!.attributes["space.outcome"]).toBe("COMPLETED"); + expect(act2!.status.code).toBe(0); + }, 120_000); +}); + diff --git a/docs/configuration-reference.md b/docs/configuration-reference.md index e432af4c6..441ef3e95 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`: optional plain `http://` or `https://` collector base URL (no credentials or query string). When set, `space-api`, `space-worker`, `space-memory-maintenance-worker`, and `space-memory-mutation-worker` export W3C-correlated spans over OTLP/HTTP JSON (`POST /v1/traces`) linking `API request -> workflow -> child workflow -> activity attempt`, plus OTLP span links across `continueAsNew` and background continuations (`trace:bg-`). Unset by default (zero network export when unset). See [Observability and error semantics](quality/observability.md). +- `SPACE_WORKER_METRICS_PORT` / `SPACE_WORKER_METRICS_HOST`: optional loopback (`127.0.0.1` or `::1`) port for the worker's Prometheus `/metrics` endpoint. Unset by default. +- `SPACE_WORKER_TEMPORAL_METRICS_BIND`: optional loopback bind address (for example `127.0.0.1:9464`) for the Temporal Core SDK Prometheus exporter. Unset by default. + diff --git a/docs/quality/observability.md b/docs/quality/observability.md index d5cd7080e..d053dd028 100644 --- a/docs/quality/observability.md +++ b/docs/quality/observability.md @@ -217,11 +217,12 @@ Without the endpoint, every activity still writes `activity.started` and 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. +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`). +The workflow interceptor propagates valid `traceparent` values through workflow +headers, memo fields and workflow inputs. It exports workflow and child-workflow +spans, and the activity interceptor exports activity-attempt spans under the +propagated parent context. - **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,17 +234,18 @@ 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 business logic is unchanged.** The worker loads + `workflow-observability-interceptors.ts` through `workflowModules`. The + interceptor propagates trace context to activities and child workflows. ### 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 workflow bundle now includes `workflow-observability-interceptors.ts`. +Workflow span export uses a Temporal sink with `callDuringReplay: false`, so +replay does not re-export spans or perform exporter I/O. 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 contract differences +include: - the new `observability` module; - the optional `traceparent` turn-input field; - the optional file search and readiness fields. @@ -251,6 +253,13 @@ module numbers. The only differences are in `@space/contracts`: A history without `traceparent` parses exactly as before. Activity inputs are not part of Temporal's determinism check. +### Deployment rollout order + +When deploying this change, **deploy workers before the API**. Workers must be +updated with the updated input schemas and workflow interceptors to accept +`traceparent` arguments and headers before the API starts transmitting them, +preventing schema validation rejections during rolling deployments. + 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 @@ -272,7 +281,8 @@ Tests that prove these properties: | --- | --- | | `packages/runtime/tests/metrics-registry.test.ts` | Unsafe label values never appear | | `apps/api/tests/observability-metrics.test.ts` | Secrets, prompts, paths and user IDs sent in URLs, headers and bodies never reach `/metrics` | -| `apps/worker/tests/activity-observability.test.ts` | Prompts and thrown messages never reach worker logs or metrics | +| `apps/worker/tests/activity-observability.test.ts` | Prompts and thrown messages never reach worker logs, metrics or OTLP spans | +| `apps/worker/tests/otlp-trace-continuity.test.ts` | End-to-end OTLP trace tree (`API -> workflow -> child workflow -> activity`), retries/failures, `continueAsNew` and `trace:bg-` span links, replay determinism (`callDuringReplay: false`), and allowlist redaction | | `apps/api/tests/trace-correlation.test.ts` | The trace lookup omits messages and metadata | | `apps/api/tests/observability-fault-injection.test.ts` | Checks all of the above end to end | @@ -282,16 +292,18 @@ No credential or transcript export was added. | File | Covers | | --- | --- | -| `packages/contracts/tests/observability.test.ts` | Categories, `traceparent` parsing, additive contract fields | +| `packages/contracts/tests/observability.test.ts` | Categories, `traceparent` parsing, deterministic W3C trace/span IDs, additive contract fields | | `packages/runtime/tests/metrics-registry.test.ts` | Registry format, label safety, series cap, process metrics | | `packages/db/tests/trace-records.test.ts` | Bounded, parameterized PostgreSQL trace lookup | | `packages/db/tests/room-agent-trace-postgres.test.ts` | Migrations 0206 and 0258 on a disposable PostgreSQL: mission trace and first error code, a continuation found from its source run, CLI sessions by the request that allocated them | | `apps/api/tests/cli-terminal-control.test.ts` | "links the request that allocated a CLI session …": the session's request id finds it, its task and revision in `/api/admin/traces/:traceId`; a rejoin does not | +| `apps/api/tests/otlp-span-exporter.test.ts` | OTLP/HTTP JSON span export, span links, attribute allowlist sanitization, and parent span continuity | | `apps/api/tests/trace-correlation.test.ts` | W3C context, log bindings, starter memo and input, Room Agent propagation, start-failure category, lookup projection | | `apps/api/tests/observability-metrics.test.ts` | New API metrics, `/metrics` access rule, redaction, CLI disconnect and reconnect counters | | `apps/api/src/__tests__/files-search-errors.test.ts` | `EMPTY_RESULT`/`OK` and each failure code | | `apps/api/tests/files-search-routes.test.ts` | The HTTP error envelope | -| `apps/worker/tests/activity-observability.test.ts` | Interceptor records, outcomes, categories, metrics, loopback endpoint | +| `apps/worker/tests/activity-observability.test.ts` | Interceptor records, outcomes, categories, metrics, loopback endpoint, workflow header parent continuity | +| `apps/worker/tests/otlp-trace-continuity.test.ts` | End-to-end OTLP trace hierarchy across API, workflows, child workflows, and activities, `continueAsNew` & `trace:bg-` links, Temporal workflow bundle determinism | | `apps/api/tests/observability-fault-injection.test.ts` | Four injected executor faults and one Temporal start failure, end to end | ## Not covered @@ -311,11 +323,13 @@ No credential or transcript export was added. mission's code is the cause when no event or agent run carries one. Supervisor failures record a code too since 2026-10-16 (previous bullet). - **Background continuations keep their own trace (followed since - 2026-10-16).** A continuation still runs as `trace:bg-`, because the - turn's workflow ID embeds its trace ID. The trace lookup now follows each + 2026-10-16, OTLP-linked since 2026-10-17).** A continuation runs as `trace:bg-`, because the + turn's workflow ID embeds its trace ID. The trace lookup follows each 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 + marks it with `continuationOf`. In OTLP trace export, each background continuation + gets its own bounded W3C trace and links back to the source turn's span via + an OTLP span link (`space.link_type = "background_continuation"`). Its events and logs remain under its own trace ID. 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 @@ -341,9 +355,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. +- **Hosts emit no spans.** The OTLP exporter sends API SERVER spans, workflow + and child-workflow spans, and worker activity-attempt spans across `space-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. - **Worker endpoints are not scraped yet.** Stable's `deploy/space-worker.service` diff --git a/package-lock.json b/package-lock.json index e0cc34a96..256594aea 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", @@ -11641,7 +11661,7 @@ } }, "packages/run-spaceapp": { - "version": "1.0.35", + "version": "1.1.6", "cpu": [ "x64", "arm64" diff --git a/packages/contracts/src/observability.ts b/packages/contracts/src/observability.ts index 0090c5e90..9f05c0122 100644 --- a/packages/contracts/src/observability.ts +++ b/packages/contracts/src/observability.ts @@ -111,6 +111,7 @@ export function parseTraceparent(value: unknown): ParsedTraceparent | null { return { traceId, parentId, flags }; } +/** Formats a W3C traceparent header string from traceId, spanId, and optional flags. */ export function formatTraceparent(traceId: string, spanId: string, flags = "01"): string { return `00-${traceId}-${spanId}-${flags}`; } @@ -120,6 +121,46 @@ export function traceIdFromTraceparent(value: unknown): string | null { return parseTraceparent(value)?.traceId ?? null; } +/** Deterministic 64-bit mixer hash producing two non-zero 32-bit unsigned integers from a seed string. */ +function mix64(seed: string, salt: number): [number, number] { + let h1 = (0xdeadbeef ^ seed.length ^ salt) >>> 0; + let h2 = (0x41c6ce57 ^ seed.length ^ Math.imul(salt, 0x9e3779b1)) >>> 0; + for (let i = 0; i < seed.length; i += 1) { + const ch = seed.charCodeAt(i); + h1 = Math.imul(h1 ^ ch, 2654435761) >>> 0; + h2 = Math.imul(h2 ^ ch, 1597334677) >>> 0; + } + h1 = (Math.imul(h1 ^ (h1 >>> 16), 2246822507) ^ Math.imul(h2 ^ (h2 >>> 13), 3266489909)) >>> 0; + h2 = (Math.imul(h2 ^ (h2 >>> 16), 2246822507) ^ Math.imul(h1 ^ (h1 >>> 13), 3266489909)) >>> 0; + return [h1 >>> 0, (h2 >>> 0) || 1]; +} + +/** + * Pure, replay-safe 16-hex span id derived deterministically from a workflow/activity + * identity string without touching Math.random, uuid4 or Node crypto. + */ +export function deterministicSpanId(seed: string): string { + const [hi, lo] = mix64(seed, 1); + return `${hi.toString(16).padStart(8, "0")}${lo.toString(16).padStart(8, "0")}`; +} + +/** + * Pure, replay-safe 32-hex W3C trace id derived deterministically from a Space + * correlation or workflow identity when no inbound W3C traceparent was supplied. + */ +export function deterministicTraceId(seed: string): string { + const [a, b] = mix64(seed, 1); + const [c, d] = mix64(seed, 2); + return `${a.toString(16).padStart(8, "0")}${b.toString(16).padStart(8, "0")}${c.toString(16).padStart(8, "0")}${d.toString(16).padStart(8, "0")}`; +} + +/** + * Pure, deterministic W3C traceparent derived from a stable trace seed and span seed. + */ +export function deterministicTraceparent(traceSeed: string, spanSeed = traceSeed, flags = "01"): string { + return formatTraceparent(deterministicTraceId(traceSeed), deterministicSpanId(spanSeed), flags); +} + /** * Admin trace lookup (GET /api/admin/traces/:traceId). Identifiers, codes, * statuses and timestamps only: event messages, payload metadata, prompts and diff --git a/packages/contracts/src/schemas.ts b/packages/contracts/src/schemas.ts index 7fb3c217a..aee77ead4 100644 --- a/packages/contracts/src/schemas.ts +++ b/packages/contracts/src/schemas.ts @@ -6909,7 +6909,9 @@ export const dummyTurnInputSchema = z.object({ roomAgentOperatorRequest: z.string().trim().min(1).max(4000).optional(), roomAgentContinuation: z.object({ sourceRunId: idSchema, - generation: z.number().int().min(1).max(16) + generation: z.number().int().min(1).max(16), + sourceTraceId: requestIdSchema.optional(), + sourceTraceparent: traceparentSchema.optional() }).optional(), agentUserMessageId: idSchema.optional(), agentAssistantMessageId: idSchema.optional(), @@ -7704,7 +7706,8 @@ export const memoryConsolidationDetailSchema = z export const memoryConsolidationWorkflowInputSchema = z .object({ runId: idSchema, - traceId: idSchema + traceId: idSchema, + traceparent: traceparentSchema.optional() }) .strict(); @@ -7916,7 +7919,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 +7992,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..4371eeb96 100644 --- a/packages/runtime/src/otlp-span-exporter.ts +++ b/packages/runtime/src/otlp-span-exporter.ts @@ -1,16 +1,24 @@ /** - * Optional OTLP/HTTP (JSON) export of API server spans and worker activity spans. Off unless - * SPACE_OTEL_EXPORTER_OTLP_ENDPOINT names a collector base URL (spans go to - * `/v1/traces`). No dependency, no collector required: Space logs stay - * joinable by W3C trace id without it. + * Optional OTLP/HTTP (JSON) export of API server spans and worker workflow/activity + * spans. Off unless SPACE_OTEL_EXPORTER_OTLP_ENDPOINT names a collector base URL + * (spans go to `/v1/traces`). No dependency, no collector required: Space logs + * stay joinable by W3C trace id without it. * - * Spans carry only the method, route template, status code, service name and - * Space request id; never URLs with parameters, headers, bodies or user data. + * Spans carry only allowlisted correlation ids, HTTP route templates, status codes, + * workflow/activity types, attempts, bounded outcomes, bounded error codes and + * failure categories; never prompts, model output, file paths, URLs with parameters, + * headers, bodies, error messages or user text. * Export is best effort: a bounded queue drops (and counts) spans instead of * growing, each POST has a timeout, and a failing collector never affects a - * request. + * request or workflow. */ +export interface SpanLink { + traceId: string; + spanId: string; + attributes?: Readonly>; +} + export interface ServerSpan { traceId: string; spanId: string; @@ -20,8 +28,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 workflows and activities use INTERNAL (1). */ kind?: number; + /** Optional OTLP span links (for continueAsNew or background continuation traces). */ + links?: readonly SpanLink[]; } export interface SpanExporterStats { @@ -49,10 +59,102 @@ export interface OtlpSpanExporterOptions { timeoutMs?: number; } +/** Helper regex factory for fixed-length hex identifier matching. */ const hexId = (length: number) => new RegExp(`^[0-9a-f]{${length}}$`); const traceIdPattern = hexId(32); const spanIdPattern = hexId(16); +const httpMethodPattern = /^[A-Z]{3,12}$/; +const httpRoutePattern = /^(?:\/[A-Za-z0-9/_:.*-]{0,239}|unmatched)$/; +const boundedCodePattern = /^[A-Z][A-Z0-9_]{1,95}$/; +const boundedTypePattern = /^[A-Za-z][A-Za-z0-9_:-]{1,127}$/; +const correlationIdPattern = /^[A-Za-z0-9][A-Za-z0-9:._-]{0,239}$/; +const forbiddenTokenPattern = /^(?:https?:|file:|ws:|wss:|ftp:|sk-[A-Za-z0-9_-]{6,})/i; + +type AttributeKind = "int" | "http_method" | "http_route" | "code" | "type" | "id"; + +const spanAttributeRules: Readonly> = { + "http.request.method": "http_method", + "http.route": "http_route", + "http.response.status_code": "int", + "space.request_id": "id", + "space.trace_id": "id", + "space.parent_trace_id": "id", + "space.workflow_id": "id", + "space.run_id": "id", + "space.parent_workflow_id": "id", + "space.parent_run_id": "id", + "space.continued_from_run_id": "id", + "space.room_id": "id", + "space.pane_id": "id", + "space.mission_id": "id", + "space.agent_run_id": "id", + "space.continuation_of_run_id": "id", + "space.continuation_generation": "int", + "space.action_id": "id", + "space.goal_id": "id", + "space.dot_id": "id", + "space.turn_id": "id", + "space.change_set_id": "id", + "space.consolidation_run_id": "id", + "space.workflow_type": "type", + "space.activity_type": "type", + "space.task_queue": "type", + "space.attempt": "int", + "space.outcome": "code", + "space.error_code": "code", + "space.reason_code": "code", + "space.category": "code", + "space.duration_ms": "int", + "space.queue_ms": "int" +}; + +const spanLinkAttributeRules: Readonly> = { + "space.link_type": "type", + "space.source_run_id": "id", + "space.source_trace_id": "id", + "space.continuation_generation": "int" +}; + +export const ALLOWED_SPAN_ATTRIBUTE_KEYS = Object.freeze(Object.keys(spanAttributeRules)); +export const ALLOWED_SPAN_LINK_ATTRIBUTE_KEYS = Object.freeze(Object.keys(spanLinkAttributeRules)); + +/** Validates whether an attribute value meets the structural and safety constraints for its kind. */ +function isAllowedAttributeValue(kind: AttributeKind, value: unknown): value is string | number { + if (kind === "int") { + return typeof value === "number" && Number.isInteger(value) && Number.isFinite(value) && value >= 0; + } + if (typeof value !== "string" || forbiddenTokenPattern.test(value)) return false; + switch (kind) { + case "http_method": + return httpMethodPattern.test(value); + case "http_route": + return httpRoutePattern.test(value); + case "code": + return boundedCodePattern.test(value); + case "type": + return boundedTypePattern.test(value); + case "id": + return correlationIdPattern.test(value); + } +} + +/** Filters span attributes against the explicit allowlist and value formats. */ +export function sanitizeSpanAttributes( + attributes: Readonly> | undefined, + rules: Readonly> = spanAttributeRules +): Record { + const out: Record = {}; + if (!attributes) return out; + for (const [key, value] of Object.entries(attributes)) { + const kind = rules[key]; + if (kind && isAllowedAttributeValue(kind, value)) { + out[key] = value; + } + } + return out; +} + /** The collector base URL, or null when export is off or the value is unsafe. */ export function otlpTracesUrl(value: string | undefined): string | null { if (!value?.trim()) return null; @@ -66,44 +168,74 @@ export function otlpTracesUrl(value: string | undefined): string | null { return `${url.origin}${url.pathname.replace(/\/+$/, "")}/v1/traces`; } +/** Converts a scalar attribute value to an OTLP typed value object. */ function attributeValue(value: string | number) { return typeof value === "number" && Number.isInteger(value) ? { intValue: String(value) } : { stringValue: String(value) }; } +/** Serializes span links ensuring trace and span ID invariants and attribute allowlists. */ +function serializeSpanLinks(links: readonly SpanLink[] | undefined) { + if (!links?.length) return []; + const out: Array<{ traceId: string; spanId: string; attributes?: Array<{ key: string; value: { intValue: string } | { stringValue: string } }> }> = []; + for (const link of links) { + if (!traceIdPattern.test(link.traceId) || !spanIdPattern.test(link.spanId)) continue; + const sanitized = sanitizeSpanAttributes(link.attributes, spanLinkAttributeRules); + const attrs = Object.entries(sanitized).map(([key, value]) => ({ key, value: attributeValue(value) })); + out.push({ + traceId: link.traceId, + spanId: link.spanId, + ...(attrs.length ? { attributes: attrs } : {}) + }); + } + return out; +} + +/** Builds an OTLP/HTTP JSON request payload from a batch of server spans. */ export function otlpTraceRequest(spans: readonly ServerSpan[], serviceName: string) { return { resourceSpans: [{ 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 sanitized = sanitizeSpanAttributes(span.attributes); + const validLinks = serializeSpanLinks(span.links); + 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(sanitized).map(([key, value]) => ({ key, value: attributeValue(value) })), + ...(validLinks.length ? { links: validLinks } : {}), + // STATUS_CODE_ERROR or STATUS_CODE_UNSET + status: { code: span.error ? 2 : 0 } + }; + }) }] }] }; } +/** Creates a background OTLP span exporter with bounded memory, timeout protection, and batching. */ export function createOtlpSpanExporter(options: OtlpSpanExporterOptions): SpanExporter { const url = otlpTracesUrl(options.endpoint); if (!url) throw new Error("Invalid OTLP endpoint."); const send: FetchLike = options.fetch ?? ((target, init) => fetch(target, init)); - const batchSize = options.batchSize ?? 256; + const configuredBatchSize = options.batchSize ?? 256; + if (!Number.isFinite(configuredBatchSize) || configuredBatchSize <= 0) { + throw new Error("batchSize must be a positive finite number."); + } + const batchSize = Math.max(1, Math.floor(configuredBatchSize)); const maxQueue = options.maxQueue ?? 4096; const timeoutMs = options.timeoutMs ?? 3000; const queue: ServerSpan[] = []; const stats: SpanExporterStats = { exported: 0, dropped: 0, failed: 0 }; let inFlight: Promise | null = null; + /** Flushes a single batch of queued spans to the OTLP traces HTTP endpoint. */ const flushOnce = async () => { const batch = queue.splice(0, batchSize); if (!batch.length) return; @@ -159,3 +291,4 @@ export function otlpSpanExporterFromEnv(env: NodeJS.ProcessEnv, serviceName: str const endpoint = env.SPACE_OTEL_EXPORTER_OTLP_ENDPOINT; return otlpTracesUrl(endpoint) ? createOtlpSpanExporter({ endpoint: endpoint!, serviceName }) : null; } +