feat(iios): email adapter dedup fix + smoke + verification

Adds adapter-owned externalEventId extraction to the ChannelAdapter contract (WebhookAdapter→eventId, EmailAdapter→Message-ID) and uses it in InboundService, so email replays dedup on Message-ID (previously only payload.eventId worked). Adds an email-dedup spec + smoke-email.mjs proving inbound → normalized, a reply joins the same email thread, bad signature rejected, and outbound EMAIL SENT. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
feat(iios): register EmailAdapter for EMAIL + email-shaped egress provider
2026-07-03 20:34:26 +05:30 · 2026-07-03 20:28:35 +05:30 · 2026-07-03 20:25:00 +05:30 · 2026-07-03 19:28:11 +05:30 · 2026-07-03 19:19:13 +05:30 · 2026-07-03 18:48:25 +05:30
43 changed files with 1459 additions and 63 deletions
@@ -0,0 +1,16 @@
 # Build context hygiene — keep the image small & builds reproducible.
 **/node_modules
 **/dist
 **/.turbo
 **/.next
 **/coverage
 .git
 .gitignore
 *.log
 **/*.tsbuildinfo
 .env
 .env.*
 !.env.example
 /tmp
 scratchpad
 docs/openapi.yaml
@@ -16,5 +16,18 @@ services:
      timeout: 5s
      retries: 10
  # Realtime fan-out across replicas (socket.io Redis adapter). Point the service at it
  # with REDIS_URL=redis://localhost:6379; without REDIS_URL the in-memory adapter is used.
  redis:
    image: redis:7-alpine
    container_name: iios-redis
    ports:
      - "6379:6379"
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 5s
      retries: 10
 volumes:
  iios_postgres_data:
@@ -0,0 +1,105 @@
 # Deploying IIOS
 IIOS deploys as **one runtime service** (`@insignia/iios-service`) plus **one datastore**
 (Postgres). Everything else in the repo is a **library** (SDKs) or a **dev demo**, not a
 deployed service.
 | Artifact | Type | How it ships |
 |---|---|---|
 | `@insignia/iios-service` | **Runtime service** | Docker image (this guide) |
 | `iios-contracts`, `iios-adapter-sdk`, `iios-kernel-client`, `iios-testkit` | Library | published to npm; imported by host apps |
 | `iios-message-web`, `iios-inbox-web`, `iios-support-web`, `iios-ai-web`, `iios-meeting-web`, `iios-community-web` | Web SDK (library) | published to npm; **bundled into host-app frontends** (optionally one CDN widget) |
 | `apps/*` (message-demo, ai-studio, route-admin…) | Dev demo | not deployed |
 The service is a **modular monolith**: a single NestJS process serves the HTTP API, the
 WebSocket (socket.io) gateway, the outbox relay, and the projectors together.
 ## Build the image
 Build context is the **monorepo root** (the image needs the workspace + lockfile):
 ```bash
 docker build -f packages/iios-service/Dockerfile -t iios-service:latest .
 ```
 The multi-stage build compiles the service + its workspace deps, generates the Prisma
 client, and prunes to a self-contained prod bundle (`pnpm deploy --legacy`). The runtime
 image runs as the non-root `node` user and starts via `docker-entrypoint.sh`, which runs
 `prisma migrate deploy` then `node dist/main.js`.
 ## Run it
 ```bash
 docker run --rm -p 3200:3200 \
  -e DATABASE_URL='postgresql://USER:PASS@HOST:5432/iios?schema=public' \
  -e APP_SECRETS='{"portal-demo":"<secret>"}' \
  -e IIOS_DEV_TOKENS=0 \
  iios-service:latest
 ```
 Health check: `GET /health` → `200`. Metrics/ops: `GET /metrics` (relay lag, projection
 cursors, retention snapshot counts).
 ## Configuration & secrets contract
 Every knob is an environment variable — see [`packages/iios-service/.env.example`](../packages/iios-service/.env.example)
 for the full, commented list. Highlights:
 - **Secrets** (inject from a vault, never bake into the image): `DATABASE_URL`,
  `APP_SECRETS` (per-app JWT signing keys), `ADAPTER_SECRETS` (webhook HMAC keys).
 - **⚠️ `IIOS_DEV_TOKENS` MUST be `0`/unset in production.** It exposes `/v1/dev/*`
  (unauthenticated token minting, webhook injection, chaos, retention sweep). This is the
  single most important prod-hardening flag.
 - **`IIOS_CELL_ID`** tags which physical cell this instance serves — the hook for splitting
  noisy/regulated tenants into isolated cells later, with no code change.
 - **Worker timers** (`IIOS_RELAY_INTERVAL_MS`, `IIOS_RETENTION_SWEEP_INTERVAL_MS`) — see
  scaling notes below.
 ## Topology
 ```
 host apps ──HTTPS/WSS──► [ LB / ingress ] ──► iios-service ×N ──► Postgres (pgvector)
 (embed SDKs)              (WS + sticky)         (API+relay+                 │
                                                  projectors)         (+ Redis, + OPA/CMP/MDM
                                                                        as external ports — later)
 ```
 - **Postgres** — managed (RDS / Cloud SQL / Neon). One logical DB, tenant-isolated by scope.
 - **Redis** — **not required for a single instance.** Set **`REDIS_URL`** when you run
  **multiple replicas with live chat**: the built-in [socket.io Redis adapter](../packages/iios-service/src/realtime/redis-io.adapter.ts)
  fans room emits across all instances, so a client on replica B sees messages emitted by
  replica A. Without `REDIS_URL` the in-memory adapter is used (single instance). Verified by
  `scripts/smoke-realtime-cluster.mjs` (two instances + one Redis; cross-instance delivery
  fails without Redis, passes with it).
  > Delivery is **at-least-once** across instances — with N>1 the inline send-emit and the
  > relay's outbox-bridge emit can both fire for the same message, so clients should **dedupe
  > by message `id`** (the payload always carries one).
 - **Platform ports** (OPA policy, CMP consent, MDM, CRRE, SAS) — today in-process permissive
  stubs (`LocalDevPorts`). For production, point these at real external services; the service
  already calls them **fail-closed**.
 ## Scaling & release strategy
 **Horizontal scaling is safe** because the event core was hardened for it:
 - The outbox relay claims rows with `FOR UPDATE SKIP LOCKED` → each event is relayed by
  exactly one replica; its co-located projectors process it once (idempotent `claim()` +
  the projection-cursor + idempotency-command ledgers give exactly-once effects).
 - The **DLQ + replay** path means a bad deploy loses nothing — fix and replay.
 **Rolling / blue-green deploys:**
 1. **Migrations:** run `prisma migrate deploy` as a **one-off pre-deploy job**, then set
   `IIOS_SKIP_MIGRATE=1` on the replicas so N pods don't race the migration. (For a single
   instance, the default boot-time migration is fine.)
 2. Use **expand-contract (backward-compatible) schema changes** so old and new pods can run
   against the same schema during the roll — this is the prerequisite for zero-downtime.
 3. Roll replicas with a `/health` readiness gate; drain WebSocket connections on `SIGTERM`.
 4. **WebSocket ingress** needs sticky sessions (and the Redis socket.io adapter once N>1).
 ## Not yet built (prod-readiness gaps)
 - **CI/CD pipeline** to build/push this image and run migrations.
 - **Real platform-port services** (OPA/CMP/MDM) — today permissive stubs.
 - **Secrets manager** integration (currently plain env).
 - A **schema-compatibility gate** in CI to enforce expand-contract migrations.
@@ -0,0 +1,107 @@
 # ADR-0001: `support-service` → IIOS naming migration
 - **Status:** Accepted
 - **Date:** 2026-07-03
 - **Deciders:** IIOS core
 - **Context tags:** naming, packaging, migration, KG-16
 ## Context
 IIOS (Insignia Interaction OS) is the greenfield realization of a vision first
 prototyped in the standalone **`support-service`** app. That prototype already stated the
 core idea — *"Everything is a message: a generic Message core (conversation / message /
 channel / user) is the foundation; Support (ticket / agent / callback) is a thin
 specialization on top"* — but it did so with **support-centric naming** and as a
 single-purpose support desk.
 IIOS generalizes that same idea to an **interaction OS** of much broader scope
 (messaging, inbox, routing, AI enrichment, calendar, plus support as *one*
 specialization among many). During P0–P9 it landed a clean, generic vocabulary:
 an `Iios`-prefixed persistence layer and `@insignia/iios-*` packages.
 Two problems make an explicit decision necessary:
 1. **Naming drift.** The prototype and IIOS use different words for the same concepts
   (`Conversation` vs `IiosThread`, `Message` vs `IiosInteraction` + `IiosMessagePart`,
   `User` vs `IiosSourceHandle` + `IiosActorRef`). Without a canonical mapping, contributors
   coming from the prototype reintroduce support-centric names, and host apps don't know
   which SDK is authoritative.
 2. **KG-16 — "support wedge evidence overgeneralized."** The critics' analysis flags the
   risk of treating the `support-service` architecture as proof of *all* IIOS capabilities.
   Support is a **specialization**, not the core; the naming must make that structural
   (support names are namespaced *under* the generic core, never the core itself).
 An unresolved sub-question also blocks packaging: the workspace scope was historically
 ambiguous between `@insignia/*` and `@lynkeduppro/*`.
 ## Decision
 **1. Canonical vocabulary is interaction-centric and `Iios`-namespaced.**
 The generic core is the noun set already shipped in `@insignia/iios-service`:
 `IiosScope`, `IiosSourceHandle`/`IiosActorRef`, `IiosChannel`, `IiosThread`,
 `IiosInteraction`/`IiosMessagePart`. **Support is a specialization layered on top** and is
 always namespaced as such (`IiosSupport*`, `IiosTicket*`, `IiosCallbackRequest`) — never
 promoted to a core concept. "Conversation" and support-centric "Message-as-ticket" framing
 are retired.
 **2. Package scope is `@insignia/iios-*`.** This resolves the `@insignia` vs `@lynkeduppro`
 ambiguity in favour of `@insignia`. All IIOS packages already follow `@insignia/iios-<name>`;
 that is now the standard.
 **3. Migration is greenfield-and-deprecate (strangler), not rename-in-place.**
 `support-service` is **frozen legacy** — we do not rename its models or ship a new version of
 it. New work happens in IIOS; host apps cut over to the `@insignia/iios-*` SDKs feature-by-
 feature. `support-service` is retired once every host app it serves has migrated.
 ### Canonical naming map (legacy → IIOS)
 | `support-service` (legacy) | IIOS (canonical) | Note |
 |---|---|---|
 | `Channel` | `IiosChannel` | generic ingress/egress channel |
 | `User` | `IiosSourceHandle` + `IiosActorRef` | identity split: external handle vs resolved actor |
 | `Conversation` | `IiosThread` | "conversation" retired; thread is generic |
 | `Message` | `IiosInteraction` + `IiosMessagePart` | envelope vs content parts |
 | `Attachment` | `IiosMessagePart` (`kind=MEDIA`, `contentRef`) | attachments are just parts |
 | `Team` / `TeamMember` | `IiosSupportQueue` / `IiosSupportTeamMember` | **support specialization**, namespaced |
 | `ConversationAssignment` | support assignment (`IiosSupport*`) | specialization, not core |
 | `Ticket` / `TicketConversation` | `IiosTicket` / `IiosTicketThreadLink` | ticket links to generic threads |
 | `CallbackRequest` | `IiosCallbackRequest` | |
 | `Meeting` | `IiosMeeting` (+ P8 calendar models) | promoted to first-class in IIOS |
 | `Notification` | `notification_outbox` (planned) | not yet built in IIOS |
 | `ProcessedCommand` | `IiosProcessedEvent` + `IiosIdempotencyCommand` | consumer ledger vs command ledger |
 | `KnowledgeArticle` | (deferred — AI/RAG) | out of current scope |
 ### Rules going forward
 - **Every persisted model is `Iios`-prefixed.** No un-prefixed domain models.
 - **Support (and any future vertical) is namespaced under the generic core**, never the other
  way round. If a name reads as "support-only," it must carry the `IiosSupport*`/`IiosTicket*`
  prefix and depend on the core — the core never depends on it (enforced by the import-boundary
  check).
 - **New packages are `@insignia/iios-<name>`.**
 - **Legacy references** in prose cite `support-service` explicitly as *legacy*, with a pointer
  to this ADR.
 ## Consequences
 **Positive**
 - One authoritative vocabulary; contributors from the prototype have a lookup table.
 - KG-16 is structurally addressed: support cannot masquerade as the core because its names are
  subordinate and the boundary check enforces the dependency direction.
 - Packaging is unblocked (`@insignia/iios-*` is canonical).
 - No risky in-place rename of a running legacy service; cutover is incremental per host app.
 **Negative / costs**
 - Two systems coexist during the strangler window; the legacy `support-service` must be kept
  running (but frozen) until host apps migrate.
 - The legacy↔IIOS mapping must be consulted when porting features; this ADR is that map.
 - `Notification` / `KnowledgeArticle` have no IIOS home yet — tracked as follow-ups, not blockers.
 ## Alternatives considered
 - **Rename `support-service` in place** to the IIOS vocabulary. Rejected: high-risk migration
  of a live service + DB for a system we intend to retire anyway; no benefit over greenfield.
 - **Keep both vocabularies and bridge with adapters.** Rejected: perpetuates naming drift and
  the KG-16 confusion indefinitely.
 - **`@lynkeduppro/*` scope.** Rejected: all IIOS packages already ship under `@insignia`;
  switching scopes now is churn with no upside. (A future rebrand can supersede this ADR.)
@@ -0,0 +1,19 @@
 # Architecture Decision Records (ADRs)
 This directory records the significant, hard-to-reverse decisions behind IIOS —
 the *why*, not just the *what*. Code shows what we did; ADRs explain the choice so a
 future contributor doesn't relitigate it or unknowingly violate it.
 ## Convention
 - One file per decision: `NNNN-kebab-title.md` (zero-padded, monotonically increasing).
 - Each ADR has: **Status**, **Context**, **Decision**, **Consequences**, and (where useful)
  **Alternatives considered**.
 - Status is one of `Proposed` · `Accepted` · `Superseded by ADR-XXXX` · `Deprecated`.
 - ADRs are immutable once `Accepted`. To change a decision, write a new ADR that supersedes it.
 ## Index
 | ADR | Title | Status |
 |-----|-------|--------|
 | [0001](0001-support-service-to-iios-naming-migration.md) | `support-service` → IIOS naming migration | Accepted |
@@ -0,0 +1,45 @@
 import { describe, it, expect } from 'vitest';
 import { EmailAdapter, emailInboundFixture, emailReplyFixture, signedFixture } from './index';
 const ctx = { scope: { orgId: 'o', appId: 'a' } };
 describe('EmailAdapter', () => {
  const adapter = new EmailAdapter();
  it('normalizes an inbound email into an ingest request', () => {
    const req = adapter.normalize(emailInboundFixture, ctx);
    expect(req.channel.type).toBe('EMAIL');
    expect(req.source.handleKind).toBe('EXTERNAL_VISITOR');
    expect(req.source.externalId).toBe(emailInboundFixture.from);
    expect(req.parts[0]?.bodyText).toBe(emailInboundFixture.text);
    expect(req.providerEventId).toBe('<m1@example.com>');
    expect(req.thread?.externalThreadId).toBe('email:<m1@example.com>');
    expect(req.thread?.subject).toBe('Need help with my order'); // Re:/Fwd: stripped
    expect((req.metadata as { messageId: string }).messageId).toBe('<m1@example.com>');
  });
  it('threads a reply onto the SAME email thread as the original', () => {
    const first = adapter.normalize(emailInboundFixture, ctx);
    const reply = adapter.normalize(emailReplyFixture, ctx);
    expect(reply.thread?.externalThreadId).toBe(first.thread?.externalThreadId); // email:<m1@example.com>
    expect(reply.providerEventId).toBe('<m2@example.com>'); // but its own dedup id
  });
  it('falls back to the reply target when there is no references chain', () => {
    const req = adapter.normalize({ ...emailReplyFixture, references: undefined }, ctx);
    expect(req.thread?.externalThreadId).toBe('email:<m1@example.com>'); // uses inReplyTo
  });
  it('derives text from html when no plain text is present', () => {
    const req = adapter.normalize({ messageId: '<h@x>', from: 'x@y', html: '<p>Hello <b>there</b></p>' }, ctx);
    expect(req.parts[0]?.bodyText).toBe('Hello there');
  });
  it('verifies an HMAC-signed body and rejects a bad/absent signature', () => {
    const secret = 'sekret';
    const { body, headers } = signedFixture(emailInboundFixture, secret);
    expect(adapter.verifySignature(body, headers, secret)).toBe(true);
    expect(adapter.verifySignature(body, { 'x-iios-signature': 'sha256=bad' }, secret)).toBe(false);
    expect(adapter.verifySignature(body, {}, secret)).toBe(false);
  });
 });
@@ -0,0 +1,71 @@
 import type { IngestInteractionRequest } from '@insignia/iios-contracts';
 import type { AdapterCapability, ChannelAdapter, NormalizeContext, OutboundCommandInput, SendResult } from './types';
 import { hmacVerify } from './hmac';
 import { SandboxSink } from './sandbox';
 /** Canonical inbound-email shape. Vendor payloads (SendGrid Inbound Parse, Mailgun
 *  Routes, Postmark) map into this — provider glue stays a thin transform. */
 export interface EmailInboundPayload {
  messageId: string; // RFC 5322 Message-ID → providerEventId (dedup on redelivery)
  from: string;
  fromName?: string;
  to?: string[];
  cc?: string[];
  subject?: string;
  text?: string;
  html?: string;
  inReplyTo?: string; // Message-ID this is a reply to
  references?: string[]; // full reference chain, root-first (threading)
  occurredAt?: string;
  headers?: Record<string, string>;
 }
 const stripReply = (s?: string): string => (s ?? '').replace(/^(re|fwd|fw)\s*:\s*/i, '').trim();
 const stripHtml = (h?: string): string => (h ?? '').replace(/<[^>]+>/g, ' ').replace(/\s+/g, ' ').trim();
 /**
 * Email channel adapter. Turns a provider's inbound email into a normalized
 * interaction with email-native threading: the whole reply chain collapses to one
 * IIOS thread (rooted at `references[0]`/`inReplyTo`), and `messageId` dedups
 * redelivery. HMAC signature for now (per-vendor schemes are a follow-up).
 */
 export class EmailAdapter implements ChannelAdapter {
  readonly channelType = 'EMAIL';
  readonly capability: AdapterCapability = {
    canReceive: true,
    canSend: true,
    supportsThreads: true,
    requiresFollowupFetch: false,
  };
  verifySignature(rawBody: string, headers: Record<string, string | undefined>, secret: string): boolean {
    const sig = headers['x-iios-signature'] ?? headers['X-Iios-Signature'];
    return hmacVerify(rawBody, secret, typeof sig === 'string' ? sig : undefined);
  }
  externalEventId(payload: unknown): string | undefined {
    const id = (payload as EmailInboundPayload)?.messageId;
    return typeof id === 'string' ? id : undefined;
  }
  normalize(payload: unknown, ctx: NormalizeContext): IngestInteractionRequest {
    const p = payload as EmailInboundPayload;
    // A reply's thread root is the first Message-ID in its chain → same IIOS thread as the original.
    const threadRoot = p.references?.[0] ?? p.inReplyTo ?? p.messageId;
    return {
      scope: ctx.scope,
      channel: { type: 'EMAIL', externalChannelId: 'email' },
      source: { handleKind: 'EXTERNAL_VISITOR', externalId: p.from, displayName: p.fromName ?? p.from },
      kind: 'MESSAGE',
      thread: { externalThreadId: `email:${threadRoot}`, subject: stripReply(p.subject) },
      parts: [{ kind: 'TEXT', bodyText: p.text ?? stripHtml(p.html), mimeType: 'text/plain' }],
      occurredAt: p.occurredAt ?? new Date().toISOString(),
      providerEventId: p.messageId,
      metadata: { to: p.to, cc: p.cc, messageId: p.messageId, inReplyTo: p.inReplyTo, references: p.references, html: p.html },
    };
  }
  async send(cmd: OutboundCommandInput): Promise<SendResult> {
    return SandboxSink.simulate(cmd);
  }
 }
@@ -1,7 +1,8 @@
 import { hmacSign } from './hmac';
 import type { WebhookPayload } from './webhook-adapter';
 import type { EmailInboundPayload } from './email-adapter';
-/** Email-shaped provider payload (as the adapter's normalize input). */
+/** Email-shaped WebhookPayload (legacy generic-webhook fixture). */
 export const emailFixture: WebhookPayload = {
  eventId: 'email-1',
  from: 'someone@example.com',
@@ -11,6 +12,30 @@ export const emailFixture: WebhookPayload = {
  text: 'Hello, I have a question.',
 };
 /** A first inbound email (starts a thread) for the EmailAdapter. */
 export const emailInboundFixture: EmailInboundPayload = {
  messageId: '<m1@example.com>',
  from: 'buyer@example.com',
  fromName: 'Buyer',
  to: ['support@tower.example'],
  subject: 'Need help with my order',
  text: 'Hi, my order has not arrived yet.',
  occurredAt: '2026-07-03T10:00:00.000Z',
 };
 /** A reply to `emailInboundFixture` — must land on the SAME email thread. */
 export const emailReplyFixture: EmailInboundPayload = {
  messageId: '<m2@example.com>',
  from: 'buyer@example.com',
  fromName: 'Buyer',
  to: ['support@tower.example'],
  subject: 'Re: Need help with my order',
  text: 'Any update on this?',
  inReplyTo: '<m1@example.com>',
  references: ['<m1@example.com>'],
  occurredAt: '2026-07-03T11:00:00.000Z',
 };
 /** WhatsApp-shaped provider payload. */
 export const whatsappFixture: WebhookPayload = {
  eventId: 'wa-1',
@@ -2,4 +2,5 @@ export * from './types';
 export { hmacSign, hmacVerify } from './hmac';
 export { SandboxSink } from './sandbox';
 export { WebhookAdapter, type WebhookPayload } from './webhook-adapter';
-export { emailFixture, whatsappFixture, signedFixture } from './fixtures';
+export { EmailAdapter, type EmailInboundPayload } from './email-adapter';
 export { emailFixture, whatsappFixture, emailInboundFixture, emailReplyFixture, signedFixture } from './fixtures';
@@ -34,6 +34,8 @@ export interface ChannelAdapter {
  channelType: string;
  capability: AdapterCapability;
  verifySignature(rawBody: string, headers: Record<string, string | undefined>, secret: string): boolean;
  /** Provider-native id used to dedup replays/retries at the inbound edge (e.g. eventId, Message-ID). */
  externalEventId?(payload: unknown): string | undefined;
  normalize(payload: unknown, ctx: NormalizeContext): IngestInteractionRequest;
  fetch?(ref: string): Promise<unknown>;
  send(cmd: OutboundCommandInput): Promise<SendResult>;
@@ -37,6 +37,11 @@ export class WebhookAdapter implements ChannelAdapter {
    return hmacVerify(rawBody, secret, typeof sig === 'string' ? sig : undefined);
  }
  externalEventId(payload: unknown): string | undefined {
    const id = (payload as WebhookPayload)?.eventId;
    return typeof id === 'string' ? id : undefined;
  }
  normalize(payload: unknown, ctx: NormalizeContext): IngestInteractionRequest {
    const p = payload as WebhookPayload;
    return {
@@ -0,0 +1,51 @@
 # ─────────────────────────────────────────────────────────────────────────────
 # iios-service configuration contract.
 # Copy to `.env` for local dev; in prod, inject via your secrets manager / orchestrator.
 # 🔒 = secret (never commit a real value). ⚠️ = must be set correctly for prod safety.
 # ─────────────────────────────────────────────────────────────────────────────
 # ── Core ─────────────────────────────────────────────────────────────────────
 DATABASE_URL=postgresql://iios:iios@localhost:5434/iios?schema=public   # 🔒 Postgres connection
 PORT=3200
 # NODE_ENV=production   # set by the Docker image
 # Realtime fan-out across replicas (socket.io Redis adapter). REQUIRED when running
 # more than one instance with live chat; omit for a single instance (in-memory adapter).
 # REDIS_URL=redis://localhost:6379                                       # 🔒
 # ── Secrets ──────────────────────────────────────────────────────────────────
 # JSON map of appId → HS256 signing secret used to verify session JWTs (SessionVerifier).
 APP_SECRETS={"portal-demo":"dev-secret"}                                 # 🔒
 # JSON map of channelType → inbound webhook HMAC secret (adapter signature check).
 ADAPTER_SECRETS={"WEBHOOK":"dev-adapter-secret"}                         # 🔒
 # Default scope an unauthenticated adapter webhook ingests into.
 ADAPTER_APP=portal-demo
 ADAPTER_ORG=org_demo
 # ── ⚠️ Production-safety flags ────────────────────────────────────────────────
 # Enables /v1/dev/* (token mint, webhook inject, chaos, retention sweep). MUST be unset
 # or 0 in production — leaving it on exposes unauthenticated token minting.
 IIOS_DEV_TOKENS=0
 # Skip `prisma migrate deploy` at boot (set to 1 when a separate migration job runs).
 IIOS_SKIP_MIGRATE=0
 # ── Tenant isolation ─────────────────────────────────────────────────────────
 IIOS_CELL_ID=cell-default            # physical cell this instance serves (blast-radius partition)
 # ── Background workers ───────────────────────────────────────────────────────
 IIOS_RELAY_INTERVAL_MS=500           # outbox relay tick; 0 disables the timer
 IIOS_OUTBOX_MAX_ATTEMPTS=5           # relay dead-letters an event after N failed publishes
 IIOS_RETENTION_SWEEP_INTERVAL_MS=0   # retention sweep tick; 0 disables (run via job instead)
 IIOS_RETENTION_POLICY_VERSION=v1
 IIOS_RETENTION_ARCHIVE_DAYS=90       # default archive window (per-class override: _INTERNAL/_RESTRICTED/_CONFIDENTIAL/_REGULATED)
 IIOS_RETENTION_DELETE_DAYS=365       # default delete (redact) window; same per-class overrides
 # ── Rate limits / quotas / budgets ───────────────────────────────────────────
 IIOS_OUTBOUND_LIMIT=5                # per-(channel,target) sends per window
 IIOS_OUTBOUND_WINDOW_MS=60000
 IIOS_TENANT_OUTBOUND_LIMIT=10000     # per-tenant egress cap per window (noisy-neighbor guard)
 IIOS_TENANT_OUTBOUND_WINDOW_MS=60000
 IIOS_AI_BUDGET_UNITS=100000          # per-scope AI cost-unit budget (KG-12)
 # ── Capability providers (governed egress targets) ───────────────────────────
 # Per-channel provider endpoint the CapabilityBroker calls, e.g.:
 # IIOS_PROVIDER_URL_EMAIL=https://provider.internal/email
@@ -0,0 +1,54 @@
 # syntax=docker/dockerfile:1
 # Production image for @insignia/iios-service (the single IIOS backend).
 # Build context MUST be the monorepo ROOT (it needs the workspace + lockfile):
 #   docker build -f packages/iios-service/Dockerfile -t iios-service:latest .
 #
 # Multi-stage: (1) build the service + its workspace deps and prune to a
 # self-contained prod bundle via `pnpm deploy`; (2) a slim runtime that only
 # carries that bundle. Migrations run at container start (see docker-entrypoint.sh).
 ARG NODE_VERSION=22-bookworm-slim
 ARG PNPM_VERSION=10.6.2
 # ── build ───────────────────────────────────────────────────────────────────
 FROM node:${NODE_VERSION} AS build
 ARG PNPM_VERSION
 ENV PNPM_HOME=/pnpm PATH=/pnpm:$PATH
 # openssl + ca-certificates are required by Prisma's engine on debian-slim.
 RUN apt-get update && apt-get install -y --no-install-recommends openssl ca-certificates \
  && rm -rf /var/lib/apt/lists/*
 RUN corepack enable && corepack prepare pnpm@${PNPM_VERSION} --activate
 WORKDIR /repo
 COPY . .
 # Full install (needs devDeps to compile), then build ONLY the service + its
 # workspace dependencies, generate the Prisma client, and emit a pruned prod bundle.
 RUN --mount=type=cache,id=pnpm-store,target=/pnpm/store \
    pnpm install --frozen-lockfile
 # Generate the Prisma client BEFORE compiling — nest build (tsc) needs its types,
 # otherwise Prisma results are `any` and noImplicitAny fails the build.
 RUN pnpm --filter @insignia/iios-service exec prisma generate \
 && pnpm --filter "@insignia/iios-service..." build \
 && pnpm --filter @insignia/iios-service --prod --legacy deploy /app
 # ── runtime ──────────────────────────────────────────────────────────────────
 FROM node:${NODE_VERSION} AS runtime
 ENV NODE_ENV=production
 # Prisma engine needs openssl at runtime too.
 RUN apt-get update && apt-get install -y --no-install-recommends openssl ca-certificates \
  && rm -rf /var/lib/apt/lists/*
 WORKDIR /app
 COPY --from=build /app ./
 # Regenerate the Prisma client against the final filesystem (deterministic).
 RUN node_modules/.bin/prisma generate
 # Run as the built-in non-root `node` user.
 RUN chmod +x docker-entrypoint.sh && chown -R node:node /app
 USER node
 EXPOSE 3200
 ENV PORT=3200
 # migrate deploy → start; see docker-entrypoint.sh.
 ENTRYPOINT ["./docker-entrypoint.sh"]
@@ -0,0 +1,19 @@
 #!/bin/sh
 # Container entrypoint for iios-service.
 # 1) Apply pending DB migrations (idempotent; safe to run on every boot).
 # 2) Exec the server as PID 1 so signals (SIGTERM) reach Node for graceful shutdown.
 #
 # NOTE: at higher replica counts, prefer running `prisma migrate deploy` ONCE as a
 # separate pre-deploy job/init-container and set IIOS_SKIP_MIGRATE=1 here, so N
 # replicas don't race the migration on rollout.
 set -e
 if [ "${IIOS_SKIP_MIGRATE:-0}" != "1" ]; then
  echo "[entrypoint] applying migrations (prisma migrate deploy)…"
  node_modules/.bin/prisma migrate deploy
 else
  echo "[entrypoint] IIOS_SKIP_MIGRATE=1 — skipping migrations"
 fi
 echo "[entrypoint] starting iios-service on :${PORT:-3200}"
 exec node dist/main.js
@@ -12,21 +12,24 @@
    "prisma:studio": "prisma studio"
  },
  "dependencies": {
    "@insignia/iios-contracts": "workspace:*",
    "@insignia/iios-adapter-sdk": "workspace:*",
    "@insignia/iios-contracts": "workspace:*",
    "@nestjs/common": "^11.1.27",
    "@nestjs/core": "^11.1.27",
    "@nestjs/platform-express": "^11.1.27",
    "@nestjs/platform-socket.io": "^11.1.27",
    "@nestjs/websockets": "^11.1.27",
    "@prisma/client": "^6.2.1",
-    "socket.io": "^4.8.3",
+    "@socket.io/redis-adapter": "^8.3.0",
    "class-transformer": "^0.5.1",
    "class-validator": "^0.15.1",
    "dotenv": "^16.4.7",
    "ioredis": "^5.11.1",
    "jsonwebtoken": "^9.0.3",
    "prisma": "^6.2.1",
    "reflect-metadata": "^0.2.2",
-    "rxjs": "^7.8.2"
+    "rxjs": "^7.8.2",
    "socket.io": "^4.8.3"
  },
  "devDependencies": {
    "@insignia/iios-testkit": "workspace:*",
@@ -35,7 +38,6 @@
    "@types/express": "^5.0.6",
    "@types/jsonwebtoken": "^9.0.10",
    "@types/node": "^26.0.1",
    "prisma": "^6.2.1",
    "socket.io-client": "^4.8.3",
    "typescript": "^5.7.3"
  }
@@ -0,0 +1,29 @@
 -- AlterTable
 ALTER TABLE "IiosInteraction" ADD COLUMN     "dataClass" TEXT NOT NULL DEFAULT 'internal';
 -- CreateTable
 CREATE TABLE "IiosRetentionPolicySnapshot" (
    "id" TEXT NOT NULL,
    "policyKey" TEXT NOT NULL,
    "scopeSnapshotId" TEXT NOT NULL,
    "targetType" TEXT NOT NULL,
    "targetId" TEXT NOT NULL,
    "dataClass" TEXT NOT NULL,
    "retainUntil" TIMESTAMP(3),
    "archiveAfter" TIMESTAMP(3) NOT NULL,
    "deleteAfter" TIMESTAMP(3) NOT NULL,
    "sourceVersion" TEXT NOT NULL DEFAULT 'v1',
    "status" TEXT NOT NULL DEFAULT 'ACTIVE',
    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
    CONSTRAINT "IiosRetentionPolicySnapshot_pkey" PRIMARY KEY ("id")
 );
 -- CreateIndex
 CREATE INDEX "IiosRetentionPolicySnapshot_status_deleteAfter_idx" ON "IiosRetentionPolicySnapshot"("status", "deleteAfter");
 -- CreateIndex
 CREATE INDEX "IiosRetentionPolicySnapshot_scopeSnapshotId_status_idx" ON "IiosRetentionPolicySnapshot"("scopeSnapshotId", "status");
 -- CreateIndex
 CREATE UNIQUE INDEX "IiosRetentionPolicySnapshot_targetType_targetId_key" ON "IiosRetentionPolicySnapshot"("targetType", "targetId");
@@ -435,6 +435,7 @@ model IiosInteraction {
  occurredAt          DateTime            @default(now())
  receivedAt          DateTime            @default(now())
  status              IiosDeliveryState   @default(RECEIVED)
  dataClass           String              @default("internal") // envelope data class; drives retention window (P9)
  policyDecisionRef   String?
  consentReceiptRef   String?
  crreBundleRef       String?
@@ -493,6 +494,27 @@ model IiosProcessedEvent {
  @@id([consumerName, eventId])
 }
 /// Per-resource retention snapshot (P9). Absolute lifecycle timestamps frozen from the
 /// resource's age + policy window; read by the retention sweep. delete = redact-in-place.
 model IiosRetentionPolicySnapshot {
  id              String    @id @default(cuid())
  policyKey       String
  scopeSnapshotId String
  targetType      String // 'interaction'
  targetId        String
  dataClass       String
  retainUntil     DateTime?
  archiveAfter    DateTime
  deleteAfter     DateTime
  sourceVersion   String    @default("v1")
  status          String    @default("ACTIVE") // ACTIVE | ARCHIVED | REDACTED
  createdAt       DateTime  @default(now())
  @@unique([targetType, targetId])
  @@index([status, deleteAfter])
  @@index([scopeSnapshotId, status])
 }
 /// Legal/compliance hold (P9 DSR). An ACTIVE, unexpired hold over a target BLOCKS
 /// erasure/retention deletion until an authorised release. Never auto-released.
 model IiosComplianceHold {
@@ -0,0 +1,65 @@
 // P9 slice-8 DSR smoke: right-to-erasure redacts a subject's PII in place (a ticket
 // subject), an active compliance hold blocks erasure until released, erasure is
 // idempotent, and holds are tenant-fenced (KG-03 / KG-02). Requires the service
 // running with IIOS_DEV_TOKENS=1.
 import 'dotenv/config';
 const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200';
 const APP_ID = 'portal-demo';
 const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); };
 async function devToken(userId, orgId) {
  const r = await fetch(`${SERVICE}/v1/dev/token`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ appId: APP_ID, userId, name: userId, orgId }),
  });
  if (!r.ok) throw new Error(`devToken ${r.status} (service with IIOS_DEV_TOKENS=1?)`);
  return (await r.json()).token;
 }
 function call(token, path, method = 'GET', body) {
  return fetch(`${SERVICE}${path}`, {
    method,
    headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` },
    body: body ? JSON.stringify(body) : undefined,
  });
 }
 async function json(token, path, method = 'GET', body) {
  const r = await call(token, path, method, body);
  if (!r.ok) throw new Error(`${method} ${path} ${r.status}: ${await r.text()}`);
  return r.json();
 }
 const ticketSubject = async (token, id) => (await json(token, '/v1/support/tickets')).find((t) => t.id === id)?.subject;
 const A = await devToken('dsr-a', 'org_dsr_A');
 const B = await devToken('dsr-b', 'org_dsr_B');
 const PII = 'my SSN is 123-45';
 // A creates a ticket carrying PII.
 const ticket = await json(A, '/v1/support/tickets', 'POST', { subject: PII });
 assert((await ticketSubject(A, ticket.id)) === PII, 'ticket carries the subject PII');
 // A places a compliance hold on their own subject.
 const hold = await json(A, '/v1/dsr/holds', 'POST', { targetType: 'data_subject', targetId: 'self', reason: 'litigation' });
 assert((await json(A, '/v1/dsr/holds')).some((h) => h.id === hold.id), 'compliance hold is listed');
 // Erasure is blocked while the hold is active.
 const blocked = await call(A, '/v1/dsr/erase', 'POST');
 assert(blocked.status === 409, `erase blocked by the hold → 409 (got ${blocked.status})`);
 assert((await ticketSubject(A, ticket.id)) === PII, 'PII still present while held');
 // Release the hold → erasure now redacts the PII in place.
 await json(A, `/v1/dsr/holds/${hold.id}/release`, 'POST');
 const erased = await json(A, '/v1/dsr/erase', 'POST');
 assert(erased.status === 'ERASED', `erase succeeds after release (status ${erased.status})`);
 assert((await ticketSubject(A, ticket.id)) === '[redacted]', 'ticket subject is redacted (erased in place, not deleted)');
 // Idempotent: a second erase is a no-op.
 const again = await json(A, '/v1/dsr/erase', 'POST');
 assert(again.status === 'ALREADY_ERASED', `re-erase is idempotent (status ${again.status})`);
 // Tenant fence: tenant B never sees A's holds.
 assert(!(await json(B, '/v1/dsr/holds')).some((h) => h.id === hold.id), "tenant B's hold list excludes A's hold");
 console.log('\nP9 DSR smoke: PASS');
 process.exit(0);
@@ -0,0 +1,88 @@
 // Email adapter smoke: a signed inbound email → normalized interaction; a reply
 // (In-Reply-To) lands on the SAME email thread; outbound EMAIL → SENT (sandbox).
 // Requires the service running with IIOS_DEV_TOKENS=1 (relay timer on). No real network.
 import 'dotenv/config';
 import crypto from 'node:crypto';
 import { randomUUID } from 'node:crypto';
 import jwt from 'jsonwebtoken';
 import { PrismaClient } from '@prisma/client';
 const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200';
 const APP_ID = 'portal-demo';
 const APP_SECRET = JSON.parse(process.env.APP_SECRETS ?? '{"portal-demo":"dev-secret"}')[APP_ID];
 const EMAIL_SECRET = (() => {
  try { return JSON.parse(process.env.ADAPTER_SECRETS ?? '{}').EMAIL ?? 'dev-adapter-secret'; } catch { return 'dev-adapter-secret'; }
 })();
 const prisma = new PrismaClient();
 const token = jwt.sign({ sub: 'inspector', appId: APP_ID, orgId: `org_${APP_ID}` }, APP_SECRET, { algorithm: 'HS256', expiresIn: '1h' });
 const sign = (body) => 'sha256=' + crypto.createHmac('sha256', EMAIL_SECRET).update(body).digest('hex');
 const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
 const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); };
 async function postEmail(payload) {
  const body = JSON.stringify(payload);
  return fetch(`${SERVICE}/v1/adapters/EMAIL/webhook`, {
    method: 'POST',
    headers: { 'content-type': 'application/json', 'x-iios-signature': sign(body) },
    body,
  });
 }
 async function inboundList() {
  const r = await fetch(`${SERVICE}/v1/adapters/inbound`, { headers: { authorization: `Bearer ${token}` } });
  if (!r.ok) throw new Error(`GET /v1/adapters/inbound ${r.status}`);
  return r.json();
 }
 async function waitNormalized(messageId) {
  for (let i = 0; i < 40; i++) {
    const e = (await inboundList()).find((x) => x.externalEventId === messageId);
    if (e && e.status === 'NORMALIZED' && e.interactionId) return e.interactionId;
    await sleep(300);
  }
  return null;
 }
 // Unique Message-IDs per run so re-runs don't dedup against earlier ones.
 const run = randomUUID().slice(0, 8);
 const mid1 = `<${run}-1@smoke>`;
 const mid2 = `<${run}-2@smoke>`;
 // 1) A first inbound email → normalized interaction.
 const r1 = await postEmail({ messageId: mid1, from: 'buyer@smoke', fromName: 'Buyer', subject: 'Need help with my order', text: 'my order is late' });
 assert(r1.status === 202, 'signed email accepted (202)');
 const i1 = await waitNormalized(mid1);
 assert(i1, 'first email normalized into an interaction');
 // 2) A reply (In-Reply-To) → same email thread.
 const r2 = await postEmail({ messageId: mid2, from: 'buyer@smoke', subject: 'Re: Need help with my order', text: 'any update?', inReplyTo: mid1, references: [mid1] });
 assert(r2.status === 202, 'reply email accepted (202)');
 const i2 = await waitNormalized(mid2);
 assert(i2, 'reply email normalized into an interaction');
 const [it1, it2] = await Promise.all([
  prisma.iiosInteraction.findUniqueOrThrow({ where: { id: i1 }, include: { thread: true } }),
  prisma.iiosInteraction.findUniqueOrThrow({ where: { id: i2 }, include: { thread: true } }),
 ]);
 assert(it1.threadId === it2.threadId, 'reply joined the SAME email thread as the original');
 assert(it1.thread?.externalThreadRef === `email:${mid1}`, `thread rooted at the original Message-ID (${it1.thread?.externalThreadRef})`);
 // 3) Bad signature → rejected.
 const bad = await fetch(`${SERVICE}/v1/adapters/EMAIL/webhook`, {
  method: 'POST',
  headers: { 'content-type': 'application/json', 'x-iios-signature': 'sha256=bad' },
  body: JSON.stringify({ messageId: `<${run}-bad@smoke>`, from: 'x', text: 'x' }),
 });
 assert(bad.status === 401, 'bad signature rejected (401)');
 // 4) Outbound EMAIL → governed egress → SENT (sandbox).
 const sendRes = await fetch(`${SERVICE}/v1/adapters/EMAIL/send`, {
  method: 'POST',
  headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` },
  body: JSON.stringify({ target: 'buyer@smoke', payload: { subject: 'Re: Need help', text: 'thanks, looking into it' } }),
 });
 const cmd = await sendRes.json();
 assert(cmd.status === 'SENT', `outbound email SENT via broker (status ${cmd.status})`);
 console.log('\nEmail adapter smoke: PASS');
 await prisma.$disconnect();
 process.exit(0);
@@ -0,0 +1,53 @@
 // P9 scaling smoke: cross-instance realtime fan-out via the socket.io Redis adapter.
 // Alice connects to instance A, Bob to instance B (two separate processes sharing one
 // Redis + DB). Alice sends on A; Bob must receive it live on B — which only works if the
 // Redis adapter fans room emits across replicas. Set A_URL / B_URL (default 3201/3202).
 import 'dotenv/config';
 import { io } from 'socket.io-client';
 import jwt from 'jsonwebtoken';
 const A_URL = process.env.A_URL ?? 'http://localhost:3201';
 const B_URL = process.env.B_URL ?? 'http://localhost:3202';
 const APP_ID = 'portal-demo';
 const SECRET = JSON.parse(process.env.APP_SECRETS ?? '{"portal-demo":"dev-secret"}')[APP_ID];
 const sign = (userId) =>
  jwt.sign({ sub: userId, name: userId, appId: APP_ID, orgId: `org_${APP_ID}` }, SECRET, { algorithm: 'HS256', expiresIn: '1h' });
 const connect = (url, userId) =>
  io(`${url}/message`, { auth: { token: sign(userId) }, transports: ['websocket'], forceNew: true });
 const once = (socket, event, ms = 8000) =>
  new Promise((res, rej) => {
    const t = setTimeout(() => rej(new Error(`timeout waiting for "${event}"`)), ms);
    socket.once(event, (v) => { clearTimeout(t); res(v); });
  });
 const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); };
 const alice = connect(A_URL, 'alice-cluster');
 const bob = connect(B_URL, 'bob-cluster');
 try {
  await Promise.all([once(alice, 'connect'), once(bob, 'connect')]);
  assert(true, `Alice connected to A (${A_URL}), Bob to B (${B_URL})`);
  // Alice opens a thread on instance A; Bob joins the SAME thread on instance B.
  const opened = await alice.emitWithAck('open_thread', {});
  const threadId = opened.threadId;
  assert(!!threadId, `Alice opened thread ${threadId} on instance A`);
  await bob.emitWithAck('open_thread', { threadId });
  assert(true, 'Bob joined the same thread on instance B');
  // Alice sends on A → Bob must receive it on B (cross-instance fan-out via Redis).
  const bobReceives = once(bob, 'message');
  await alice.emitWithAck('send_message', { threadId, content: 'hello across instances' });
  const received = await bobReceives;
  assert(received.content === 'hello across instances', `Bob received "${received.content}" on B — emitted from A ✓`);
  console.log('\nP9 cross-instance realtime smoke: PASS');
 } catch (err) {
  console.error('✗', err.message);
  process.exit(1);
 } finally {
  alice.close();
  bob.close();
 }
 process.exit(0);
@@ -0,0 +1,51 @@
 // P9 slice-9 retention smoke: an aged interaction's snapshot passes its delete window
 // and the sweep redacts the content in place (never a hard delete), visible as a
 // REDACTED snapshot on /metrics. Requires the service running with IIOS_DEV_TOKENS=1
 // and IIOS_RETENTION_DELETE_DAYS=0 IIOS_RETENTION_ARCHIVE_DAYS=0 (new content due at once;
 // scheduled interval left OFF so only the explicit dev sweep runs).
 import 'dotenv/config';
 const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200';
 const APP_ID = 'portal-demo';
 const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); };
 const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
 async function devToken(userId, orgId) {
  const r = await fetch(`${SERVICE}/v1/dev/token`, {
    method: 'POST',
    headers: { 'content-type': 'application/json' },
    body: JSON.stringify({ appId: APP_ID, userId, name: userId, orgId }),
  });
  if (!r.ok) throw new Error(`devToken ${r.status} (service with IIOS_DEV_TOKENS=1?)`);
  return (await r.json()).token;
 }
 async function api(token, path, method = 'GET', body) {
  const r = await fetch(`${SERVICE}${path}`, {
    method,
    headers: { 'content-type': 'application/json', authorization: `Bearer ${token}` },
    body: body ? JSON.stringify(body) : undefined,
  });
  if (!r.ok) throw new Error(`${method} ${path} ${r.status}: ${await r.text()}`);
  return r.json();
 }
 const redactedCount = (m) => m.retention?.byStatus?.REDACTED ?? 0;
 const token = await devToken('ret-a', 'org_ret_A');
 const base = redactedCount(await api(token, '/metrics'));
 console.log(`baseline REDACTED snapshots = ${base}`);
 // Inject content and give the relay a moment to normalize it into an interaction.
 await api(token, '/v1/dev/webhook/WEBHOOK', 'POST', { text: 'retention smoke content' });
 await sleep(2000);
 // Sweep: 0-day windows make every interaction's snapshot immediately due → redacted.
 const swept = await api(token, '/v1/dev/retention/sweep', 'POST');
 assert(swept.redacted >= 1, `sweep redacted the aged interaction(s) (redacted=${swept.redacted})`);
 const after = await api(token, '/metrics');
 assert(redactedCount(after) > base, `/metrics shows more REDACTED snapshots (${base} → ${redactedCount(after)})`);
 assert((after.retention?.total ?? 0) >= 1, 'retention.total reflects captured snapshots');
 console.log('\nP9 retention smoke: PASS');
 process.exit(0);
@@ -1,6 +1,6 @@
 import { Injectable } from '@nestjs/common';
 import type { ScopeVector } from '@insignia/iios-contracts';
-import { WebhookAdapter, type ChannelAdapter } from '@insignia/iios-adapter-sdk';
+import { WebhookAdapter, EmailAdapter, type ChannelAdapter } from '@insignia/iios-adapter-sdk';
 export interface AdapterConfig {
  secret: string;
@@ -8,10 +8,9 @@ export interface AdapterConfig {
 }
 /**
- * Registers channel adapters + their per-channel config (secret + scope). In P5
+ * Registers channel adapters + their per-channel config (secret + scope). EMAIL uses
- * one reference WebhookAdapter backs several channelTypes (WEBHOOK/EMAIL/WHATSAPP)
+ * the email-native EmailAdapter (RFC threading); the rest use the reference
- * so the fixtures exercise the same anti-corruption pipeline. Secrets come from
+ * WebhookAdapter. Secrets come from ADAPTER_SECRETS (JSON map); scope defaults to demo.
 * ADAPTER_SECRETS (JSON map); scope defaults to the demo scope.
 */
@Injectable()
 export class AdapterRegistry {
@@ -29,8 +28,9 @@ export class AdapterRegistry {
      appId: process.env.ADAPTER_APP ?? 'portal-demo',
    };
    for (const channelType of ['WEBHOOK', 'EMAIL', 'WHATSAPP', 'PORTAL']) {
      const adapter: ChannelAdapter = channelType === 'EMAIL' ? new EmailAdapter() : new WebhookAdapter(channelType);
      this.adapters.set(channelType, {
-        adapter: new WebhookAdapter(channelType),
+        adapter,
        config: { secret: secrets[channelType] ?? 'dev-adapter-secret', scope },
      });
    }
@@ -1,14 +1,16 @@
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { ConflictException } from '@nestjs/common';
 import { PrismaClient } from '@prisma/client';
 import { DlqService } from '../outbox/dlq.service';
 import { resetDb } from '../test-utils/reset-db';
 import { ProjectionCursorService } from '../projection/projection-cursor.service';
 import { makeFakePorts } from '@insignia/iios-testkit';
 import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
-import { signedFixture, emailFixture } from '@insignia/iios-adapter-sdk';
+import { signedFixture, emailFixture, emailInboundFixture, emailReplyFixture } from '@insignia/iios-adapter-sdk';
 import { AdapterRegistry } from './adapter.registry';
 import { InboundService } from './inbound.service';
 import { OutboundService } from './outbound.service';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { CapabilityProviderRegistry } from '../capability/capability.registry';
 import { RawEventProjector } from './raw-event.projector';
@@ -79,9 +81,60 @@ describe('Adapter inbound (P5)', () => {
  });
 });
 describe('Email adapter — inbound + threading', () => {
  const receiveEmail = async (payload: object) => {
    const { body, headers } = signedFixture(payload, SECRET);
    return inbound().receive('EMAIL', body, headers);
  };
  const normalizeAll = async () => {
    for (const ev of await rawReceivedEvents()) await projector().onRawReceived(ev);
  };
  it('a signed inbound email → a normalized interaction with the body on the email thread', async () => {
    const ack = await receiveEmail(emailInboundFixture);
    expect(ack.status).toBe('RECEIVED');
    await normalizeAll();
    const raw = await prisma.iiosInboundRawEvent.findUniqueOrThrow({ where: { id: ack.rawEventId } });
    expect(raw.status).toBe('NORMALIZED');
    const interaction = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: raw.interactionId! }, include: { parts: true, thread: true } });
    expect(interaction.parts[0]?.bodyText).toBe(emailInboundFixture.text);
    expect(interaction.thread?.externalThreadRef).toBe('email:<m1@example.com>');
  });
  it('a reply email (In-Reply-To) lands on the SAME email thread as the original', async () => {
    await receiveEmail(emailInboundFixture);
    await receiveEmail(emailReplyFixture);
    await normalizeAll();
    const interactions = await prisma.iiosInteraction.findMany({ orderBy: { occurredAt: 'asc' } });
    expect(interactions).toHaveLength(2);
    expect(interactions[0]!.threadId).toBe(interactions[1]!.threadId); // reply joined the thread
    expect(await prisma.iiosThread.count()).toBe(1); // one email conversation
  });
  it('redelivered email (same Message-ID) → one raw event, one interaction (dedup)', async () => {
    const svc = inbound();
    const { body, headers } = signedFixture(emailInboundFixture, SECRET);
    await svc.receive('EMAIL', body, headers);
    const second = await svc.receive('EMAIL', body, headers);
    expect(second.deduped).toBe(true);
    await normalizeAll();
    expect(await prisma.iiosInboundRawEvent.count()).toBe(1);
    expect(await prisma.iiosInteraction.count()).toBe(1);
  });
  it('bad signature on EMAIL → REJECTED, nothing ingested (fail-closed)', async () => {
    const { body } = signedFixture(emailInboundFixture, SECRET);
    await expect(inbound().receive('EMAIL', body, { 'x-iios-signature': 'sha256=bad' })).rejects.toThrow();
    expect(await prisma.iiosInboundRawEvent.count({ where: { status: 'REJECTED' } })).toBe(1);
    expect(await prisma.iiosInteraction.count()).toBe(0);
  });
 });
 describe('Adapter outbound (P5)', () => {
  const outbound = (limit?: number): OutboundService => {
-    const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()));
+    const s = new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService));
    if (limit) s.limit = limit;
    return s;
  };
@@ -112,7 +165,7 @@ describe('Adapter outbound (P5)', () => {
  it('broker obligation DENY_OUTBOUND → command BLOCKED, no send (P9)', async () => {
    const ports = makeFakePorts();
    ports.opa.setDecision({ allow: true, obligations: [{ kind: 'DENY_OUTBOUND', reason: 'recipient opted out' }] });
-    const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()));
+    const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()), new IdempotencyService(asService));
    const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'hi' }, 'blk-1');
    expect(cmd.status).toBe('BLOCKED');
    expect(cmd.providerRef).toBe('blocked');
@@ -121,7 +174,7 @@ describe('Adapter outbound (P5)', () => {
  it('CMP allow → command records the consent receipt (P9 slice 2)', async () => {
    const ports = makeFakePorts();
    ports.cmp.setStatus('ALLOW');
-    const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()));
+    const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()), new IdempotencyService(asService));
    const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'hi' }, 'consent-1', 'scope1', 'support');
    expect(cmd.status).toBe('SENT');
    expect(cmd.consentReceiptRef).toBe('fake-receipt');
@@ -130,7 +183,7 @@ describe('Adapter outbound (P5)', () => {
  it('CMP DENY for the purpose → command BLOCKED, no send (P9 slice 2)', async () => {
    const ports = makeFakePorts();
    ports.cmp.denyPurpose('marketing');
-    const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()));
+    const svc = new OutboundService(asService, new CapabilityBroker(ports, new CapabilityProviderRegistry()), new IdempotencyService(asService));
    const cmd = await svc.send('WEBHOOK', 'user@x', { text: 'promo' }, 'consent-2', 'scope1', 'marketing');
    expect(cmd.status).toBe('BLOCKED');
    expect(await prisma.iiosDeliveryAttempt.count({ where: { commandId: cmd.id, status: 'SENT' } })).toBe(0);
@@ -146,4 +199,18 @@ describe('Adapter outbound (P5)', () => {
    expect(a2.status).toBe('RATE_LIMITED');
    expect(b1.status).toBe('SENT');
  });
  it('idempotency ledger: same key + a different target → conflict, no second command (P9 slice 10)', async () => {
    const svc = outbound();
    const first = await svc.send('WEBHOOK', 'user@a', { text: 'hi' }, 'dup-key');
    expect(first.status).toBe('SENT');
    await expect(svc.send('WEBHOOK', 'user@b', { text: 'hi' }, 'dup-key')).rejects.toBeInstanceOf(ConflictException);
    expect(await prisma.iiosOutboundCommand.count({ where: { idempotencyKey: 'dup-key' } })).toBe(1);
  });
  it('opt-in: a keyless send creates a command but writes no idempotency-ledger row (P9 slice 10)', async () => {
    const cmd = await outbound().send('WEBHOOK', 'user@x', { text: 'hi' });
    expect(cmd.status).toBe('SENT');
    expect(await prisma.iiosIdempotencyCommand.count()).toBe(0);
  });
 });
@@ -28,7 +28,8 @@ export class InboundService {
    const { adapter, config } = reg;
    const payload = this.safeParse(rawBody);
-    const externalEventId = typeof payload?.eventId === 'string' ? payload.eventId : undefined;
+    // The adapter extracts its provider-native id (eventId, Message-ID, …) for replay dedup.
    const externalEventId = adapter.externalEventId?.(payload) ?? (typeof payload?.eventId === 'string' ? payload.eventId : undefined);
    // Fail-closed: a bad/absent signature is recorded and rejected — never ingested.
    if (!adapter.verifySignature(rawBody, headers, config.secret)) {
@@ -3,6 +3,7 @@ import { Injectable } from '@nestjs/common';
 import { Prisma } from '@prisma/client';
 import { PrismaService } from '../prisma/prisma.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { recordAudit } from '../observability/audit';
 /**
@@ -23,35 +24,41 @@ export class OutboundService {
  constructor(
    private readonly prisma: PrismaService,
    private readonly broker: CapabilityBroker,
    private readonly idempotency: IdempotencyService,
  ) {}
  async send(
    channelType: string,
    target: string,
    payload: Record<string, unknown>,
-    idempotencyKey: string = randomUUID(),
+    idempotencyKey?: string,
    scopeId?: string,
    purpose?: string,
  ) {
-    const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey } });
+    const key = idempotencyKey ?? randomUUID();
    // The actual send. The inner findUnique stays as a backstop so a FAILED-then-retried
    // command returns the existing row instead of colliding on idempotencyKey @unique.
    const body = async () => {
      const existing = await this.prisma.iiosOutboundCommand.findUnique({ where: { idempotencyKey: key } });
      if (existing) return existing;
      // Per-target rate limit AND per-tenant quota (a noisy tenant can't starve shared egress).
      if (!(await this.allow(channelType, target)) || !(await this.allowTenant(scopeId))) {
        const cmd = await this.prisma.iiosOutboundCommand.create({
-        data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'RATE_LIMITED', idempotencyKey },
+          data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'RATE_LIMITED', idempotencyKey: key },
        });
        await this.prisma.iiosDeliveryAttempt.create({ data: { commandId: cmd.id, attemptNo: 1, status: 'RATE_LIMITED' } });
        return cmd;
      }
      const cmd = await this.prisma.iiosOutboundCommand.create({
-      data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey },
+        data: { channelType, target, scopeId, payload: payload as Prisma.InputJsonValue, status: 'PENDING', idempotencyKey: key },
      });
      let result;
      try {
-      result = await this.broker.invoke({ capability: 'channel.send', channelType, target, payload, idempotencyKey, scopeId, purpose });
+        result = await this.broker.invoke({ capability: 'channel.send', channelType, target, payload, idempotencyKey: key, scopeId, purpose });
      } catch (err) {
        // Fail-closed (e.g. PolicyDeniedError): mark the command FAILED, record the attempt, propagate.
        await this.prisma.$transaction([
@@ -72,6 +79,12 @@ export class OutboundService {
      ]);
      await recordAudit(this.prisma, { action: 'channel.send', resourceType: 'outbound_command', resourceId: cmd.id, scopeId });
      return updated;
    };
    // Opt-in idempotency: an explicit key routes through the ledger (request-hash conflict → 409,
    // response replay); a keyless send just runs with a fresh uuid (old behavior).
    if (!idempotencyKey) return body();
    return this.idempotency.run({ scopeId, commandName: 'channel.send', key: idempotencyKey, request: { channelType, target, payload, purpose } }, body);
  }
  /** Per-tenant egress quota (P9): count this scope's commands in the window vs the cap. */
@@ -16,6 +16,7 @@ import { AiModule } from './ai/ai.module';
 import { CalendarModule } from './calendar/calendar.module';
 import { CapabilityModule } from './capability/capability.module';
 import { DsrModule } from './dsr/dsr.module';
 import { RetentionModule } from './retention/retention.module';
 import { HealthController } from './health.controller';
 import { MetricsController } from './observability/metrics.controller';
 import { DevController } from './dev/dev.controller';
@@ -39,6 +40,7 @@ import { DevController } from './dev/dev.controller';
    CalendarModule,
    CapabilityModule,
    DsrModule,
    RetentionModule,
  ],
  controllers: [HealthController, MetricsController, DevController],
 })
@@ -7,6 +7,7 @@ import { ActorResolver } from '../identity/actor.resolver';
 import { AiJobService } from '../ai/ai.service';
 import { AiBudgetGuard } from '../ai/ai-budget.guard';
 import { OutboundService } from '../adapters/outbound.service';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { CapabilityProviderRegistry } from '../capability/capability.registry';
 import { CalendarService } from './calendar.service';
@@ -22,7 +23,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
 const START = '2026-07-02T10:00:00.000Z';
 const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
-const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())));
+const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)));
 async function scheduledMeeting(attendees: { userId: string; visibility?: 'FULL' | 'NONE' }[]) {
  const m = await cal().scheduleDirect(alice, { meetingType: 'INTERNAL', title: 'Board sync', startAt: START, attendees });
@@ -11,6 +11,7 @@ import { ActorResolver } from '../identity/actor.resolver';
 import { AiJobService } from '../ai/ai.service';
 import { AiBudgetGuard } from '../ai/ai-budget.guard';
 import { OutboundService } from '../adapters/outbound.service';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { CapabilityProviderRegistry } from '../capability/capability.registry';
 import { CalendarService } from './calendar.service';
@@ -27,7 +28,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
 const START = '2026-07-02T10:00:00.000Z';
 const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
-const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())));
+const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)));
 async function makeInteraction(text: string): Promise<string> {
  const m = new MessageService(asService, makeFakePorts(), actors);
@@ -9,6 +9,7 @@ import { ActorResolver } from '../identity/actor.resolver';
 import { AiJobService } from '../ai/ai.service';
 import { AiBudgetGuard } from '../ai/ai-budget.guard';
 import { OutboundService } from '../adapters/outbound.service';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { CapabilityProviderRegistry } from '../capability/capability.registry';
 import { CalendarService } from './calendar.service';
@@ -24,7 +25,7 @@ const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'po
 const START = '2026-07-02T10:00:00.000Z';
 const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
-const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()));
+const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService));
 const cal = (ports?: FakePorts) => new CalendarService(asService, ports ?? makeFakePorts(), actors, ai(), outbound());
 async function makeInteraction(text: string): Promise<string> {
@@ -2,6 +2,7 @@ import { Injectable, NotFoundException } from '@nestjs/common';
 import type { CapabilityProvider } from '@insignia/iios-contracts';
 import { SandboxProvider } from './sandbox.provider';
 import { HttpProvider } from './http.provider';
 import { EmailProvider } from './email.provider';
 const DEFAULT_CHANNELS = ['WEBHOOK', 'EMAIL', 'WHATSAPP', 'PORTAL'];
@@ -19,7 +20,9 @@ export class CapabilityProviderRegistry {
    for (const ch of DEFAULT_CHANNELS) this.register(new SandboxProvider([ch]));
    for (const ch of DEFAULT_CHANNELS) {
      const url = process.env[`IIOS_PROVIDER_URL_${ch}`];
-      if (url) this.register(new HttpProvider(ch, url));
+      if (!url) continue;
      // EMAIL gets an email-shaped envelope provider; other channels use the generic HTTP one.
      this.register(ch === 'EMAIL' ? new EmailProvider(url) : new HttpProvider(ch, url));
    }
  }
@@ -0,0 +1,40 @@
 import type { CapabilityProvider, CapabilityRequest, ProviderResult } from '@insignia/iios-contracts';
 interface EmailPayload {
  subject?: string;
  text?: string;
  html?: string;
  inReplyTo?: string;
 }
 /**
 * Email egress provider: shapes the outbound command into an email envelope
 * ({to, subject, text, html, inReplyTo}) and POSTs it to a configured send endpoint
 * (a vendor mail API / SMTP relay behind a URL). Activated only when
 * IIOS_PROVIDER_URL_EMAIL is set; a transport failure is surfaced as FAILED, never thrown.
 */
 export class EmailProvider implements CapabilityProvider {
  readonly name = 'email-http';
  readonly channelTypes = ['EMAIL'];
  readonly capabilities = { canSend: true };
  constructor(private readonly url: string) {}
  async send(req: CapabilityRequest): Promise<ProviderResult> {
    const started = Date.now();
    const p = (req.payload ?? {}) as EmailPayload;
    const email = { to: req.target, subject: p.subject ?? '(no subject)', text: p.text, html: p.html, inReplyTo: p.inReplyTo };
    try {
      const res = await fetch(this.url, {
        method: 'POST',
        headers: { 'content-type': 'application/json', 'idempotency-key': req.idempotencyKey },
        body: JSON.stringify(email),
      });
      const latencyMs = Date.now() - started;
      if (!res.ok) return { providerRef: `email-${res.status}`, outcome: 'FAILED', errorCode: `HTTP_${res.status}`, latencyMs };
      return { providerRef: `email-${req.idempotencyKey}`, outcome: 'SENT', latencyMs };
    } catch (err) {
      return { providerRef: 'email-error', outcome: 'FAILED', errorCode: (err as Error).message.slice(0, 60), latencyMs: Date.now() - started };
    }
  }
 }
@@ -8,6 +8,7 @@ import { AdapterRegistry } from '../adapters/adapter.registry';
 import { PrismaService } from '../prisma/prisma.service';
 import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
 import { SessionVerifier } from '../platform/session.verifier';
 import { RetentionService } from '../retention/retention.service';
 /**
 * Dev-only helpers (gated by IIOS_DEV_TOKENS=1). Never enable in production.
@@ -20,6 +21,7 @@ export class DevController {
    private readonly prisma: PrismaService,
    private readonly actors: ActorResolver,
    private readonly session: SessionVerifier,
    private readonly retention: RetentionService,
  ) {}
  @Post('token')
@@ -96,6 +98,13 @@ export class DevController {
    return { eventId: id, scopeId: scope.id };
  }
  /** Run the retention sweep on demand (drives the retention smoke). */
  @Post('retention/sweep')
  async retentionSweep() {
    this.assertDev();
    return this.retention.sweep();
  }
  private principal(authorization?: string): MessagePrincipal {
    const token = (authorization ?? '').replace(/^Bearer\s+/i, '');
    if (!token) throw new BadRequestException('Authorization bearer token is required');
@@ -12,6 +12,7 @@ import { AiJobService } from './ai/ai.service';
 import { AiBudgetGuard } from './ai/ai-budget.guard';
 import { CalendarService } from './calendar/calendar.service';
 import { OutboundService } from './adapters/outbound.service';
 import { IdempotencyService } from './idempotency/idempotency.service';
 import { CapabilityBroker } from './capability/capability.broker';
 import { CapabilityProviderRegistry } from './capability/capability.registry';
 import type { PrismaService } from './prisma/prisma.service';
@@ -27,7 +28,7 @@ const B: MessagePrincipal = { userId: 'b1', orgId: 'org_B', appId: 'portal-demo'
 const START = '2026-07-02T10:00:00.000Z';
 const ai = () => new AiJobService(asService, makeFakePorts(), makeFakeInference(), new AiBudgetGuard(asService), actors);
-const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()));
+const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService));
 const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), outbound(), actors);
 const cal = () => new CalendarService(asService, makeFakePorts(), actors, ai(), outbound());
@@ -4,6 +4,7 @@ import { NestFactory } from '@nestjs/core';
 import { ValidationPipe } from '@nestjs/common';
 import { AppModule } from './app.module';
 import { traceMiddleware } from './observability/trace.middleware';
 import { RedisIoAdapter } from './realtime/redis-io.adapter';
 async function bootstrap(): Promise<void> {
  const app = await NestFactory.create(AppModule, { rawBody: true });
@@ -12,6 +13,15 @@ async function bootstrap(): Promise<void> {
    new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }),
  );
  app.enableCors({ origin: true, credentials: true });
  // Multi-replica realtime: fan socket.io room emits across instances via Redis.
  // Opt-in — without REDIS_URL the default in-memory adapter is used (single instance).
  if (process.env.REDIS_URL) {
    const redisAdapter = new RedisIoAdapter(app, process.env.REDIS_URL);
    await redisAdapter.connect();
    app.useWebSocketAdapter(redisAdapter);
  }
  const port = Number(process.env.PORT ?? 3200);
  await app.listen(port);
  // eslint-disable-next-line no-console
@@ -11,6 +11,7 @@ import { RouteService } from '../routing/route.service';
 import { AiJobService } from '../ai/ai.service';
 import { AiBudgetGuard } from '../ai/ai-budget.guard';
 import { OutboundService } from '../adapters/outbound.service';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { CapabilityProviderRegistry } from '../capability/capability.registry';
 import { runWithTrace } from './trace-context';
@@ -23,7 +24,7 @@ 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);
-const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()));
+const outbound = () => new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService));
 const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), outbound(), actors);
 async function makeInteraction(text: string): Promise<{ interactionId: string; scopeId: string }> {
@@ -3,6 +3,7 @@ import { PrismaService } from '../prisma/prisma.service';
 import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
 import { SessionVerifier } from '../platform/session.verifier';
 import { ProjectionCursorService } from '../projection/projection-cursor.service';
 import { RetentionService } from '../retention/retention.service';
 /**
 * Tenant-scoped metrics (P9 observability, JSON). Returns the CALLER's tenant
@@ -17,6 +18,7 @@ export class MetricsController {
    private readonly actors: ActorResolver,
    private readonly session: SessionVerifier,
    private readonly cursors: ProjectionCursorService,
    private readonly retention: RetentionService,
  ) {}
  @Get()
@@ -59,6 +61,7 @@ export class MetricsController {
      tenant,
      relay: { outboxPending, oldestPendingMs },
      projections,
      retention: await this.retention.summary(),
    };
  }
@@ -0,0 +1,46 @@
 import { Logger, type INestApplicationContext } from '@nestjs/common';
 import { IoAdapter } from '@nestjs/platform-socket.io';
 import { createAdapter } from '@socket.io/redis-adapter';
 import { Redis } from 'ioredis';
 import type { Server, ServerOptions } from 'socket.io';
 /**
 * Socket.io adapter backed by Redis pub/sub (P9 scaling). With the default in-memory
 * adapter, `server.to(room).emit()` only reaches sockets on the SAME instance; behind a
 * load balancer a client on replica B never sees a message emitted by replica A. This
 * adapter fans room emits across every replica through Redis, so realtime works at N>1.
 * Wired only when REDIS_URL is set (single-instance dev keeps the in-memory adapter).
 */
 export class RedisIoAdapter extends IoAdapter {
  private readonly logger = new Logger(RedisIoAdapter.name);
  private adapterConstructor?: ReturnType<typeof createAdapter>;
  private clients: Redis[] = [];
  constructor(
    app: INestApplicationContext,
    private readonly url: string,
  ) {
    super(app);
  }
  /** Establish the pub/sub clients + the socket.io Redis adapter factory. */
  async connect(): Promise<void> {
    const pubClient = new Redis(this.url, { maxRetriesPerRequest: null });
    const subClient = pubClient.duplicate();
    this.clients = [pubClient, subClient];
    pubClient.on('error', (err) => this.logger.error(`redis pub error: ${err.message}`));
    subClient.on('error', (err) => this.logger.error(`redis sub error: ${err.message}`));
    this.adapterConstructor = createAdapter(pubClient, subClient);
    this.logger.log('socket.io Redis adapter connected — realtime fans out across replicas');
  }
  createIOServer(port: number, options?: ServerOptions): Server {
    const server = super.createIOServer(port, options) as Server;
    if (this.adapterConstructor) server.adapter(this.adapterConstructor);
    return server;
  }
  async dispose(): Promise<void> {
    await Promise.all(this.clients.map((c) => c.quit().catch(() => undefined)));
  }
 }
@@ -0,0 +1,10 @@
 import { Global, Module } from '@nestjs/common';
 import { RetentionService } from './retention.service';
 /** Data-retention sweep, global so the dev controller + /metrics can reach it. */
@Global()
@Module({
  providers: [RetentionService],
  exports: [RetentionService],
 })
 export class RetentionModule {}
@@ -0,0 +1,119 @@
 import { randomUUID } from 'node:crypto';
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { PrismaClient } from '@prisma/client';
 import { makeFakePorts } from '@insignia/iios-testkit';
 import { resetDb } from '../test-utils/reset-db';
 import { ActorResolver } from '../identity/actor.resolver';
 import { IngestService } from '../interactions/ingest.service';
 import { RetentionService } from './retention.service';
 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 svc = () => new RetentionService(asService);
 async function ingest(text: string, orgId = 'org_demo'): Promise<{ interactionId: string; scopeId: string }> {
  const res = await new IngestService(asService, makeFakePorts(), actors).ingest(
    {
      scope: { orgId, appId: 'portal-demo' },
      channel: { type: 'WEBHOOK', externalChannelId: 'webhook' },
      source: { handleKind: 'EXTERNAL_VISITOR', externalId: 'ext-1' },
      kind: 'MESSAGE',
      thread: { externalThreadId: `wh-${randomUUID().slice(0, 6)}` },
      parts: [{ kind: 'TEXT', bodyText: text }],
      occurredAt: new Date().toISOString(),
      providerEventId: `pe-${randomUUID()}`,
    },
    `ing-${randomUUID()}`,
  );
  const it = await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: res.interactionId } });
  return { interactionId: res.interactionId, scopeId: it.scopeId };
 }
 const past = new Date(Date.now() - 86_400_000);
 const future = new Date(Date.now() + 86_400_000);
 const setSnapshot = (targetId: string, data: { archiveAfter?: Date; deleteAfter?: Date }) =>
  prisma.iiosRetentionPolicySnapshot.update({ where: { targetType_targetId: { targetType: 'interaction', targetId } }, data });
 const bodyOf = async (interactionId: string) => (await prisma.iiosMessagePart.findFirstOrThrow({ where: { interactionId } })).bodyText;
 beforeAll(async () => { await prisma.$connect(); });
 afterAll(async () => { await prisma.$disconnect(); });
 beforeEach(async () => { await resetDb(prisma); });
 describe('RetentionService (P9 — retention sweep)', () => {
  it('ensureSnapshots captures one ACTIVE snapshot per interaction with archive < delete', async () => {
    const { interactionId } = await ingest('hello');
    await svc().ensureSnapshots();
    const snap = await prisma.iiosRetentionPolicySnapshot.findUniqueOrThrow({ where: { targetType_targetId: { targetType: 'interaction', targetId: interactionId } } });
    expect(snap.status).toBe('ACTIVE');
    expect(snap.dataClass).toBe('internal');
    expect(snap.archiveAfter.getTime()).toBeLessThan(snap.deleteAfter.getTime());
  });
  it('redacts an interaction whose delete window has passed', async () => {
    const { interactionId } = await ingest('my secret');
    await svc().ensureSnapshots();
    await setSnapshot(interactionId, { archiveAfter: past, deleteAfter: past });
    const res = await svc().applySweep();
    expect(res.redacted).toBe(1);
    expect(await bodyOf(interactionId)).toBe('[redacted]');
    expect((await prisma.iiosInteraction.findUniqueOrThrow({ where: { id: interactionId } })).status).toBe('REDACTED');
    expect((await prisma.iiosRetentionPolicySnapshot.findUniqueOrThrow({ where: { targetType_targetId: { targetType: 'interaction', targetId: interactionId } } })).status).toBe('REDACTED');
    expect(await prisma.iiosAuditLink.count({ where: { action: 'retention.redacted' } })).toBe(1);
  });
  it('archives (not deletes) when only the archive window has passed, then deletes later', async () => {
    const { interactionId } = await ingest('still needed');
    await svc().ensureSnapshots();
    await setSnapshot(interactionId, { archiveAfter: past, deleteAfter: future });
    let res = await svc().applySweep();
    expect(res).toMatchObject({ archived: 1, redacted: 0 });
    expect(await bodyOf(interactionId)).toBe('still needed'); // content untouched by archive
    await setSnapshot(interactionId, { deleteAfter: past });
    res = await svc().applySweep();
    expect(res.redacted).toBe(1);
    expect(await bodyOf(interactionId)).toBe('[redacted]');
  });
  it('an active compliance hold blocks the sweep until released', async () => {
    const { interactionId, scopeId } = await ingest('held content');
    await svc().ensureSnapshots();
    await setSnapshot(interactionId, { archiveAfter: past, deleteAfter: past });
    const hold = await prisma.iiosComplianceHold.create({ data: { scopeId, targetType: 'message', targetId: interactionId, holdReason: 'legal' } });
    let res = await svc().applySweep();
    expect(res).toMatchObject({ redacted: 0, skippedHeld: 1 });
    expect(await bodyOf(interactionId)).toBe('held content');
    await prisma.iiosComplianceHold.update({ where: { id: hold.id }, data: { status: 'RELEASED' } });
    res = await svc().applySweep();
    expect(res.redacted).toBe(1);
    expect(await bodyOf(interactionId)).toBe('[redacted]');
  });
  it('is idempotent — a REDACTED snapshot is terminal', async () => {
    const { interactionId } = await ingest('x');
    await svc().ensureSnapshots();
    await setSnapshot(interactionId, { archiveAfter: past, deleteAfter: past });
    await svc().applySweep();
    expect(await svc().applySweep()).toMatchObject({ archived: 0, redacted: 0, skippedHeld: 0 });
  });
  it('sweep is tenant-scoped', async () => {
    const a = await ingest('tenant a', 'org_A');
    const b = await ingest('tenant b', 'org_B');
    await svc().ensureSnapshots();
    await setSnapshot(a.interactionId, { archiveAfter: past, deleteAfter: past });
    await setSnapshot(b.interactionId, { archiveAfter: past, deleteAfter: past });
    const res = await svc().applySweep(a.scopeId);
    expect(res.redacted).toBe(1);
    expect(await bodyOf(a.interactionId)).toBe('[redacted]');
    expect(await bodyOf(b.interactionId)).toBe('tenant b'); // scope B untouched
  });
 });
@@ -0,0 +1,127 @@
 import { Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common';
 import { PrismaService } from '../prisma/prisma.service';
 import { recordAudit } from '../observability/audit';
 const DAY_MS = 86_400_000;
 export interface SweepResult {
  archived: number;
  redacted: number;
  skippedHeld: number;
 }
 /**
 * Data retention (P9). Each interaction gets a per-resource retention snapshot with
 * frozen archive/delete timestamps; a scheduled sweep archives (status change) then
 * deletes = redacts-in-place (reusing the DSR tombstone semantics) once a resource ages
 * past its window. An active compliance hold blocks both. Never a hard delete.
 */
@Injectable()
 export class RetentionService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(RetentionService.name);
  private timer?: ReturnType<typeof setInterval>;
  constructor(private readonly prisma: PrismaService) {}
  onModuleInit(): void {
    const ms = Number(process.env.IIOS_RETENTION_SWEEP_INTERVAL_MS ?? 0);
    if (ms > 0) {
      this.timer = setInterval(() => {
        void this.sweep().catch((err) => this.logger.warn(`retention sweep failed: ${(err as Error).message}`));
      }, ms);
    }
  }
  onModuleDestroy(): void {
    if (this.timer) clearInterval(this.timer);
  }
  private windowDays(dataClass: string, kind: 'ARCHIVE' | 'DELETE'): number {
    const cls = process.env[`IIOS_RETENTION_${kind}_DAYS_${dataClass.toUpperCase()}`];
    const glob = process.env[`IIOS_RETENTION_${kind}_DAYS`];
    return Number(cls ?? glob ?? (kind === 'ARCHIVE' ? 90 : 365));
  }
  /** Lazily capture a per-resource snapshot for any interaction that lacks one. */
  async ensureSnapshots(scopeId?: string): Promise<number> {
    const interactions = await this.prisma.iiosInteraction.findMany({
      where: scopeId ? { scopeId } : {},
      select: { id: true, scopeId: true, dataClass: true, receivedAt: true },
    });
    let created = 0;
    for (const i of interactions) {
      const exists = await this.prisma.iiosRetentionPolicySnapshot.findUnique({
        where: { targetType_targetId: { targetType: 'interaction', targetId: i.id } },
      });
      if (exists) continue;
      const base = i.receivedAt.getTime();
      await this.prisma.iiosRetentionPolicySnapshot.create({
        data: {
          policyKey: `default:${i.dataClass}:v1`,
          scopeSnapshotId: i.scopeId,
          targetType: 'interaction',
          targetId: i.id,
          dataClass: i.dataClass,
          archiveAfter: new Date(base + this.windowDays(i.dataClass, 'ARCHIVE') * DAY_MS),
          deleteAfter: new Date(base + this.windowDays(i.dataClass, 'DELETE') * DAY_MS),
          sourceVersion: process.env.IIOS_RETENTION_POLICY_VERSION ?? 'v1',
        },
      });
      created++;
    }
    return created;
  }
  /** Act on due snapshots: archive, or delete (redact-in-place), honoring compliance holds. */
  async applySweep(scopeId?: string): Promise<SweepResult> {
    const now = new Date();
    const due = await this.prisma.iiosRetentionPolicySnapshot.findMany({
      where: { status: { in: ['ACTIVE', 'ARCHIVED'] }, archiveAfter: { lte: now }, ...(scopeId ? { scopeSnapshotId: scopeId } : {}) },
    });
    const result: SweepResult = { archived: 0, redacted: 0, skippedHeld: 0 };
    for (const s of due) {
      const held = await this.prisma.iiosComplianceHold.findFirst({
        where: { targetId: s.targetId, status: 'ACTIVE', OR: [{ expiresAt: null }, { expiresAt: { gt: now } }] },
      });
      if (held) {
        result.skippedHeld++;
        continue;
      }
      if (s.deleteAfter <= now) {
        await this.prisma.$transaction([
          this.prisma.iiosMessagePart.updateMany({ where: { interactionId: s.targetId }, data: { bodyText: '[redacted]', contentRef: null } }),
          this.prisma.iiosInboundRawEvent.updateMany({ where: { interactionId: s.targetId }, data: { payload: { redacted: true } } }),
          this.prisma.iiosInteraction.update({ where: { id: s.targetId }, data: { status: 'REDACTED' } }),
          this.prisma.iiosRetentionPolicySnapshot.update({ where: { id: s.id }, data: { status: 'REDACTED' } }),
        ]);
        await recordAudit(this.prisma, { action: 'retention.redacted', resourceType: 'interaction', resourceId: s.targetId, scopeId: s.scopeSnapshotId });
        result.redacted++;
      } else if (s.status === 'ACTIVE') {
        await this.prisma.iiosRetentionPolicySnapshot.update({ where: { id: s.id }, data: { status: 'ARCHIVED' } });
        await recordAudit(this.prisma, { action: 'retention.archived', resourceType: 'interaction', resourceId: s.targetId, scopeId: s.scopeSnapshotId });
        result.archived++;
      }
    }
    return result;
  }
  /** Full sweep: capture missing snapshots, then act on due ones. */
  async sweep(scopeId?: string): Promise<SweepResult> {
    await this.ensureSnapshots(scopeId);
    return this.applySweep(scopeId);
  }
  /** Snapshot lifecycle counts for /metrics (global ops). */
  async summary(): Promise<{ total: number; byStatus: Record<string, number> }> {
    const groups = await this.prisma.iiosRetentionPolicySnapshot.groupBy({ by: ['status'], _count: true });
    const byStatus: Record<string, number> = {};
    let total = 0;
    for (const g of groups) {
      byStatus[g.status] = g._count;
      total += g._count;
    }
    return { total, byStatus };
  }
 }
@@ -13,6 +13,7 @@ import { RouteService } from './route.service';
 import { RouteProjector } from './route.projector';
 import { IngestService } from '../interactions/ingest.service';
 import { OutboundService } from '../adapters/outbound.service';
 import { IdempotencyService } from '../idempotency/idempotency.service';
 import { CapabilityBroker } from '../capability/capability.broker';
 import { CapabilityProviderRegistry } from '../capability/capability.registry';
 import { OutboxBus } from '../outbox/outbox.bus';
@@ -106,7 +107,7 @@ describe('Route simulator (P6, preview-first)', () => {
 describe('RouteService approve/deny → execute (P6)', () => {
  const admin: MessagePrincipal = { userId: 'admin', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Admin' };
-  const svc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors);
+  const svc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)), actors);
  it('approving a REVIEW decision executes the forward to the sandbox', async () => {
    const { interactionId, scopeId } = await makeInteraction('party at 9 PM');
@@ -160,7 +161,7 @@ describe('RouteService approve/deny → execute (P6)', () => {
 });
 describe('RouteProjector — AUTOMATIC forwarding (P6)', () => {
-  const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry())), actors);
+  const routeSvc = () => new RouteService(asService, new RouteSimulator(asService, new RuleEngine()), new OutboundService(asService, new CapabilityBroker(makeFakePorts(), new CapabilityProviderRegistry()), new IdempotencyService(asService)), actors);
  const projector = () => new RouteProjector(asService, new OutboxBus(), new DlqService(asService), routeSvc(), new ProjectionCursorService(asService));
  // Ingest an inbound (channel-bearing) interaction — routable, unlike native DMs.
@@ -8,7 +8,7 @@ import type { PrismaClient } from '@prisma/client';
 export async function resetDb(prisma: PrismaClient): Promise<void> {
  await prisma.$executeRawUnsafe(
    `TRUNCATE TABLE
-      "IiosComplianceHold","IiosProjectionCursor","IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink",
+      "IiosRetentionPolicySnapshot","IiosComplianceHold","IiosProjectionCursor","IiosIdempotencyCommand","IiosDlqItem","IiosAuditLink",
      "IiosMeetingActionItem","IiosActionItem","IiosTranscriptSegment","IiosMeetingTranscript","IiosMeetingSummary","IiosMeetingParticipant","IiosMeeting","IiosMeetingRequest",
      "IiosCalendarSyncCursor","IiosCalendarEvent","IiosCalendarProviderAccount","IiosAvailabilityWindow",
      "IiosAiEvidenceLink","IiosAiClaim","IiosAiToolCall","IiosAiArtifact","IiosAiModelRun","IiosAiJob","IiosEmbeddingRef",
@@ -352,6 +352,9 @@ importers:
      '@prisma/client':
        specifier: ^6.2.1
        version: 6.19.3(prisma@6.19.3(typescript@5.9.3))(typescript@5.9.3)
      '@socket.io/redis-adapter':
        specifier: ^8.3.0
        version: 8.3.0(socket.io-adapter@2.5.8)
      class-transformer:
        specifier: ^0.5.1
        version: 0.5.1
@@ -361,9 +364,15 @@ importers:
      dotenv:
        specifier: ^16.4.7
        version: 16.6.1
      ioredis:
        specifier: ^5.11.1
        version: 5.11.1
      jsonwebtoken:
        specifier: ^9.0.3
        version: 9.0.3
      prisma:
        specifier: ^6.2.1
        version: 6.19.3(typescript@5.9.3)
      reflect-metadata:
        specifier: ^0.2.2
        version: 0.2.2
@@ -392,9 +401,6 @@ importers:
      '@types/node':
        specifier: ^26.0.1
        version: 26.0.1
      prisma:
        specifier: ^6.2.1
        version: 6.19.3(typescript@5.9.3)
      socket.io-client:
        specifier: ^4.8.3
        version: 4.8.3
@@ -1164,6 +1170,9 @@ packages:
      '@types/node':
        optional: true
  '@ioredis/commands@1.10.0':
    resolution: {integrity: sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==}
  '@jridgewell/gen-mapping@0.3.13':
    resolution: {integrity: sha512-2kkt/7niJ6MgEPxF0bYdQ6etZaA+fQvDcLKckhy1yIQOzaoKjBBjSj63/aLVjYE3qhRt5dvM+uUyfCg6UKCBbA==}
@@ -1426,6 +1435,12 @@ packages:
  '@socket.io/component-emitter@3.1.2':
    resolution: {integrity: sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==}
  '@socket.io/redis-adapter@8.3.0':
    resolution: {integrity: sha512-ly0cra+48hDmChxmIpnESKrc94LjRL80TEmZVscuQ/WWkRP81nNj8W8cCGMqbI4L6NCuAaPRSzZF1a9GlAxxnA==}
    engines: {node: '>=10.0.0'}
    peerDependencies:
      socket.io-adapter: ^2.5.4
  '@standard-schema/spec@1.1.0':
    resolution: {integrity: sha512-l2aFy5jALhniG5HgqrD6jXLi/rUWrKvqN/qJx6yoJsgKhblVd+iqqU4RCXavm/jPityDo5TCvKMnpjKnOriy0w==}
@@ -1834,6 +1849,10 @@ packages:
    resolution: {integrity: sha512-JQHZ2QMW6l3aH/j6xCqQThY/9OH4D/9ls34cgkUBiEeocRTU04tHfKPBsUK1PqZCUQM7GiA0IIXJSuXHI64Kbg==}
    engines: {node: '>=0.8'}
  cluster-key-slot@1.1.1:
    resolution: {integrity: sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==}
    engines: {node: '>=0.10.0'}
  color-convert@2.0.1:
    resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==}
    engines: {node: '>=7.0.0'}
@@ -1908,6 +1927,15 @@ packages:
  csstype@3.2.3:
    resolution: {integrity: sha512-z1HGKcYy2xA8AGQfwrn0PAy+PB7X/GSj3UVJW9qKyn43xWa+gl5nXmU4qqLMRzWVLFC8KusUX8T/0kCiOYpAIQ==}
  debug@4.3.7:
    resolution: {integrity: sha512-Er2nc/H7RrMXZBFCEim6TCmMk02Z8vLC2Rbi1KEBggpo0fS6l0S1nnapwmIi3yW/+GOJap1Krg4w0Hg80oCqgQ==}
    engines: {node: '>=6.0'}
    peerDependencies:
      supports-color: '*'
    peerDependenciesMeta:
      supports-color:
        optional: true
  debug@4.4.3:
    resolution: {integrity: sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==}
    engines: {node: '>=6.0'}
@@ -1935,6 +1963,10 @@ packages:
  defu@6.1.7:
    resolution: {integrity: sha512-7z22QmUWiQ/2d0KkdYmANbRUVABpZ9SNYyH5vx6PZ+nE5bcC0l7uFvEfHlyld/HcGBFTL536ClDt3DEcSlEJAQ==}
  denque@2.1.0:
    resolution: {integrity: sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==}
    engines: {node: '>=0.10'}
  depd@2.0.0:
    resolution: {integrity: sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw==}
    engines: {node: '>= 0.8'}
@@ -2200,6 +2232,10 @@ packages:
  inherits@2.0.4:
    resolution: {integrity: sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==}
  ioredis@5.11.1:
    resolution: {integrity: sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==}
    engines: {node: '>=12.22.0'}
  ipaddr.js@1.9.1:
    resolution: {integrity: sha512-0KI/607xoxSToH7GjN1FfSbLoU0+btTicjsQSWQlh/hZykN8KpmMf7uYwPW3R+akZ6R/w18ZlXSHBYXiYUPO3g==}
    engines: {node: '>= 0.10'}
@@ -2452,6 +2488,9 @@ packages:
    resolution: {integrity: sha512-J6l92tKHX6w8Jy5nO1Vuc01NoIiRGi/d6qBKVxh+IQ8Cr3b6HbVNfKiF8ZpFKufTwpwxMmce2W3iQZ861ZRyTg==}
    engines: {node: '>=18'}
  notepack.io@3.0.1:
    resolution: {integrity: sha512-TKC/8zH5pXIAMVQio2TvVDTtPRX+DJPHDqjRbxogtFiByHyzKmy96RA0JtCQJ+WouyyL4A10xomQzgbUT+1jCg==}
  nypm@0.6.8:
    resolution: {integrity: sha512-Q9K4Diu6l5u6xJQogeFSs/zKtyMSgFKFtRQV+tHP4kL7KPm2grpBU0dFIwFaXwNxN0MtfKWc43VpCugAa+LPsw==}
    engines: {node: '>=18'}
@@ -2620,6 +2659,14 @@ packages:
    resolution: {integrity: sha512-GDhwkLfywWL2s6vEjyhri+eXmfH6j1L7JE27WhqLeYzoh/A3DBaYGEj2H/HFZCn/kMfim73FXxEJTw06WtxQwg==}
    engines: {node: '>= 14.18.0'}
  redis-errors@1.2.0:
    resolution: {integrity: sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==}
    engines: {node: '>=4'}
  redis-parser@3.0.0:
    resolution: {integrity: sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==}
    engines: {node: '>=4'}
  reflect-metadata@0.2.2:
    resolution: {integrity: sha512-urBwgfrvVP/eAyXx4hluJivBKzuEbSQs9rKWCrCkbSxNv8mxPcUZKeuoF3Uy4mJl3Lwprp6yy5/39VWigZ4K6Q==}
@@ -2754,6 +2801,9 @@ packages:
  stackback@0.0.2:
    resolution: {integrity: sha512-1XMJE5fQo1jGH6Y/7ebnwPOBEkIEnT4QF32d5R1+VXdXveM0IBMJt8zfaxX1P3QhVwrYe+576+jkANtSS2mBbw==}
  standard-as-callback@2.1.0:
    resolution: {integrity: sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==}
  statuses@2.0.2:
    resolution: {integrity: sha512-DvEy55V3DB7uknRo+4iOGT5fP1slR8wQohVdknigZPMpMstaKJQWhwiYBACJE3Ul2pTnATihhBYnRhZQHGBiRw==}
    engines: {node: '>= 0.8'}
@@ -2953,6 +3003,10 @@ packages:
  ufo@1.6.4:
    resolution: {integrity: sha512-JFNbkD1Svwe0KvGi8GOeLcP4kAWQ609twvCdcHxq1oSL8svv39ZuSvajcD8B+5D0eL4+s1Is2D/O6KN3qcTeRA==}
  uid2@1.0.0:
    resolution: {integrity: sha512-+I6aJUv63YAcY9n4mQreLUt0d4lvwkkopDNmpomkAUz0fAkEMV9pRWxN0EjhW1YfRhcuyHg2v3mwddCDW1+LFQ==}
    engines: {node: '>= 4.0.0'}
  uid@2.0.2:
    resolution: {integrity: sha512-u3xV3X7uzvi5b1MncmZo3i2Aw222Zk1keqLA1YkHldREkAhAqi65wuPfe7lHx8H/Wzy+8CE7S7uS3jekIM5s8g==}
    engines: {node: '>=8'}
@@ -3716,6 +3770,8 @@ snapshots:
    optionalDependencies:
      '@types/node': 26.0.1
  '@ioredis/commands@1.10.0': {}
  '@jridgewell/gen-mapping@0.3.13':
    dependencies:
      '@jridgewell/sourcemap-codec': 1.5.5
@@ -3968,6 +4024,15 @@ snapshots:
  '@socket.io/component-emitter@3.1.2': {}
  '@socket.io/redis-adapter@8.3.0(socket.io-adapter@2.5.8)':
    dependencies:
      debug: 4.3.7
      notepack.io: 3.0.1
      socket.io-adapter: 2.5.8
      uid2: 1.0.0
    transitivePeerDependencies:
      - supports-color
  '@standard-schema/spec@1.1.0': {}
  '@tokenizer/inflate@0.4.1':
@@ -4446,6 +4511,8 @@ snapshots:
  clone@1.0.4: {}
  cluster-key-slot@1.1.1: {}
  color-convert@2.0.1:
    dependencies:
      color-name: 1.1.4
@@ -4504,6 +4571,10 @@ snapshots:
  csstype@3.2.3: {}
  debug@4.3.7:
    dependencies:
      ms: 2.1.3
  debug@4.4.3:
    dependencies:
      ms: 2.1.3
@@ -4520,6 +4591,8 @@ snapshots:
  defu@6.1.7: {}
  denque@2.1.0: {}
  depd@2.0.0: {}
  destr@2.0.5: {}
@@ -4900,6 +4973,18 @@ snapshots:
  inherits@2.0.4: {}
  ioredis@5.11.1:
    dependencies:
      '@ioredis/commands': 1.10.0
      cluster-key-slot: 1.1.1
      debug: 4.4.3
      denque: 2.1.0
      redis-errors: 1.2.0
      redis-parser: 3.0.0
      standard-as-callback: 2.1.0
    transitivePeerDependencies:
      - supports-color
  ipaddr.js@1.9.1: {}
  is-arrayish@0.2.1: {}
@@ -5105,6 +5190,8 @@ snapshots:
  node-releases@2.0.50: {}
  notepack.io@3.0.1: {}
  nypm@0.6.8:
    dependencies:
      citty: 0.2.2
@@ -5258,6 +5345,12 @@ snapshots:
  readdirp@4.1.2: {}
  redis-errors@1.2.0: {}
  redis-parser@3.0.0:
    dependencies:
      redis-errors: 1.2.0
  reflect-metadata@0.2.2: {}
  require-from-string@2.0.2: {}
@@ -5460,6 +5553,8 @@ snapshots:
  stackback@0.0.2: {}
  standard-as-callback@2.1.0: {}
  statuses@2.0.2: {}
  std-env@3.10.0: {}
@@ -5624,6 +5719,8 @@ snapshots:
  ufo@1.6.4: {}
  uid2@1.0.0: {}
  uid@2.0.2:
    dependencies:
      '@lukeed/csprng': 1.1.0
Author	SHA1	Message	Date
maaz519	f0ea1ca553	feat(iios): email adapter dedup fix + smoke + verification CI / build (push) Failing after 2m8s Details Adds adapter-owned externalEventId extraction to the ChannelAdapter contract (WebhookAdapter→eventId, EmailAdapter→Message-ID) and uses it in InboundService, so email replays dedup on Message-ID (previously only payload.eventId worked). Adds an email-dedup spec + smoke-email.mjs proving inbound → normalized, a reply joins the same email thread, bad signature rejected, and outbound EMAIL SENT. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 20:34:26 +05:30
maaz519	f0afca89e3	feat(iios): register EmailAdapter for EMAIL + email-shaped egress provider AdapterRegistry now backs the EMAIL channel with the email-native EmailAdapter (RFC threading) instead of the generic WebhookAdapter. Adds EmailProvider — an email-envelope egress provider ({to,subject,text,html,inReplyTo}) registered for EMAIL when IIOS_PROVIDER_URL_EMAIL is set (sandbox stays the default). Spec: signed email → normalized interaction on the email thread; a reply joins the same thread (one conversation); bad signature → rejected (fail-closed). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 20:28:35 +05:30
maaz519	ae50f56404	feat(iios): EmailAdapter (email-native normalize + threading) in adapter-sdk Adds a canonical EmailInboundPayload + EmailAdapter implementing the ChannelAdapter contract: HMAC signature verify + normalize with email-native threading (whole reply chain → one IIOS thread via references/inReplyTo root), messageId as providerEventId for dedup, html→text fallback, Re:/Fwd: stripping. Plus emailInboundFixture/emailReplyFixture and unit tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 20:25:00 +05:30
maaz519	427396653f	feat(iios): socket.io Redis adapter for cross-replica realtime (P9 scaling) Adds an opt-in RedisIoAdapter (wired in main.ts when REDIS_URL is set) so socket.io room emits fan out across instances via Redis pub/sub — a client on replica B now receives messages emitted by replica A. Without REDIS_URL the in-memory adapter is kept (single-instance dev unchanged). Adds redis to docker-compose, REDIS_URL to the env contract, and smoke-realtime-cluster.mjs which proves cross-instance delivery fails without Redis and passes with it. Delivery is at-least-once at N>1; clients dedupe by message id (documented). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 19:28:11 +05:30
maaz519	a520b26398	chore(iios): production Dockerfile + migrate entrypoint + env/secrets contract Adds a multi-stage Dockerfile for iios-service (build → pnpm deploy prune → slim non-root runtime), a docker-entrypoint that runs `prisma migrate deploy` then starts the server as PID 1, a .dockerignore, a fully-commented .env.example config contract, and docs/DEPLOYMENT.md (topology, scaling, release strategy, prod-readiness gaps). Moves prisma to dependencies so the CLI ships in the prod bundle. Image builds, migrates, and serves /health 200. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 19:19:13 +05:30
maaz519	5bd1f646ca	docs(adr): support-service → IIOS naming migration (ADR-0001) Establishes a docs/adr convention and records the naming-migration decision: interaction-centric Iios-prefixed vocabulary as canonical, support as a namespaced specialization (KG-16), @insignia/iios-* package scope, and a greenfield-and-deprecate (strangler) migration off the legacy support-service. Includes the legacy→IIOS naming map. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 18:48:25 +05:30
maaz519	a933e2e3af	feat(iios): idempotency ledger on outbound egress (409 on key reuse) (P9) OutboundService.send is now opt-in idempotent: an explicit idempotencyKey routes the send through IdempotencyService.run (Slice 6), so reusing a key with a different target/payload returns a 409 conflict and a same-key/same-request retry replays the cached command. The inner findUnique early-return stays as a backstop so a FAILED-then-retried send never collides on the command's unique key. Keyless sends keep the old behavior (no ledger row). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 18:44:22 +05:30
maaz519	cdba0f0179	feat(iios): retention sweep dev endpoint + /metrics + smoke (P9) POST /v1/dev/retention/sweep runs the sweep on demand; /metrics gains a global retention section (snapshot counts by status). smoke-retention.mjs (0-day windows) proves an aged interaction is redacted in place and surfaces as a REDACTED snapshot on /metrics. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 12:57:51 +05:30
maaz519	035315ef73	feat(iios): retention_policy_snapshot + sweep (archive→redact, hold-aware) (P9) Adds IiosRetentionPolicySnapshot (per-interaction frozen lifecycle timestamps) + IiosInteraction.dataClass, and a RetentionService whose sweep captures snapshots then archives (status change) and, past delete_after, redacts content in place (reusing the Slice-8 tombstone) — never a hard delete. An active compliance hold blocks both; every action is audited and tenant-scopeable. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 12:50:49 +05:30
maaz519	a5cd8f0ec4	feat(iios): /v1/dsr erasure + compliance-hold REST + smoke (P9) Tenant-scoped /v1/dsr: POST /erase (self-service right-to-erasure), POST/GET /holds + POST /holds/:id/release. smoke-dsr.mjs proves PII present → hold blocks erase (409) → release → erase redacts the ticket subject in place → re-erase idempotent → cross-tenant cannot see the hold. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-03 06:27:42 +05:30