From ac3b450e4b31fe0dbf7daac2d8d3e54825a802c9 Mon Sep 17 00:00:00 2001 From: Rickylabs Date: Sat, 11 Jul 2026 14:18:24 +0200 Subject: [PATCH] fix(telemetry): enforce Flow-B correlation outcome floor --- .../worklog.md | 133 ++++++++++++++++++ .../gates/scaffold/validate-flow-b-traces.ts | 6 +- .../telemetry/src/instrumentation/queue.ts | 16 +++ .../telemetry/src/instrumentation/worker.ts | 7 + .../src/runtime/trigger-runtime-processor.ts | 11 +- plugins/workers/worker/job-dispatcher.ts | 1 + plugins/workers/worker/job-execution.ts | 5 +- 7 files changed, 174 insertions(+), 5 deletions(-) create mode 100644 .llm/runs/fix-599-flowb-attribute-floor--codex/worklog.md diff --git a/.llm/runs/fix-599-flowb-attribute-floor--codex/worklog.md b/.llm/runs/fix-599-flowb-attribute-floor--codex/worklog.md new file mode 100644 index 000000000..66b3ea913 --- /dev/null +++ b/.llm/runs/fix-599-flowb-attribute-floor--codex/worklog.md @@ -0,0 +1,133 @@ +# #599 Flow-B correlation/outcome floor — implementation worklog + +## Identity + +- Branch/worktree: `fix/599-flowb-attribute-floor` / `/home/codex/repos/ns-wt-599` +- Implementation lane: WSL Codex under beta-7 shipping orchestrator session `df71d36c` +- PR lifecycle and final slice review: owned by the orchestrator; this lane will not open or update a PR. +- PLAN-EVAL: owner-waived by the slice brief (drift D1); this short plan is the authorized substitute before implementation. + +## Research + +- Issue #599's four acceptance boxes are the contract. +- TC-6/TC-7 define the canonical cross-domain pair as `netscript.correlation.id` and `netscript.outcome`; legacy/domain status keys remain compatibility attributes. +- `@netscript/telemetry` is doctrine Archetype 2 (Integration), current verdict **Refactor**. `plugins/workers` is Archetype 5 (Plugin), current verdict **Refactor**; its runtime composes core worker and telemetry contracts rather than redefining them. +- Runtime behavior is also in scope: queue spans are emitted by telemetry's `TracedQueue`, worker job spans by workers-core instrumentation, and trigger product spans by the triggers plugin runtime. +- Public-surface impact: **none**. The slice adds values to existing attribute maps and preserves an existing optional `JobMessage.correlationId`; it adds no export, entrypoint, CLI command, public type, or function signature. +- Relevant doctrine risks: AP-9 (do not introduce a speculative helper), AP-13/AP-25 (do not add effects), and existing telemetry/workers structural debt. The focused additive changes create or deepen no architecture debt. + +## Design + +### Public surface + +No new public API. Existing constants/builders are reused where practical; compatibility attributes remain emitted. + +### Domain vocabulary + +- Correlation floor: `netscript.correlation.id` +- Outcome floor: `netscript.outcome` +- Product spans: trigger ingress/detect/process, queue enqueue/dequeue, and job execute +- Outcomes use the existing operation lifecycle values (`pending`/`running`/`completed`/`failed` as applicable). + +### Ports + +No new ports. Existing OpenTelemetry span contracts and existing `JobMessage` transport contract are consumed. + +### Constants + +Reuse `NetScriptCorrelationAttributes.CORRELATION_ID` / existing worker telemetry constants and the established `netscript.outcome` convention. Do not create a parallel convention. + +### Plan and commit slice + +1. Add focused regression assertions, implement the canonical pair on product spans, preserve `correlationId` in `executeDenoJob()`, and tighten the existing T8 validator sets. Prove with scoped wrapper check/lint, telemetry and workers package tests, and `behavior.otel.traces`; record verbatim command outputs here, then commit and push with the mandated explicit refspec. + +### Deferred scope + +- No telemetry package restructuring, worker executor decomposition, alias removal, or full `scaffold.runtime` run; those are outside #599 and merge-readiness remains with the orchestrator. +- IMPL-EVAL and substantive Tier-A slice sign-off remain with the separate orchestrator session. + +### Contributor path + +Start at the domain attribute maps/builders, follow their existing instrumentation call sites, then update the single Flow-B trace validator so product emission and end-to-end acceptance stay coupled. + +## Evidence + +- Acceptance 1: `traceJobExecution()` now emits canonical `netscript.outcome` at running/completed/failed transitions while retaining legacy `job.status`. +- Acceptance 2: trigger ingress/detect/process and queue enqueue/dequeue spans emit `netscript.correlation.id` plus `netscript.outcome`. Trigger correlation is the event ID already propagated to the worker message; queue correlation is read from that existing message envelope. +- Acceptance 3: `processWorkerJob()` threads the received `correlationId` through `executeWorkerJob()` and `executeDenoJob()` copies it into the rebuilt in-process `JobMessage`, allowing `WorkerPool` to pass it to the callback context. +- Acceptance 4: the existing T8 validator's single `correlatedSpans` set now contains ingress, process, enqueue, dequeue, execute, callback boundary, and fan-in; the same set is checked for canonical outcome. No fork was created. +- No public exports or API contracts were added. No `deno.lock` change is owned by this slice. + +## Gate results + +### Scoped check + +Command: + +```text +deno run --allow-read --allow-run .llm/tools/run-deno-check.ts --root packages/telemetry --root plugins/workers --root plugins/triggers --root packages/cli/e2e --ext ts,tsx +``` + +Verbatim output: + +```json +{"source":{"mode":"selection","cwd":"/home/codex/repos/ns-wt-599"},"command":"deno check --quiet --unstable-kv ","selection":{"filesSelected":335,"batches":3,"failedBatches":0},"summary":{"totalOccurrences":0,"uniqueOccurrences":0,"uniqueCodes":0,"uniquePaths":0},"groups":[]} +``` + +### Scoped lint + +Command: + +```text +deno run --allow-read --allow-run .llm/tools/run-deno-lint.ts --root packages/telemetry --root plugins/workers --root plugins/triggers --root packages/cli/e2e --ext ts,tsx +``` + +Verbatim output: + +```json +{"source":{"mode":"command","cwd":"/home/codex/repos/ns-wt-599","exitCode":0},"selection":{"filesSelected":335,"batches":2},"summary":{"totalOccurrences":0,"uniqueOccurrences":0,"uniqueRules":0,"uniquePaths":0},"groups":[]} +``` + +### Package tests + +Commands and verbatim verdict lines: + +```text +(cd packages/telemetry && deno task test) +ok | 50 passed | 0 failed (1s) + +(cd plugins/workers && deno task test) +ok | 24 passed | 0 failed (2s) +``` + +Full verbatim command output was captured during the run at `.llm/tmp/599-telemetry-test.log` and `.llm/tmp/599-workers-test.log` (scratch evidence, intentionally not committed). + +### T8 Flow-B gate + +Exact gate ID: `behavior.otel.traces`. The CLI's executable single-gate form is: + +```text +deno task e2e:cli gate scaffold.runtime behavior.otel.traces +``` + +Verbatim output: + +```text +Task e2e:cli deno run --allow-all packages/cli/e2e/cli.ts 'gate' 'scaffold.runtime' 'behavior.otel.traces' +Running scaffold.runtime +> behavior.otel.traces: Validate OTEL trace chain via Dashboard API + FAILED 42ms +Summary: passed=0 failed=1 +error: Uncaught (in promise) RemoteError: CLI E2E gate failed +``` + +The structured gate log proves this was a prerequisite/environment failure before any trace assertion: `NotFound ... readfile '.../.netscript/e2e/aspire-start.json'`. A standalone `gate` invocation creates a fresh scratch project but does not run the fixture/Aspire setup gates. The brief forbids this lane from running full `scaffold.runtime`; Tier-A/orchestrator must execute T8 against its prepared runtime during merge readiness. + +## Drift + +- D1: PLAN-EVAL/external evaluator dispatch is owner-waived in the slice brief; implementation proceeds after this recorded plan. +- D2: the focused T8 gate cannot execute standalone because its fresh scratch project has no `aspire-start.json` or live generated runtime. The gate implementation compiled, but live trace validation remains an orchestrator prerequisite; no product assertion failed. + +## Reconcile + +- Issue #599 remains open and PR lifecycle belongs to the beta-7 orchestrator. This slice implements all four boxes; Tier-A retains substantive review and closing-keyword responsibility. diff --git a/packages/cli/e2e/src/application/gates/scaffold/validate-flow-b-traces.ts b/packages/cli/e2e/src/application/gates/scaffold/validate-flow-b-traces.ts index 3d52abf73..e0f2a9f6e 100644 --- a/packages/cli/e2e/src/application/gates/scaffold/validate-flow-b-traces.ts +++ b/packages/cli/e2e/src/application/gates/scaffold/validate-flow-b-traces.ts @@ -95,7 +95,7 @@ function validateFlowB(traces: readonly TelemetryTrace[]): void { 'fan-in link preserves per-message attributes through the SDK provider', ); - const correlatedSpans = [execute, fanIn]; + const correlatedSpans = [ingress, process, enqueue, dequeue, execute, callbackBoundary, fanIn]; for (const span of correlatedSpans) { tcAssert( 'TC-6/TC-7', @@ -106,10 +106,10 @@ function validateFlowB(traces: readonly TelemetryTrace[]): void { tcAssert( 'TC-6/TC-7', new Set(correlatedSpans.map((span) => span.attributes['netscript.correlation.id'])).size === 1, - 'job.execute and fan-in consumer carry the same correlation id', + 'Flow-B product and boundary spans carry the same correlation id', ); - for (const span of [callbackBoundary, fanIn]) { + for (const span of correlatedSpans) { tcAssert( 'TC-6/TC-7', hasOutcome(span), diff --git a/packages/telemetry/src/instrumentation/queue.ts b/packages/telemetry/src/instrumentation/queue.ts index 4cfb094c7..387544c0a 100644 --- a/packages/telemetry/src/instrumentation/queue.ts +++ b/packages/telemetry/src/instrumentation/queue.ts @@ -29,6 +29,7 @@ import { MessagingAttributes, MessagingOperations, MessagingSystems, + NetScriptAttributeDomains, SpanNames, } from '../attributes/mod.ts'; import type { @@ -120,9 +121,11 @@ export class TracedQueue implements MessageQueue { destination: this.options.queueName, operation: MessagingOperations.PUBLISH, messageId, + correlationId: correlationIdFromMessage(message), priority: options?.priority, }), ); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'pending'); // Prepare headers with the queue.enqueue span as the propagated parent. // Existing application headers are preserved, but trace context is replaced @@ -144,6 +147,7 @@ export class TracedQueue implements MessageQueue { ...options, headers, }); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'completed'); addSpanEvent(span, 'message.enqueued', { [MessagingAttributes.MESSAGE_ID]: messageId, @@ -226,10 +230,12 @@ export class TracedQueue implements MessageQueue { destination: this.options.queueName, operation: MessagingOperations.RECEIVE, messageId: ctx.messageId, + correlationId: correlationIdFromMessage(message), deliveryCount: ctx.deliveryCount, }), }, ); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'pending'); // Create a context that includes the queue.dequeue span // This ensures that any spans created by the handler (e.g., job.execute) @@ -247,10 +253,12 @@ export class TracedQueue implements MessageQueue { }); setSpanOk(span); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'completed'); addSpanEvent(span, 'message.processed'); } catch (error) { const message = error instanceof Error ? error.message : String(error); setSpanError(span, message, error instanceof Error ? error : undefined); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'failed'); addSpanEvent(span, 'message.failed', { 'error.message': message }); throw error; } finally { @@ -348,6 +356,14 @@ export class TracedQueue implements MessageQueue { } } +function correlationIdFromMessage(message: unknown): string | undefined { + if (typeof message !== 'object' || message === null || !('correlationId' in message)) { + return undefined; + } + const correlationId = message.correlationId; + return typeof correlationId === 'string' ? correlationId : undefined; +} + // ============================================================================ // FACTORY FUNCTION // ============================================================================ diff --git a/packages/telemetry/src/instrumentation/worker.ts b/packages/telemetry/src/instrumentation/worker.ts index 7294dfd11..2b3ed6b46 100644 --- a/packages/telemetry/src/instrumentation/worker.ts +++ b/packages/telemetry/src/instrumentation/worker.ts @@ -31,6 +31,7 @@ import { ExecutionAttributes, JobAttributes, JobStatuses, + NetScriptAttributeDomains, NetScriptCorrelationAttributes, SpanNames, WorkerAttributes, @@ -303,6 +304,7 @@ export async function traceJobExecution( ...createJobAttributes(job), [ExecutionAttributes.EXECUTION_ID]: execution.executionId, [JobAttributes.JOB_STATUS]: JobStatuses.RUNNING, + [NetScriptAttributeDomains.OUTCOME]: JobStatuses.RUNNING, }; if (execution.attempt) { @@ -339,6 +341,10 @@ export async function traceJobExecution( JobAttributes.JOB_STATUS, result.success ? JobStatuses.COMPLETED : JobStatuses.FAILED, ); + span.setAttribute( + NetScriptAttributeDomains.OUTCOME, + result.success ? JobStatuses.COMPLETED : JobStatuses.FAILED, + ); if (result.exitCode !== undefined) { span.setAttribute(JobAttributes.JOB_EXIT_CODE, result.exitCode); @@ -362,6 +368,7 @@ export async function traceJobExecution( span.setAttribute(JobAttributes.JOB_DURATION_MS, durationMs); span.setAttribute(JobAttributes.JOB_STATUS, JobStatuses.FAILED); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, JobStatuses.FAILED); addSpanEvent(span, 'job.exception', { 'exception.message': errorMessage, diff --git a/plugins/triggers/src/runtime/trigger-runtime-processor.ts b/plugins/triggers/src/runtime/trigger-runtime-processor.ts index 1f7d254f7..d0bcef540 100644 --- a/plugins/triggers/src/runtime/trigger-runtime-processor.ts +++ b/plugins/triggers/src/runtime/trigger-runtime-processor.ts @@ -23,7 +23,12 @@ import { } from '@netscript/plugin-triggers-core/stores'; import type { KvStore } from '@netscript/kv'; import { traceJobDispatch } from '@netscript/telemetry/instrumentation'; -import { SpanNames, TriggerAttributes } from '@netscript/telemetry/attributes'; +import { + NetScriptAttributeDomains, + NetScriptCorrelationAttributes, + SpanNames, + TriggerAttributes, +} from '@netscript/telemetry/attributes'; import { type Context, contextWithSpan, @@ -107,6 +112,8 @@ class TracedTriggerProcessor implements TriggerProcessorPort { [TriggerAttributes.EVENT_ID]: String(event.id), [TriggerAttributes.EVENT_KIND]: String(event.kind), [TriggerAttributes.EVENT_STATUS]: String(event.status), + [NetScriptCorrelationAttributes.CORRELATION_ID]: String(event.id), + [NetScriptAttributeDomains.OUTCOME]: 'pending', }; return await this.#runSpan( @@ -152,10 +159,12 @@ class TracedTriggerProcessor implements TriggerProcessorPort { const childContext = contextWithSpan(span, parentContext); try { const result = await fn(childContext); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'completed'); span.setStatus({ code: SpanStatusCode.OK }); return result; } catch (error: unknown) { const message = error instanceof Error ? error.message : String(error); + span.setAttribute(NetScriptAttributeDomains.OUTCOME, 'failed'); span.setStatus({ code: SpanStatusCode.ERROR, message }); if (error instanceof Error) { span.recordException(error); diff --git a/plugins/workers/worker/job-dispatcher.ts b/plugins/workers/worker/job-dispatcher.ts index d0c8c4606..8cead157f 100644 --- a/plugins/workers/worker/job-dispatcher.ts +++ b/plugins/workers/worker/job-dispatcher.ts @@ -145,6 +145,7 @@ export async function processWorkerJob( payload, abortController.signal, subprocessHeaders, + correlationId, ); addJobStepEvent('state_update', { status: result.success ? 'completed' : 'failed' }); diff --git a/plugins/workers/worker/job-execution.ts b/plugins/workers/worker/job-execution.ts index 3c2e940ac..0ebfdc0cf 100644 --- a/plugins/workers/worker/job-execution.ts +++ b/plugins/workers/worker/job-execution.ts @@ -15,10 +15,11 @@ export async function executeWorkerJob( payload: Record | undefined, signal: AbortSignal, traceHeaders: Record, + correlationId?: string, ): Promise { const executionType = jobDef.executionType ?? 'deno'; if (executionType === 'deno') { - return await executeDenoJob(context, jobDef, payload, traceHeaders); + return await executeDenoJob(context, jobDef, payload, traceHeaders, correlationId); } return await executePolyglotTask(context, jobDef, payload, signal, traceHeaders); } @@ -28,6 +29,7 @@ async function executeDenoJob( jobDef: JobDefinition, payload: Record | undefined, traceHeaders: Record, + correlationId?: string, ): Promise { const entrypoint = resolveDenoEntrypoint(context, jobDef); @@ -42,6 +44,7 @@ async function executeDenoJob( triggeredBy: 'queue', triggeredAt: new Date().toISOString(), payload, + correlationId, traceparent: traceHeaders['traceparent'], tracestate: traceHeaders['tracestate'], };