feat(p9): propagate trace id into emitted events + structured relay/gateway logs

Task O.2: ai/calendar/route/ingest emit sites now stamp traceparent + insignia.
correlationId from interaction.traceId ?? currentTraceId() ?? generated (mirroring
message.send; inbox/inbound already carried correlationId). toTraceparent is hex-safe
(UUID used as-is, other ids hashed). OutboxRelay logs a structured relay.publish per
event (eventType/aggregateId/correlationId); MessageGateway logs ws.deliver. Tests:
an event's correlationId equals the active request trace, with a generated fallback.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/iios-service/src/ai/ai.service.ts

@@ -8,6 +8,7 @@ import {
   InferencePort,
   InferenceResult,
 } from '@insignia/iios-contracts';
+import { currentTraceId, newTraceId, toTraceparent } from '../observability/trace-context';
 import { PrismaService } from '../prisma/prisma.service';
 import { PLATFORM_PORTS } from '../platform/platform-ports';
 import { decideOrThrow } from '../platform/fail-closed';
@@ -222,6 +223,7 @@ export class AiJobService {
   }
 
   private async emit(type: string, scopeId: string, aggregateId: string, data: Record<string, unknown>): Promise<void> {
+    const tid = currentTraceId() ?? newTraceId();
     const event: CloudEvent = {
       specversion: '1.0',
       id: `evt_ai_${aggregateId}_${type.split('.').slice(-2, -1)[0]}`,
@@ -230,7 +232,8 @@ export class AiJobService {
       subject: `ai_artifact/${aggregateId}`,
       time: new Date().toISOString(),
       datacontenttype: 'application/json',
-      insignia: { scopeSnapshotId: scopeId, idempotencyKey: `${type}:${aggregateId}`, dataClass: 'internal' },
+      traceparent: toTraceparent(tid),
+      insignia: { scopeSnapshotId: scopeId, correlationId: tid, idempotencyKey: `${type}:${aggregateId}`, dataClass: 'internal' },
       data,
     };
     await this.prisma.iiosOutboxEvent.create({

packages/iios-service/src/calendar/calendar.service.ts

@@ -1,6 +1,7 @@
 import { BadRequestException, Inject, Injectable, NotFoundException } from '@nestjs/common';
 import { Prisma, IiosMeetingType, IiosConsentStatus } from '@prisma/client';
 import { CloudEvent, IIOS_EVENTS, IiosPlatformPorts } from '@insignia/iios-contracts';
+import { currentTraceId, newTraceId, toTraceparent } from '../observability/trace-context';
 import { PrismaService } from '../prisma/prisma.service';
 import { PLATFORM_PORTS } from '../platform/platform-ports';
 import { decideOrThrow } from '../platform/fail-closed';
@@ -373,6 +374,7 @@ export class CalendarService {
   }
 
   protected async emit(type: string, scopeId: string, aggregateId: string, data: Record<string, unknown>): Promise<void> {
+    const tid = currentTraceId() ?? newTraceId();
     const event: CloudEvent = {
       specversion: '1.0',
       id: `evt_cal_${aggregateId}_${type.split('.').slice(-2, -1)[0]}`,
@@ -381,7 +383,8 @@ export class CalendarService {
       subject: `meeting/${aggregateId}`,
       time: new Date().toISOString(),
       datacontenttype: 'application/json',
-      insignia: { scopeSnapshotId: scopeId, idempotencyKey: `${type}:${aggregateId}`, dataClass: 'internal' },
+      traceparent: toTraceparent(tid),
+      insignia: { scopeSnapshotId: scopeId, correlationId: tid, idempotencyKey: `${type}:${aggregateId}`, dataClass: 'internal' },
       data,
     };
     await this.prisma.iiosOutboxEvent.create({

packages/iios-service/src/interactions/ingest.service.ts

@@ -7,6 +7,7 @@ import {
   IngestInteractionRequest,
   IngestInteractionResponse,
 } from '@insignia/iios-contracts';
+import { currentTraceId, newTraceId, toTraceparent } from '../observability/trace-context';
 import { PrismaService } from '../prisma/prisma.service';
 import { PLATFORM_PORTS } from '../platform/platform-ports';
 import { decideOrThrow } from '../platform/fail-closed';
@@ -188,8 +189,10 @@ export class IngestService {
           subject: `interaction/${created.id}`,
           time: new Date().toISOString(),
           datacontenttype: 'application/json',
+          traceparent: toTraceparent(created.traceId ?? currentTraceId() ?? newTraceId()),
           insignia: {
             scopeSnapshotId: scope.id,
+            correlationId: created.traceId ?? currentTraceId() ?? newTraceId(),
             policyDecisionId: decision.decisionRef,
             idempotencyKey,
             dataClass: 'internal',

packages/iios-service/src/messaging/message.gateway.ts

@@ -11,6 +11,7 @@ import {
 } from '@nestjs/websockets';
 import { Server, Socket } from 'socket.io';
 import { IIOS_EVENTS } from '@insignia/iios-contracts';
+import { logJson } from '../observability/logger';
 import { MessageService, type MessagePrincipal } from './message.service';
 import { SessionVerifier } from '../platform/session.verifier';
 import { OutboxBus } from '../outbox/outbox.bus';
@@ -45,8 +46,10 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
       const threadId = ce?.data?.threadId;
       if (!interactionId || !threadId) return;
       if (this.emittedLocally.delete(interactionId)) return; // already emitted inline
+      const correlationId = (payload as { insignia?: { correlationId?: string } })?.insignia?.correlationId;
       void this.messages.getMessageById(interactionId).then((m) => {
         if (m) this.server.to(threadId).emit('message', m);
+        logJson('info', 'ws.deliver', { interactionId, threadId, correlationId });
       });
     });
   }

packages/iios-service/src/observability/event-trace.spec.ts

@@ -0,0 +1,53 @@
+import { randomUUID } from 'node:crypto';
+import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { resetDb } from '../test-utils/reset-db';
+import { makeFakePorts, makeFakeInference } from '@insignia/iios-testkit';
+import { IIOS_EVENTS } from '@insignia/iios-contracts';
+import { MessageService, type MessagePrincipal } from '../messaging/message.service';
+import { ActorResolver } from '../identity/actor.resolver';
+import { AiJobService } from '../ai/ai.service';
+import { AiBudgetGuard } from '../ai/ai-budget.guard';
+import { runWithTrace } from './trace-context';
+import type { PrismaService } from '../prisma/prisma.service';
+
+const url = process.env.DATABASE_URL ?? 'postgresql://iios:iios@localhost:5434/iios?schema=public';
+const prisma = new PrismaClient({ datasources: { db: { url } } });
+const asService = prisma as unknown as PrismaService;
+const actors = new ActorResolver(asService);
+const user: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo' };
+
+const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
+
+async function makeInteraction(text: string): Promise<string> {
+  const m = new MessageService(asService, makeFakePorts(), actors);
+  const { threadId } = await m.openThread(null, user);
+  const msg = await m.send(threadId, user, { content: text }, `k-${randomUUID()}`);
+  return msg.id;
+}
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('event trace propagation (P9 observability)', () => {
+  it('an emitted CloudEvent carries the request trace id in correlationId + traceparent', async () => {
+    const interactionId = await makeInteraction('There is a party at 9 PM tonight!');
+    await runWithTrace('trace-xyz-123', async () => {
+      await ai().runJob({ interactionId, jobType: 'CLASSIFY', principal: user });
+    });
+    const evt = await prisma.iiosOutboxEvent.findFirstOrThrow({ where: { eventType: IIOS_EVENTS.aiArtifactProposed } });
+    const ce = evt.cloudEvent as { traceparent?: string; insignia?: { correlationId?: string; idempotencyKey?: string } };
+    expect(ce.insignia?.correlationId).toBe('trace-xyz-123');
+    expect(ce.traceparent).toMatch(/^00-[0-9a-f]{32}-[0-9a-f]{16}-01$/);
+    expect(ce.insignia?.idempotencyKey).toBeTruthy(); // idempotency key still present
+  });
+
+  it('falls back to a generated trace id when no request context is active', async () => {
+    const interactionId = await makeInteraction('hello');
+    await ai().runJob({ interactionId, jobType: 'SUMMARIZE', principal: user }); // no runWithTrace
+    const evt = await prisma.iiosOutboxEvent.findFirstOrThrow({ where: { eventType: IIOS_EVENTS.aiArtifactProposed } });
+    const ce = evt.cloudEvent as { insignia?: { correlationId?: string } };
+    expect(ce.insignia?.correlationId).toBeTruthy(); // a correlation id is always present
+  });
+});

packages/iios-service/src/observability/trace-context.ts

@@ -1,5 +1,5 @@
 import { AsyncLocalStorage } from 'node:async_hooks';
-import { randomBytes, randomUUID } from 'node:crypto';
+import { createHash, randomBytes, randomUUID } from 'node:crypto';
 
 /**
  * Request-scoped trace context (P9 observability). A module-level AsyncLocalStorage
@@ -30,7 +30,9 @@ export function newTraceId(): string {
 
 /** W3C traceparent for a trace id: 00-<32 hex>-<16 hex span>-01. */
 export function toTraceparent(traceId: string): string {
-  const trace = traceId.replace(/-/g, '').padEnd(32, '0').slice(0, 32);
+  const clean = traceId.replace(/-/g, '').toLowerCase();
+  // A UUID (32 hex after dropping dashes) is used as-is; any other id is hashed to 32 hex.
+  const trace = /^[0-9a-f]{32}$/.test(clean) ? clean : createHash('sha256').update(traceId).digest('hex').slice(0, 32);
   const span = randomBytes(8).toString('hex');
   return `00-${trace}-${span}-01`;
 }

packages/iios-service/src/outbox/outbox.relay.ts

@@ -1,6 +1,7 @@
 import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { PrismaService } from '../prisma/prisma.service';
 import { OutboxBus } from './outbox.bus';
+import { logJson } from '../observability/logger';
 
 /**
  * Transactional-outbox relay (Bottom-Up §12). At-least-once by design; the
@@ -60,6 +61,8 @@ export class OutboxRelay implements OnModuleInit, OnModuleDestroy {
       try {
         const fresh = await this.consumeOnce(eventId);
         if (fresh) this.bus.publish(evt.eventType, evt.cloudEvent);
+        const ce = evt.cloudEvent as { insignia?: { correlationId?: string } } | null;
+        logJson('info', 'relay.publish', { eventType: evt.eventType, aggregateId: evt.aggregateId, correlationId: ce?.insignia?.correlationId });
         await this.prisma.iiosOutboxEvent.update({
           where: { eventId },
           data: { status: 'PUBLISHED', publishedAt: new Date() },

packages/iios-service/src/routing/route.service.ts

@@ -1,6 +1,7 @@
 import { BadRequestException, Injectable, NotFoundException } from '@nestjs/common';
 import { Prisma, IiosRouteBinding, IiosRouteDecision, IiosRouteMode, IiosOutputFormat } from '@prisma/client';
 import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { currentTraceId, newTraceId, toTraceparent } from '../observability/trace-context';
 import { PrismaService } from '../prisma/prisma.service';
 import { RouteSimulator, type OriginRef } from './route-simulator';
 import { OutboundService } from '../adapters/outbound.service';
@@ -118,6 +119,7 @@ export class RouteService {
   }
 
   private async emitApproved(scopeId: string, decisionId: string): Promise<void> {
+    const tid = currentTraceId() ?? newTraceId();
     const event: CloudEvent = {
       specversion: '1.0',
       id: `evt_routeapproved_${decisionId}`,
@@ -126,7 +128,8 @@ export class RouteService {
       subject: `route_decision/${decisionId}`,
       time: new Date().toISOString(),
       datacontenttype: 'application/json',
-      insignia: { scopeSnapshotId: scopeId, idempotencyKey: `routeapproved:${decisionId}`, dataClass: 'internal' },
+      traceparent: toTraceparent(tid),
+      insignia: { scopeSnapshotId: scopeId, correlationId: tid, idempotencyKey: `routeapproved:${decisionId}`, dataClass: 'internal' },
       data: { decisionId },
     };
     await this.prisma.iiosOutboxEvent.create({