From 1d3e18f417b6e6a6d535aa4bdc35f44d197c1226 Mon Sep 17 00:00:00 2001 From: maaz519 Date: Fri, 3 Jul 2026 01:04:24 +0530 Subject: [PATCH] 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) --- packages/iios-service/src/ai/ai.service.ts | 5 +- .../src/calendar/calendar.service.ts | 5 +- .../src/interactions/ingest.service.ts | 3 ++ .../src/messaging/message.gateway.ts | 3 ++ .../src/observability/event-trace.spec.ts | 53 +++++++++++++++++++ .../src/observability/trace-context.ts | 6 ++- .../iios-service/src/outbox/outbox.relay.ts | 3 ++ .../iios-service/src/routing/route.service.ts | 5 +- 8 files changed, 78 insertions(+), 5 deletions(-) create mode 100644 packages/iios-service/src/observability/event-trace.spec.ts diff --git a/packages/iios-service/src/ai/ai.service.ts b/packages/iios-service/src/ai/ai.service.ts index 76d5c5e..1aad2cd 100644 --- a/packages/iios-service/src/ai/ai.service.ts +++ b/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): Promise { + 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({ diff --git a/packages/iios-service/src/calendar/calendar.service.ts b/packages/iios-service/src/calendar/calendar.service.ts index 74a073b..6bc67c2 100644 --- a/packages/iios-service/src/calendar/calendar.service.ts +++ b/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): Promise { + 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({ diff --git a/packages/iios-service/src/interactions/ingest.service.ts b/packages/iios-service/src/interactions/ingest.service.ts index 3fd9061..1240282 100644 --- a/packages/iios-service/src/interactions/ingest.service.ts +++ b/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', diff --git a/packages/iios-service/src/messaging/message.gateway.ts b/packages/iios-service/src/messaging/message.gateway.ts index 8e7b168..d0dd7c0 100644 --- a/packages/iios-service/src/messaging/message.gateway.ts +++ b/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 }); }); }); } diff --git a/packages/iios-service/src/observability/event-trace.spec.ts b/packages/iios-service/src/observability/event-trace.spec.ts new file mode 100644 index 0000000..8c5c730 --- /dev/null +++ b/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 { + 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 + }); +}); diff --git a/packages/iios-service/src/observability/trace-context.ts b/packages/iios-service/src/observability/trace-context.ts index 54cb29c..0922327 100644 --- a/packages/iios-service/src/observability/trace-context.ts +++ b/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`; } diff --git a/packages/iios-service/src/outbox/outbox.relay.ts b/packages/iios-service/src/outbox/outbox.relay.ts index bf26a61..6f6f1ca 100644 --- a/packages/iios-service/src/outbox/outbox.relay.ts +++ b/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() }, diff --git a/packages/iios-service/src/routing/route.service.ts b/packages/iios-service/src/routing/route.service.ts index 4ebdaef..c422323 100644 --- a/packages/iios-service/src/routing/route.service.ts +++ b/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 { + 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({