feat(observability): OpenTelemetry tracing + log/trace correlation

Adds OTel auto-instrumentation (http, express, prisma/pg, redis, ...) exporting
OTLP to Tempo, so iios-service requests appear as distributed traces in Grafana.

observability/tracing.ts is imported FIRST in main.ts — auto-instrumentation
patches modules as they are require()d, so anything loaded earlier is never
traced. Env-driven (OTEL_SERVICE_NAME / OTEL_EXPORTER_OTLP_ENDPOINT); with the
endpoint unset the SDK never starts, so dev and tests carry zero overhead.
Health/metrics probes are excluded — they would otherwise swamp Tempo and bury
real request traces.

The existing P9 trace middleware now adopts the ACTIVE OTel trace id, so an
http.request log line and its distributed trace share one id (Loki -> Tempo
pivot in Grafana), falling back to x-trace-id / traceparent / generated.

pnpm-lock.yaml regenerated — CI and the Dockerfile both use
'pnpm install --frozen-lockfile', which would hard-fail on an unchanged lock.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

.github/workflows/deploy-iios-service.yml

@@ -42,6 +42,7 @@ jobs:
 
       - name: Bump k8s-pods image tag (ArgoCD deploys)
         run: |
+          rm -rf /tmp/kp   # dind-builder is a persistent host: clear any stale checkout from a prior run
           git clone --depth 1 -b main \
             "https://mcp-bot:${{ secrets.K8S_PODS_TOKEN }}@git.lynkedup.cloud/platform-engineering/k8s-pods.git" /tmp/kp
           cd /tmp/kp

docs/DEPLOYMENT.md

@@ -46,7 +46,14 @@ Every knob is an environment variable — see [`packages/iios-service/.env.examp
 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).
+  `APP_SECRETS` (per-app HS256 JWT keys), `ADAPTER_SECRETS` (webhook HMAC keys),
+  `MEDIA_SECRET` (signs presigned media upload/download URLs).
+- **Auth (real IdP):** set `SUPABASE_URL` or `AUTH_ISSUERS` to trust real OIDC access
+  tokens, verified against the issuer's public **JWKS** (ES256) — **no secret to store**.
+  `AUTH_ISSUERS` is a JSON registry `[{url, appId, orgId?}]` routing many issuers → isolated
+  app scopes. The dev HS256 path (`APP_SECRETS`) stays for local/tests.
+- **Media storage:** `MEDIA_DIR` + `PUBLIC_URL` configure the **dev** local-disk store; for
+  prod, bind the `StoragePort` to object storage (see topology) — the API/SDK don't change.
 - **⚠️ `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.
@@ -76,7 +83,12 @@ for the full, commented list. Highlights:
   > 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**.
+  already calls them **fail-closed**. The **session** port already verifies real OIDC tokens
+  (Supabase/JWKS) when configured.
+- **Media storage** (`StoragePort`) — dev = **local disk** (`MEDIA_DIR`). ⚠️ Local disk is
+  **single-instance and non-durable**; with N>1 replicas or for persistence, bind it to shared
+  **object storage** (S3/R2/Supabase Storage). Bytes always flow **client ↔ storage directly**
+  via presigned URLs — they never transit the service — so this scales independently.
 
 ## Scaling & release strategy
 

docs/IIOS_API_AND_SDK_GUIDE.md

@@ -58,6 +58,8 @@ Use it on every call: `-H "authorization: Bearer <token>"`.
 - **Different `orgId` = different tenant.** Mint two tokens with `org_A` / `org_B` to test isolation.
 - Token TTL: 2h.
 
+**Real IdP tokens (production path).** Beyond the dev HS256 tokens, `SessionVerifier` also verifies **real OIDC access tokens** (ES256) against a trusted issuer's public **JWKS** — set `SUPABASE_URL` (single issuer) or `AUTH_ISSUERS` (a registry mapping many issuers → app scopes). The verifier routes a token by its `iss` claim, so two projects/IdPs map to two isolated `appId` scopes on one service, and app A's tokens can't reach app B. No shared secret needed. The dev token path stays for tests/local.
+
 ### Error responses (what QA will see)
 | Status | When |
 |---|---|
@@ -101,12 +103,24 @@ Grouped by domain. All paths are relative to the base URL. **Auth = Bearer token
 ### 5.3 Threads & Messaging (native chat)
 | Method | Path | Body / Headers | Returns |
 |---|---|---|---|
+| GET | `/v1/threads` | — | `ThreadSummary[]` — your threads (members, unread, last message, `membership`) |
+| POST | `/v1/threads` | `{membership?, creatorRole?, subject?}` | `{threadId, status, history}` (201) — `membership`/`creatorRole`/`subject` are **opaque app attributes** the kernel stores but never interprets |
+| POST | `/v1/threads/:id/participants` | `{userId, role?}` | `{threadId, participantCount}` (201) — **governed** (DM cap / group-admin via OPA) |
 | GET | `/v1/threads/:id/messages` | — | `{threadId, messages[]}` |
-| POST | `/v1/threads/:id/messages` | `{content, contentRef?}` + `idempotency-key?` header | `Message` (201) |
+| POST | `/v1/threads/:id/messages` | `{content, contentRef?, mimeType?, sizeBytes?, checksumSha256?, parentInteractionId?}` + `idempotency-key?` header | `Message` (201) |
+| GET | `/v1/threads/my-annotations` | `?type=save` | `{message, threadId, threadSubject}[]` — messages you annotated (e.g. saved) |
 
-**Realtime (Socket.IO, namespace `/message`):** connect, then emit client→server events:
+A **`Message`** carries `{id, threadId, senderId, senderName, content, attachment?, parentInteractionId?, annotations[], traceId, createdAt}`. `senderId` is the sender's stable externalId (email/username) — the reliable "is this mine?" check. `attachment` = `{contentRef, mimeType, sizeBytes, kind}` (§5.10). `annotations` = `[{type, value, users[]}]` — the reaction/pin/save aggregate.
-- `open_thread` `{threadId?|otherUserId}`, `send_message` `{threadId, content, idempotencyKey?}`, `read` `{threadId}`, `delivered` `{threadId}`.
+
-- Server emits to the thread room: `message` (a Message) and `receipt` `{interactionId, actorId, kind: READ|DELIVERED}`.
+**Realtime (Socket.IO, namespace `/message`).** Token verified on connect (`auth.token`), then emit client→server:
+- `open_thread` `{threadId?, membership?, creatorRole?, subject?}` — opens, or **creates** when no id; a **governed join** for membership threads. Returns `{threadId, status, history}` or `{error}` (acked, never a throw — clients don't hang).
+- `send_message` `{threadId, content, contentRef?, mimeType?, sizeBytes?, parentInteractionId?, mentions?}` — `mentions` is an **opaque userId notify-list** (the app parses `@`; the kernel never does).
+- `add_participant` `{threadId, userId, role?}` · `read` `{threadId, interactionId}` · `delivered` `{…}` · `typing` `{threadId}`.
+- `annotate` `{threadId, interactionId, type, value}` — toggle a **generic annotation** (chat app uses `type: reaction|pin|save`; `value` = emoji, or empty).
+
+Server → thread room: `message` (a Message), `receipt` `{interactionId, actorId, kind: READ|DELIVERED}`, `typing` `{threadId, userId}`, `annotation` `{interactionId, type, value, op: add|remove, users[], userId}`.
+
+**Governance & primitives (all fail-closed via OPA).** Membership (`iios.thread.participant.add`, `iios.thread.join`), annotating (`iios.interaction.annotate` — participant-only), and send all gate on policy. **Reactions/pins/saves are one generic primitive** — an *interaction annotation* the kernel stores + aggregates as opaque `(type, value)` but never interprets (the same primitive backs pins/saves/flags/tags). **Mentions → Inbox:** `mentions[]` flows into the message event; the inbox projector fans out a `MENTION` inbox item to each mentioned participant (never the sender); reading the thread resolves it.
 
 ### 5.4 Inbox (work queue)
 | Method | Path | Query / Body | Returns |
@@ -114,6 +128,8 @@ Grouped by domain. All paths are relative to the base URL. **Auth = Bearer token
 | GET | `/v1/inbox/items` | `?state=OPEN|SNOOZED|DONE|ARCHIVED|CANCELLED|STALE` | `InboxItem[]` |
 | PATCH | `/v1/inbox/items/:id` | `{state, reason?}` | `InboxItem` |
 
+Item **kinds** include `NEEDS_REPLY` (unreplied thread activity, one per owner+thread) and `MENTION` (someone @-mentioned you; `priority: HIGH`, one per source message). Reading the thread resolves both to `DONE`.
+
 ### 5.5 Support (tickets, escalation, callbacks, queues, agents)
 | Method | Path | Body / Query | Returns |
 |---|---|---|---|
@@ -179,6 +195,52 @@ Grouped by domain. All paths are relative to the base URL. **Auth = Bearer token
 
 `ScheduleMeetingDto`: `{meetingType, title, startAt, endAt?, timezone?, attendees?:[{userId, displayName?, role?, visibility?}], requestId?}`. `meetingType ∈ {ZOOM, PHONE, IN_PERSON, CALLBACK, INTERNAL}`; consent `status ∈ {UNKNOWN, GRANTED, DENIED, REVOKED}`. **Transcript/summary is `BLOCKED` until every attendee has `GRANTED` consent.**
 
+### 5.10 Media (attachments)
+
+The DB stores a **reference**; the bytes live behind a **storage port** (dev = local disk `MEDIA_DIR`; prod = swap to S3/R2/Supabase). Bytes go **client ↔ storage directly** via short-lived signed URLs — they never pass through the kernel.
+
+| Method | Path | Auth | Body | Returns |
+|---|---|---|---|---|
+| POST | `/v1/media/presign-upload` | Bearer | `{mime, sizeBytes}` | `{objectKey, uploadUrl}` — **OPA-gated** (size/type) |
+| PUT | `/v1/media/upload/:token` | token in URL | raw bytes | `{objectKey, sizeBytes, checksumSha256}` |
+| POST | `/v1/media/presign-download` | Bearer | `{contentRef, mime?}` | `{url}` — **tenant-fenced**, signed 1h |
+| GET | `/v1/media/blob/:token` | token in URL | — | streams the bytes with their `Content-Type` |
+
+**Attaching to a message:** `send`/`send_message` accept `{contentRef, mimeType, sizeBytes, checksumSha256?}`. The stored `Message` then carries `attachment: { contentRef, mimeType, sizeBytes, kind }` where `kind ∈ {image, video, audio, file}` (a friendly view of the generic `MEDIA_REF`/`VOICE_REF`/`FILE_REF` part the kernel writes).
+
+**Governance (fail-closed, built in):**
+- **Upload policy** `iios.media.upload` — **≤ 25 MB** and an allowlist (`image/*`, `video/*`, `audio/*`, `application/pdf`, common Office/text/zip). A violation → **403** *before* any bytes are sent.
+- **Signed tokens** (HS256, `MEDIA_SECRET`) — upload URL lives **5 min**, download URL **1 h**; a tampered/expired token → 403.
+- **Tenant fence** — object keys are prefixed with the caller's `scopeId`; a download for an object outside your scope → 403.
+
+**The flow (with governance):**
+
+```mermaid
+sequenceDiagram
+  autonumber
+  participant C as Client (SDK uploadMedia)
+  participant API as IIOS Media API
+  participant OPA as OPA policy
+  participant ST as StoragePort (disk → S3)
+  participant K as Message kernel
+
+  C->>API: POST /v1/media/presign-upload {mime, sizeBytes}
+  API->>OPA: decide iios.media.upload (≤25MB? type allowed?)
+  alt denied
+    OPA-->>C: 403 (too large / type not allowed)
+  else allowed
+    API-->>C: { objectKey, uploadUrl }  (signed, 5-min)
+    C->>ST: PUT bytes → uploadUrl  (direct, not via kernel)
+    ST-->>C: { sizeBytes, checksumSha256 }
+    C->>K: send_message { contentRef, mime, size }
+    K-->>C: Message { attachment }
+    C->>API: POST /v1/media/presign-download { contentRef }
+    API->>API: tenant-fence (objectKey in my scope?)
+    API-->>C: { url }  (signed, 1-hour)
+    C->>ST: GET url → bytes (stream, inline render / download)
+  end
+```
+
 ---
 
 ## 6. SDK reference
@@ -191,7 +253,9 @@ import { RestClient } from '@insignia/iios-kernel-client';
 const client = new RestClient({ serviceUrl: 'http://localhost:3200', token });
 ```
 Methods (all return typed promises):
-- **Messaging:** `getThreadMessages(threadId)`, `sendMessage(threadId, content, {contentRef?, idempotencyKey?})`
+- **Messaging:** `listThreads()`, `createThread({membership?, creatorRole?, subject?})`, `addParticipant(threadId, userId, role?)`, `getThreadMessages(threadId)`, `sendMessage(threadId, content, {attachment?, parentInteractionId?, mentions?, idempotencyKey?})`
+- **Reactions / pins / saves (annotations):** `MessageSocket.react(threadId, interactionId, emoji)` · `pin(...)` · `save(...)` (generic `annotate` under the hood); `listMyAnnotated('save')` for a cross-thread saved list. Subscribe to the `annotation` event for live updates.
+- **Media:** `uploadMedia(file, {onProgress?}) → {contentRef, mimeType, sizeBytes, checksumSha256, kind}` (presign → PUT-with-progress → normalized ref); `mediaUrl(contentRef) → signed view URL` (cached, short-lived). *This plumbing is identical for every app, so it lives in the SDK; the app only renders by `kind`.*
 - **Inbox:** `listInboxItems(state?)`, `patchInboxItem(id, {state, reason?})`
 - **Support:** `createTicket({subject, priority?, threadId?})`, `escalate(threadId, subject?)`, `listTickets('mine'|'assigned')`, `patchTicket(id, state)`, `requestCallback({...})`, `createQueue(name)`, `joinQueue(id)`, `joinDefaultQueue()`, `setAvailability(state)`
 - **Routing:** `createBinding(input)`, `listBindings()`, `simulateRoute({interactionId, originChannelType, originRef?})`, `listRouteDecisions(state?)`, `approveDecision(id)`, `denyDecision(id)`
@@ -269,7 +333,7 @@ Run against a live service (from `packages/iios-service`, `node scripts/<name>`)
 | `smoke-capability.mjs` | governed egress + real HTTP provider (needs `IIOS_PROVIDER_URL_EMAIL`) |
 | `smoke-tenant.mjs` | cross-tenant 403 + list isolation |
 
-Automated unit/integration suite: `pnpm test` (114 tests). Import-boundary check: `pnpm boundary`.
+Automated unit/integration suite: `pnpm test` (192 tests). Import-boundary check: `pnpm boundary`.
 
 ---
 
@@ -289,7 +353,11 @@ Automated unit/integration suite: `pnpm test` (114 tests). Import-boundary check
 |---|---|---|
 | `PORT` | `3200` | HTTP port |
 | `DATABASE_URL` | `postgresql://iios:iios@localhost:5434/iios?schema=public` | Postgres |
-| `APP_SECRETS` | `{"portal-demo":"dev-secret"}` | per-app JWT secrets (`{appId: secret}`) |
+| `APP_SECRETS` | `{"portal-demo":"dev-secret"}` | per-app HS256 JWT secrets (`{appId: secret}`) |
+| `SUPABASE_URL` / `AUTH_ISSUERS` | — | trusted OIDC issuer(s) — verify real IdP tokens (ES256) via JWKS. `AUTH_ISSUERS` is a JSON array `[{url, appId, orgId?}]` mapping issuers → app scopes; `SUPABASE_URL` is the single-issuer shorthand |
+| `MEDIA_DIR` | `<tmp>/iios-media` | local media storage dir (dev `StoragePort`) |
+| `MEDIA_SECRET` | `dev-media-secret` | signs media upload/download URLs |
+| `PUBLIC_URL` | `http://localhost:$PORT` | base used to build presigned media URLs |
 | `IIOS_DEV_TOKENS` | `0` | set `1` to enable `/v1/dev/*` |
 | `ADAPTER_SECRETS` | `{}` | per-channel HMAC secrets (default `dev-adapter-secret`) |
 | `IIOS_OUTBOUND_LIMIT` / `_WINDOW_MS` | `5` / `60000` | per-(channel,target) rate limit |
@@ -305,7 +373,9 @@ Automated unit/integration suite: `pnpm test` (114 tests). Import-boundary check
 
 - **Interaction kind:** MESSAGE, EMAIL, SYSTEM_NOTICE, INBOX_WORK, SUPPORT_CASE, MEETING_REQUEST, DIGEST, SUMMARY, NOTIFICATION
 - **Channel types:** WEBHOOK, EMAIL, WHATSAPP, PORTAL
-- **Inbox state:** OPEN, SNOOZED, DONE, ARCHIVED, CANCELLED, STALE · **Inbox kind:** NEEDS_REPLY, NEEDS_REVIEW, NEEDS_APPROVAL, SUPPORT_UPDATE, MEETING_FOLLOWUP, DIGEST, SYSTEM_ALERT, CRM_OWNER_INTEREST
+- **Inbox state:** OPEN, SNOOZED, DONE, ARCHIVED, CANCELLED, STALE · **Inbox kind:** NEEDS_REPLY, NEEDS_REVIEW, NEEDS_APPROVAL, **MENTION**, SUPPORT_UPDATE, MEETING_FOLLOWUP, DIGEST, SYSTEM_ALERT, CRM_OWNER_INTEREST
+- **Message part kind:** TEXT, HTML, MARKDOWN, MEDIA_REF, FILE_REF, VOICE_REF, LOCATION, STRUCTURED_JSON · **Attachment kind (DTO):** image, video, audio, file
+- **Interaction annotation (app-level, opaque to the kernel):** `type` = reaction | pin | save … ; `value` = emoji (reactions) or empty
 - **Ticket state:** NEW, OPEN, PENDING, RESOLVED, CLOSED · **priority:** LOW, NORMAL, HIGH, URGENT
 - **Route mode:** MANUAL, AUTOMATIC, HYBRID, SIMULATION_ONLY · **output format:** FORWARD, THREADED, DIGEST, SUMMARY, TRANSCRIPT · **decision:** ALLOW, DENY, REVIEW, SUPPRESS, SIMULATED
 - **AI job:** CLASSIFY, SUMMARIZE, EXTRACT · **artifact:** CLASSIFICATION, SUMMARY, TRANSCRIPT, DIGEST, EXTRACTION · **artifact status:** PROPOSED, ACCEPTED, REJECTED, SUPERSEDED

docs/IIOS_OVERVIEW_FOR_CEO.md

@@ -40,7 +40,7 @@ Today the external world (real WhatsApp, real AI models, real calendars, the rea
 | **P8** | **Calendar & meetings** — schedule, attendee consent, transcript, summary, action items | `iios-meeting-web` | Support/scheduling; callback→meeting | meeting-studio (5178) |
 | **P9** | *(next)* **Production hardening** — real providers, real policy/consent, scale, retention | — | Ops / platform | — |
 
-All of it runs on **one NestJS service** (`iios-service`) with **one Postgres database** (46 tables across 9 migrations), fronted by **one low-level client** (`iios-kernel-client`) that every React SDK is built on.
+All of it runs on **one NestJS service** (`iios-service`) with **one Postgres database** (53 tables across 19 migrations), fronted by **one low-level client** (`iios-kernel-client`) that every React SDK is built on.
 
 ---
 
@@ -159,7 +159,9 @@ Each product SDK is a handful of React hooks — a front-end dev wires the UI, t
 | Calendar/Zoom providers | Simulated sync | P9 |
 | Multi-tenant scale, retention, SLOs | Not yet | P9 |
 
-**Proof it works:** 95 automated tests pass; every capability has a runnable demo and an end-to-end smoke script; the layer-boundary check enforces the architecture.
+**Proof it works:** 192 automated tests pass; every capability has a runnable demo and an end-to-end smoke script; the layer-boundary check enforces the architecture.
+
+**P9 has begun (real providers).** A production chat app (`chat-web`) now runs on IIOS with **real Supabase login** (the session port verifies real OIDC tokens via JWKS — the first stubbed port turned real), plus richer chat built generically on the kernel: **emoji reactions, pinned & saved messages, @mentions → inbox, and media sharing** (images/video/audio/docs on a swappable storage port). Each was a thin, generic addition — no chat-specific logic in the kernel — which is the reuse thesis paying off.
 
 ---
 
@@ -175,4 +177,4 @@ Narrative arc for the CEO: **one engine → chat → inbox → support → chann
 
 ---
 
-*Appendix — repo facts: 11 packages, 6 demo apps, one NestJS service, 46 Postgres tables across 9 migrations (kernel → messaging → inbox → support → adapters → routing → ai → calendar). Boundary-enforced dependency law; 95 passing tests; 7 end-to-end smoke scripts.*
+*Appendix — repo facts: 11 packages, 6 demo apps, one NestJS service, 53 Postgres tables across 19 migrations (kernel → messaging → inbox → support → adapters → routing → ai → calendar → annotations/mentions → media). Boundary-enforced dependency law; 192 passing tests; 8+ end-to-end smoke scripts. First real-provider swap live: Supabase auth (JWKS-verified) + a media storage port.*

packages/iios-service/package.json

@@ -29,7 +29,11 @@
     "prisma": "^6.2.1",
     "reflect-metadata": "^0.2.2",
     "rxjs": "^7.8.2",
-    "socket.io": "^4.8.3"
+    "socket.io": "^4.8.3",
+    "web-push": "^3.6.7",
+    "@opentelemetry/api": "^1.9.1",
+    "@opentelemetry/sdk-node": "^0.220.0",
+    "@opentelemetry/auto-instrumentations-node": "^0.78.0"
   },
   "devDependencies": {
     "@insignia/iios-testkit": "workspace:*",
@@ -38,6 +42,7 @@
     "@types/express": "^5.0.6",
     "@types/jsonwebtoken": "^9.0.10",
     "@types/node": "^26.0.1",
+    "@types/web-push": "^3.6.4",
     "socket.io-client": "^4.8.3",
     "typescript": "^5.7.3"
   }

packages/iios-service/prisma/migrations/20260709133349_add_notifications/migration.sql

@@ -0,0 +1,30 @@
+-- AlterTable
+ALTER TABLE "IiosThreadParticipant" ADD COLUMN     "muted" BOOLEAN NOT NULL DEFAULT false;
+
+-- CreateTable
+CREATE TABLE "IiosNotificationSubscription" (
+    "id" TEXT NOT NULL,
+    "scopeId" TEXT NOT NULL,
+    "actorId" TEXT NOT NULL,
+    "kind" TEXT NOT NULL DEFAULT 'webpush',
+    "endpoint" TEXT NOT NULL,
+    "p256dh" TEXT NOT NULL,
+    "auth" TEXT NOT NULL,
+    "userAgent" TEXT,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "lastSeenAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "IiosNotificationSubscription_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "IiosNotificationSubscription_endpoint_key" ON "IiosNotificationSubscription"("endpoint");
+
+-- CreateIndex
+CREATE INDEX "IiosNotificationSubscription_actorId_idx" ON "IiosNotificationSubscription"("actorId");
+
+-- AddForeignKey
+ALTER TABLE "IiosNotificationSubscription" ADD CONSTRAINT "IiosNotificationSubscription_scopeId_fkey" FOREIGN KEY ("scopeId") REFERENCES "IiosScope"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "IiosNotificationSubscription" ADD CONSTRAINT "IiosNotificationSubscription_actorId_fkey" FOREIGN KEY ("actorId") REFERENCES "IiosActorRef"("id") ON DELETE RESTRICT ON UPDATE CASCADE;

packages/iios-service/prisma/schema.prisma

@@ -309,6 +309,7 @@ model IiosScope {
   supportQueues IiosSupportQueue[]
   tickets       IiosTicket[]
   callbacks     IiosCallbackRequest[]
+  notificationSubscriptions IiosNotificationSubscription[]
 
   @@index([orgId, appId, tenantId])
 }
@@ -358,6 +359,7 @@ model IiosActorRef {
   ticketsRequested   IiosTicket[]            @relation("TicketRequester")
   ticketsAssigned    IiosTicket[]            @relation("TicketAssignee")
   callbacksRequested IiosCallbackRequest[]
+  notificationSubscriptions IiosNotificationSubscription[]
 }
 
 /// A configured channel surface (PORTAL in P1).
@@ -412,12 +414,32 @@ model IiosThreadParticipant {
   actorId         String
   actor           IiosActorRef @relation(fields: [actorId], references: [id])
   participantRole String       @default("MEMBER")
+  muted           Boolean      @default(false)
   joinedAt        DateTime     @default(now())
   leftAt          DateTime?
 
   @@id([threadId, actorId])
 }
 
+/// A device/browser push subscription for an actor (Web Push endpoint + keys).
+/// The notification engine delivers to these when the actor is absent.
+model IiosNotificationSubscription {
+  id         String       @id @default(cuid())
+  scopeId    String
+  scope      IiosScope    @relation(fields: [scopeId], references: [id], onDelete: Cascade)
+  actorId    String
+  actor      IiosActorRef @relation(fields: [actorId], references: [id])
+  kind       String       @default("webpush")
+  endpoint   String       @unique
+  p256dh     String
+  auth       String
+  userAgent  String?
+  createdAt  DateTime     @default(now())
+  lastSeenAt DateTime     @default(now())
+
+  @@index([actorId])
+}
+
 /// The semantic envelope. Idempotent per (scope, idempotencyKey).
 model IiosInteraction {
   id                  String              @id @default(cuid())

packages/iios-service/src/app.module.ts

@@ -10,6 +10,7 @@ import { ThreadsModule } from './threads/threads.module';
 import { MessageModule } from './messaging/message.module';
 import { InboxModule } from './inbox/inbox.module';
 import { MediaModule } from './media/media.module';
+import { NotificationModule } from './notifications/notification.module';
 import { SupportModule } from './support/support.module';
 import { AdaptersModule } from './adapters/adapters.module';
 import { RoutingModule } from './routing/routing.module';
@@ -35,6 +36,7 @@ import { DevController } from './dev/dev.controller';
     MessageModule,
     InboxModule,
     MediaModule,
+    NotificationModule,
     SupportModule,
     AdaptersModule,
     RoutingModule,

packages/iios-service/src/main.ts

@@ -1,3 +1,4 @@
+import './observability/tracing'; // MUST be first — auto-instrumentation patches modules on require
 import 'dotenv/config';
 import 'reflect-metadata';
 import { NestFactory } from '@nestjs/core';

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

@@ -4,6 +4,7 @@ import {
   ConnectedSocket,
   MessageBody,
   OnGatewayConnection,
+  OnGatewayDisconnect,
   OnGatewayInit,
   SubscribeMessage,
   WebSocketGateway,
@@ -15,6 +16,7 @@ import { logJson } from '../observability/logger';
 import { MessageService, type MessagePrincipal } from './message.service';
 import { SessionVerifier } from '../platform/session.verifier';
 import { OutboxBus } from '../outbox/outbox.bus';
+import { PresenceService } from '../notifications/presence.service';
 
 interface SocketState {
   principal: MessagePrincipal;
@@ -27,7 +29,7 @@ interface SocketState {
  * propagates messages persisted elsewhere (REST/ingest) to connected clients.
  */
 @WebSocketGateway({ namespace: '/message', cors: { origin: true } })
-export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
+export class MessageGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
   private readonly logger = new Logger(MessageGateway.name);
   private readonly emittedLocally = new Set<string>();
 
@@ -37,6 +39,7 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
     private readonly messages: MessageService,
     private readonly session: SessionVerifier,
     private readonly bus: OutboxBus,
+    private readonly presence: PresenceService,
   ) {}
 
   afterInit(): void {
@@ -65,6 +68,18 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
     }
   }
 
+  handleDisconnect(client: Socket): void {
+    this.presence.clearSocket(client.id);
+  }
+
+  /** The app reports which thread is in the foreground (or null when blurred) — presence. */
+  @SubscribeMessage('focus_thread')
+  focusThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string | null }): void {
+    const state = client.data as SocketState | undefined;
+    if (!state?.principal) return;
+    this.presence.setFocus(client.id, state.principal.userId, body?.threadId ?? null);
+  }
+
   @SubscribeMessage('open_thread')
   async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string; membership?: string; creatorRole?: string; subject?: string }) {
     const { principal } = client.data as SocketState;

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

@@ -2,10 +2,11 @@ import { Module } from '@nestjs/common';
 import { MessageService } from './message.service';
 import { MessageGateway } from './message.gateway';
 import { OutboxModule } from '../outbox/outbox.module';
+import { PresenceService } from '../notifications/presence.service';
 
 @Module({
   imports: [OutboxModule],
-  providers: [MessageService, MessageGateway],
+  providers: [MessageService, MessageGateway, PresenceService],
-  exports: [MessageService],
+  exports: [MessageService, PresenceService],
 })
 export class MessageModule {}

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

@@ -53,6 +53,7 @@ export interface ThreadSummary {
   participants: string[];
   participantCount: number;
   unread: number;
+  muted?: boolean;
   lastMessage?: string;
   lastAt?: Date;
 }
@@ -154,14 +155,27 @@ export class MessageService {
     return { threadId, participantCount: participantCount + 1 };
   }
 
+  /** Toggle the caller's per-thread notification mute flag. */
+  async muteThread(threadId: string, principal: MessagePrincipal, muted: boolean): Promise<{ threadId: string; muted: boolean }> {
+    const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
+    if (!thread) throw new NotFoundException('thread not found');
+    const actor = await this.actors.resolveActor(thread.scopeId, principal);
+    await this.prisma.iiosThreadParticipant.update({
+      where: { threadId_actorId: { threadId, actorId: actor.id } },
+      data: { muted },
+    });
+    return { threadId, muted };
+  }
+
   /** Generic "my threads": every thread the caller participates in, with last message + unread. */
   async listThreads(principal: MessagePrincipal): Promise<ThreadSummary[]> {
     const scope = await this.actors.findScope(principal);
     if (!scope) return [];
     const actor = await this.actors.resolveActor(scope.id, principal);
-    const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true } });
+    const memberships = await this.prisma.iiosThreadParticipant.findMany({ where: { actorId: actor.id }, select: { threadId: true, muted: true } });
     const threadIds = memberships.map((m) => m.threadId);
     if (threadIds.length === 0) return [];
+    const mutedBy = new Map(memberships.map((m) => [m.threadId, m.muted]));
 
     const [threads, unreads, allParts] = await Promise.all([
       this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } } }),
@@ -193,6 +207,7 @@ export class MessageService {
           participants: members,
           participantCount: members.length,
           unread: unreadBy.get(t.id) ?? 0,
+          muted: mutedBy.get(t.id) ?? false,
           lastMessage: last?.parts[0]?.bodyText ?? undefined,
           lastAt: last?.occurredAt,
         };

packages/iios-service/src/notifications/notification.controller.ts

@@ -0,0 +1,46 @@
+import { BadRequestException, Body, Controller, Delete, Get, Headers, Post } from '@nestjs/common';
+import { PrismaService } from '../prisma/prisma.service';
+import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver';
+import { SessionVerifier } from '../platform/session.verifier';
+import { SubscribeDto, UnsubscribeDto } from './notification.dto';
+
+@Controller('v1/notifications')
+export class NotificationController {
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly actors: ActorResolver,
+    private readonly session: SessionVerifier,
+  ) {}
+
+  /** The public VAPID key the client needs to create a push subscription. */
+  @Get('vapid-public-key')
+  vapidKey(): { key: string } {
+    return { key: process.env.VAPID_PUBLIC_KEY ?? '' };
+  }
+
+  /** Store (or refresh) a push subscription for the caller. */
+  @Post('subscribe')
+  async subscribe(@Body() body: SubscribeDto, @Headers('authorization') auth?: string): Promise<{ ok: true }> {
+    const principal = this.principal(auth);
+    const scope = await this.actors.resolveScope(principal);
+    const actor = await this.actors.resolveActor(scope.id, principal);
+    await this.prisma.iiosNotificationSubscription.upsert({
+      where: { endpoint: body.endpoint },
+      create: { scopeId: scope.id, actorId: actor.id, kind: body.kind ?? 'webpush', endpoint: body.endpoint, p256dh: body.keys.p256dh, auth: body.keys.auth, userAgent: body.userAgent },
+      update: { actorId: actor.id, p256dh: body.keys.p256dh, auth: body.keys.auth, lastSeenAt: new Date() },
+    });
+    return { ok: true };
+  }
+
+  @Delete('subscribe')
+  async unsubscribe(@Body() body: UnsubscribeDto): Promise<{ ok: true }> {
+    await this.prisma.iiosNotificationSubscription.deleteMany({ where: { endpoint: body.endpoint } });
+    return { ok: true };
+  }
+
+  private principal(auth?: string): MessagePrincipal {
+    const token = (auth ?? '').replace(/^Bearer\s+/i, '');
+    if (!token) throw new BadRequestException('Authorization bearer token is required');
+    return this.session.verify(token);
+  }
+}

packages/iios-service/src/notifications/notification.dto.ts

@@ -0,0 +1,12 @@
+import { IsIn, IsObject, IsOptional, IsString } from 'class-validator';
+
+export class SubscribeDto {
+  @IsOptional() @IsIn(['webpush']) kind?: string;
+  @IsString() endpoint!: string;
+  @IsObject() keys!: { p256dh: string; auth: string };
+  @IsOptional() @IsString() userAgent?: string;
+}
+
+export class UnsubscribeDto {
+  @IsString() endpoint!: string;
+}

packages/iios-service/src/notifications/notification.module.ts

@@ -0,0 +1,26 @@
+import { Module } from '@nestjs/common';
+import { OutboxModule } from '../outbox/outbox.module';
+import { MessageModule } from '../messaging/message.module';
+import { NotificationController } from './notification.controller';
+import { NotificationProjector } from './notification.projector';
+import { WebPushDelivery } from './web-push.delivery';
+import { NOTIFICATION_PORT } from './notification.port';
+
+@Module({
+  imports: [OutboxModule, MessageModule], // OutboxModule → bus/dlq; MessageModule → PresenceService. Prisma/Projection/Identity are @Global.
+  controllers: [NotificationController],
+  providers: [
+    NotificationProjector,
+    {
+      // Dev binds Web Push (VAPID); prod can swap this to an email/FCM adapter.
+      provide: NOTIFICATION_PORT,
+      useFactory: () =>
+        new WebPushDelivery(
+          process.env.VAPID_PUBLIC_KEY
+            ? { publicKey: process.env.VAPID_PUBLIC_KEY, privateKey: process.env.VAPID_PRIVATE_KEY ?? '', subject: process.env.VAPID_SUBJECT ?? 'mailto:dev@insignia' }
+            : undefined,
+        ),
+    },
+  ],
+})
+export class NotificationModule {}

packages/iios-service/src/notifications/notification.port.ts

@@ -0,0 +1,18 @@
+/** The delivery seam. Dev = Web Push (VAPID); prod can swap to email/FCM/APNs. */
+export interface PushSub {
+  endpoint: string;
+  p256dh: string;
+  auth: string;
+}
+
+export interface NotificationPayload {
+  title: string;
+  body: string;
+  data: { threadId: string; interactionId?: string };
+}
+
+export interface NotificationPort {
+  deliver(sub: PushSub, payload: NotificationPayload): Promise<'sent' | 'gone' | 'failed'>;
+}
+
+export const NOTIFICATION_PORT = Symbol('NOTIFICATION_PORT');

packages/iios-service/src/notifications/notification.projector.spec.ts

@@ -0,0 +1,131 @@
+import { describe, it, expect, beforeAll, afterAll, beforeEach, vi } from 'vitest';
+import { PrismaClient } from '@prisma/client';
+import { IIOS_EVENTS, type CloudEvent } from '@insignia/iios-contracts';
+import { makeFakePorts } from '@insignia/iios-testkit';
+import { resetDb } from '../test-utils/reset-db';
+import { MessageService, type MessagePrincipal } from '../messaging/message.service';
+import { ActorResolver } from '../identity/actor.resolver';
+import { OutboxBus } from '../outbox/outbox.bus';
+import { DlqService } from '../outbox/dlq.service';
+import { ProjectionCursorService } from '../projection/projection-cursor.service';
+import { PresenceService } from './presence.service';
+import { NotificationProjector } from './notification.projector';
+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 ms = () => new MessageService(asService, makeFakePorts(), actors);
+const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' };
+
+async function actorIdFor(userId: string): Promise<string> {
+  const handle = await prisma.iiosSourceHandle.findFirstOrThrow({ where: { externalId: userId } });
+  return (await prisma.iiosActorRef.findFirstOrThrow({ where: { sourceHandleId: handle.id } })).id;
+}
+async function seedSub(userId: string): Promise<string> {
+  const actorId = await actorIdFor(userId);
+  const scope = await prisma.iiosScope.findFirstOrThrow();
+  await prisma.iiosNotificationSubscription.create({ data: { scopeId: scope.id, actorId, kind: 'webpush', endpoint: `https://push/${userId}`, p256dh: 'k', auth: 'a' } });
+  return actorId;
+}
+async function sentEvents(): Promise<CloudEvent[]> {
+  const rows = await prisma.iiosOutboxEvent.findMany({ where: { eventType: IIOS_EVENTS.messageSent }, orderBy: { createdAt: 'asc' } });
+  return rows.map((r) => r.cloudEvent as unknown as CloudEvent);
+}
+function makeProjector(presence = new PresenceService(), deliverResult: 'sent' | 'gone' = 'sent') {
+  const deliver = vi.fn().mockResolvedValue(deliverResult);
+  const proj = new NotificationProjector(asService, new OutboxBus(), new DlqService(asService), new ProjectionCursorService(asService), presence, { deliver } as never);
+  return { proj, deliver, presence };
+}
+
+beforeAll(async () => { await prisma.$connect(); });
+afterAll(async () => { await prisma.$disconnect(); });
+beforeEach(async () => { await resetDb(prisma); });
+
+describe('NotificationProjector', () => {
+  it('DM: notifies the recipient (absent), never the sender', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hi bob' }, 'k1');
+    const { proj, deliver } = makeProjector();
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).toHaveBeenCalledOnce();
+    expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob');
+  });
+
+  it('group without a mention: does NOT notify', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hello all' }, 'k1');
+    const { proj, deliver } = makeProjector();
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).not.toHaveBeenCalled();
+  });
+
+  it('group WITH a mention: notifies the mentioned member', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hey @bob' }, 'k1', undefined, undefined, ['bob']);
+    const { proj, deliver } = makeProjector();
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).toHaveBeenCalledOnce();
+  });
+
+  it('group reply-to-you: notifies the parent author even without a mention', async () => {
+    const m = ms();
+    const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' };
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    const parent = await m.send(threadId, bob, { content: 'question?' }, 'k1'); // bob authors the parent
+    await m.send(threadId, alice, { content: 'answer' }, 'k2', undefined, parent.id); // alice replies to bob (no mention)
+    const { proj, deliver } = makeProjector();
+    const evs = await sentEvents();
+    await proj.onMessageSent(evs[evs.length - 1]!); // project the reply
+    expect(deliver).toHaveBeenCalledOnce();
+    expect(deliver.mock.calls[0][0].endpoint).toContain('push/bob');
+  });
+
+  it('presence gate: a recipient viewing the thread is NOT notified', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hi' }, 'k1');
+    const presence = new PresenceService();
+    presence.setFocus('sockB', 'bob', threadId);
+    const { proj, deliver } = makeProjector(presence);
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).not.toHaveBeenCalled();
+  });
+
+  it('mute gate: a muted recipient is NOT notified', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    const bobActorId = await seedSub('bob');
+    await prisma.iiosThreadParticipant.update({ where: { threadId_actorId: { threadId, actorId: bobActorId } }, data: { muted: true } });
+    await m.send(threadId, alice, { content: 'hi' }, 'k1');
+    const { proj, deliver } = makeProjector();
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(deliver).not.toHaveBeenCalled();
+  });
+
+  it('prunes a subscription that returns "gone"', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'dm' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await seedSub('bob');
+    await m.send(threadId, alice, { content: 'hi' }, 'k1');
+    const { proj } = makeProjector(new PresenceService(), 'gone');
+    await proj.onMessageSent((await sentEvents())[0]!);
+    expect(await prisma.iiosNotificationSubscription.count()).toBe(0);
+  });
+});

packages/iios-service/src/notifications/notification.projector.ts

@@ -0,0 +1,114 @@
+import { Inject, Injectable, OnModuleInit } from '@nestjs/common';
+import { CloudEvent, IIOS_EVENTS } from '@insignia/iios-contracts';
+import { PrismaService } from '../prisma/prisma.service';
+import { OutboxBus } from '../outbox/outbox.bus';
+import { DlqService } from '../outbox/dlq.service';
+import { ProjectionCursorService } from '../projection/projection-cursor.service';
+import { PresenceService } from './presence.service';
+import { NOTIFICATION_PORT, type NotificationPort } from './notification.port';
+
+interface MsgData {
+  interactionId: string;
+  threadId: string;
+  senderActorId: string;
+  mentions?: string[];
+}
+
+/**
+ * Turns `message.sent` into push notifications for absent recipients. Three gates:
+ *   1. policy   — DM always; group only if @mentioned or a reply to that recipient
+ *   2. presence — skip if the recipient is currently focused on that thread
+ *   3. mute     — skip if the recipient muted the thread
+ * Then dispatches to each of the recipient's subscriptions; a 'gone' result prunes it.
+ * Idempotent per event id (same pattern as InboxProjector). Generic: DM-vs-group is read
+ * from the opaque `membership` thread attribute here in the notification *policy*, not the kernel.
+ */
+@Injectable()
+export class NotificationProjector implements OnModuleInit {
+  private readonly consumer = 'notification-projector';
+
+  constructor(
+    private readonly prisma: PrismaService,
+    private readonly bus: OutboxBus,
+    private readonly dlq: DlqService,
+    private readonly cursor: ProjectionCursorService,
+    private readonly presence: PresenceService,
+    @Inject(NOTIFICATION_PORT) private readonly port: NotificationPort,
+  ) {}
+
+  onModuleInit(): void {
+    this.dlq.registerHandler(this.consumer, (e) => this.onMessageSent(e));
+    this.bus.on(IIOS_EVENTS.messageSent, (p) =>
+      void this.onMessageSent(p as CloudEvent).catch((err) => this.dlq.onConsumerFailure(this.consumer, p as CloudEvent, err)),
+    );
+  }
+
+  async onMessageSent(event: CloudEvent): Promise<void> {
+    if (!(await this.claim(event.id))) return; // duplicate → no apply, no cursor advance
+    await this.apply(event);
+    await this.cursor.advance(this.consumer, event);
+  }
+
+  private async apply(event: CloudEvent): Promise<void> {
+    const data = event.data as MsgData;
+    const thread = await this.prisma.iiosThread.findUnique({ where: { id: data.threadId } });
+    if (!thread) return;
+    const membership = (thread.metadata as { membership?: string } | null)?.membership;
+
+    const interaction = await this.prisma.iiosInteraction.findUnique({
+      where: { id: data.interactionId },
+      include: { parts: { where: { kind: 'TEXT' }, take: 1 }, actor: { include: { sourceHandle: true } } },
+    });
+    const senderName = interaction?.actor?.sourceHandle?.externalId ?? 'Someone';
+    const body = interaction?.parts[0]?.bodyText ?? 'sent an attachment';
+    const mentions = data.mentions ?? [];
+
+    // Parent author (for reply-to-you) — one lookup.
+    let parentAuthorActorId: string | null = null;
+    if (interaction?.parentInteractionId) {
+      const parent = await this.prisma.iiosInteraction.findUnique({
+        where: { id: interaction.parentInteractionId },
+        select: { actorId: true },
+      });
+      parentAuthorActorId = parent?.actorId ?? null;
+    }
+
+    const participants = await this.prisma.iiosThreadParticipant.findMany({
+      where: { threadId: data.threadId },
+      include: { actor: { include: { sourceHandle: true } } },
+    });
+
+    for (const p of participants) {
+      if (p.actorId === data.senderActorId) continue; // never the sender
+      const userId = p.actor?.sourceHandle?.externalId;
+
+      // gate 1 — policy
+      const mentioned = userId != null && mentions.includes(userId);
+      const repliedToMe = parentAuthorActorId != null && parentAuthorActorId === p.actorId;
+      if (!(membership === 'dm' || mentioned || repliedToMe)) continue;
+
+      // gate 2 — presence
+      if (userId && this.presence.isViewing(userId, data.threadId)) continue;
+
+      // gate 3 — mute
+      if (p.muted) continue;
+
+      const subs = await this.prisma.iiosNotificationSubscription.findMany({ where: { actorId: p.actorId } });
+      for (const s of subs) {
+        const result = await this.port.deliver(
+          { endpoint: s.endpoint, p256dh: s.p256dh, auth: s.auth },
+          { title: senderName, body, data: { threadId: data.threadId, interactionId: data.interactionId } },
+        );
+        if (result === 'gone') await this.prisma.iiosNotificationSubscription.delete({ where: { id: s.id } });
+      }
+    }
+  }
+
+  private async claim(eventId: string): Promise<boolean> {
+    const res = await this.prisma.iiosProcessedEvent.createMany({
+      data: [{ consumerName: this.consumer, eventId }],
+      skipDuplicates: true,
+    });
+    return res.count > 0;
+  }
+}

packages/iios-service/src/notifications/presence.service.spec.ts

@@ -0,0 +1,24 @@
+import { describe, it, expect } from 'vitest';
+import { PresenceService } from './presence.service';
+
+describe('PresenceService', () => {
+  it('reports viewing only for the focused thread, and clears on disconnect', () => {
+    const p = new PresenceService();
+    expect(p.isViewing('alice', 'T1')).toBe(false);
+    p.setFocus('sock1', 'alice', 'T1');
+    expect(p.isViewing('alice', 'T1')).toBe(true);
+    expect(p.isViewing('alice', 'T2')).toBe(false); // joined-elsewhere ≠ viewing
+    p.setFocus('sock1', 'alice', 'T2'); // moved focus
+    expect(p.isViewing('alice', 'T1')).toBe(false);
+    expect(p.isViewing('alice', 'T2')).toBe(true);
+    p.clearSocket('sock1');
+    expect(p.isViewing('alice', 'T2')).toBe(false);
+  });
+
+  it('any of the actor’s sockets counts as viewing', () => {
+    const p = new PresenceService();
+    p.setFocus('sockA', 'bob', 'T9');
+    p.setFocus('sockB', 'bob', null); // a second tab, no focus
+    expect(p.isViewing('bob', 'T9')).toBe(true);
+  });
+});

packages/iios-service/src/notifications/presence.service.ts

@@ -0,0 +1,25 @@
+import { Injectable } from '@nestjs/common';
+
+/**
+ * In-memory focus tracker (single instance). Keyed on userId (the stable externalId the
+ * gateway has as principal.userId). NOTE: room membership ≠ viewing — the sidebar joins
+ * every thread room for live updates, so presence uses an explicit `focus_thread` signal.
+ * Prod (multi-replica): back this with Redis.
+ */
+@Injectable()
+export class PresenceService {
+  private readonly focus = new Map<string, { userId: string; threadId: string | null }>(); // socketId → focus
+
+  setFocus(socketId: string, userId: string, threadId: string | null): void {
+    this.focus.set(socketId, { userId, threadId });
+  }
+
+  clearSocket(socketId: string): void {
+    this.focus.delete(socketId);
+  }
+
+  isViewing(userId: string, threadId: string): boolean {
+    for (const f of this.focus.values()) if (f.userId === userId && f.threadId === threadId) return true;
+    return false;
+  }
+}

packages/iios-service/src/notifications/web-push.delivery.spec.ts

@@ -0,0 +1,34 @@
+import { describe, it, expect, vi } from 'vitest';
+import { WebPushDelivery } from './web-push.delivery';
+
+const sub = { endpoint: 'https://push/x', p256dh: 'k', auth: 'a' };
+const payload = { title: 'Bob', body: 'hi', data: { threadId: 'T1' } };
+const vapid = { publicKey: 'p', privateKey: 'k', subject: 'mailto:x' };
+
+describe('WebPushDelivery', () => {
+  it('returns "sent" on success', async () => {
+    const send = vi.fn().mockResolvedValue({ statusCode: 201 });
+    const d = new WebPushDelivery(vapid, send as never);
+    expect(await d.deliver(sub, payload)).toBe('sent');
+    expect(send).toHaveBeenCalledOnce();
+  });
+
+  it('returns "gone" on 404/410 so the caller prunes', async () => {
+    const d410 = new WebPushDelivery(vapid, vi.fn().mockRejectedValue({ statusCode: 410 }) as never);
+    expect(await d410.deliver(sub, payload)).toBe('gone');
+    const d404 = new WebPushDelivery(vapid, vi.fn().mockRejectedValue({ statusCode: 404 }) as never);
+    expect(await d404.deliver(sub, payload)).toBe('gone');
+  });
+
+  it('returns "failed" on other errors', async () => {
+    const d = new WebPushDelivery(vapid, vi.fn().mockRejectedValue({ statusCode: 500 }) as never);
+    expect(await d.deliver(sub, payload)).toBe('failed');
+  });
+
+  it('is disabled (returns "failed", never calls send) without VAPID keys', async () => {
+    const send = vi.fn();
+    const d = new WebPushDelivery(undefined, send as never);
+    expect(await d.deliver(sub, payload)).toBe('failed');
+    expect(send).not.toHaveBeenCalled();
+  });
+});

packages/iios-service/src/notifications/web-push.delivery.ts

@@ -0,0 +1,40 @@
+import { Injectable } from '@nestjs/common';
+import webpush from 'web-push';
+import type { NotificationPayload, NotificationPort, PushSub } from './notification.port';
+
+type SendFn = typeof webpush.sendNotification;
+export interface Vapid {
+  publicKey: string;
+  privateKey: string;
+  subject: string;
+}
+
+/** Web Push (VAPID) delivery. A dead subscription (404/410) → 'gone' so the caller prunes it. */
+@Injectable()
+export class WebPushDelivery implements NotificationPort {
+  private readonly vapid?: Vapid;
+
+  constructor(
+    vapid?: Vapid,
+    private readonly send: SendFn = webpush.sendNotification,
+  ) {
+    // Store VAPID; pass it per-send (not global setVapidDetails) so construction never
+    // validates keys — keeps the adapter unit-testable with a mocked transport.
+    this.vapid = vapid?.publicKey ? vapid : undefined;
+  }
+
+  async deliver(sub: PushSub, payload: NotificationPayload): Promise<'sent' | 'gone' | 'failed'> {
+    if (!this.vapid) return 'failed'; // push disabled (no VAPID keys)
+    try {
+      await this.send(
+        { endpoint: sub.endpoint, keys: { p256dh: sub.p256dh, auth: sub.auth } },
+        JSON.stringify(payload),
+        { vapidDetails: this.vapid },
+      );
+      return 'sent';
+    } catch (e) {
+      const code = (e as { statusCode?: number }).statusCode;
+      return code === 404 || code === 410 ? 'gone' : 'failed';
+    }
+  }
+}

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

@@ -1,4 +1,5 @@
 import type { Request, Response, NextFunction } from 'express';
+import { trace } from '@opentelemetry/api';
 import { runWithTrace, newTraceId, traceIdFromTraceparent } from './trace-context';
 import { logJson } from './logger';
 
@@ -9,8 +10,13 @@ import { logJson } from './logger';
  * the response finishes (method, path, status, ms). Wired via app.use() in main.ts.
  */
 export function traceMiddleware(req: Request, res: Response, next: NextFunction): void {
+  // Prefer the ACTIVE OpenTelemetry trace id: that is what makes this log line
+  // joinable to its distributed trace in Grafana (Loki -> Tempo). Falls back to
+  // the inbound header, then traceparent, then a generated id, so a correlation
+  // id is always present even when tracing is disabled.
+  const otelTraceId = trace.getActiveSpan()?.spanContext().traceId;
   const headerTrace = (req.headers['x-trace-id'] as string | undefined)?.trim();
-  const traceId = headerTrace || traceIdFromTraceparent(req.headers['traceparent'] as string | undefined) || newTraceId();
+  const traceId = otelTraceId || headerTrace || traceIdFromTraceparent(req.headers['traceparent'] as string | undefined) || newTraceId();
   res.setHeader('x-trace-id', traceId);
   const startedAt = Date.now();
   runWithTrace(traceId, () => {

packages/iios-service/src/observability/tracing.ts

@@ -0,0 +1,50 @@
+/**
+ * OpenTelemetry bootstrap. MUST be imported before anything else in main.ts —
+ * auto-instrumentation works by patching modules (http, express, pg, redis, …)
+ * as they are require()d, so any module loaded before this runs is never traced.
+ *
+ * Env-driven on purpose: the OTel SDK reads OTEL_SERVICE_NAME,
+ * OTEL_EXPORTER_OTLP_ENDPOINT and OTEL_EXPORTER_OTLP_PROTOCOL itself, so the
+ * collector target is a deployment concern rather than a code change. When
+ * OTEL_EXPORTER_OTLP_ENDPOINT is unset the SDK never starts, so local dev and
+ * tests run with zero tracing overhead and no exporter errors.
+ *
+ * Traces land in Tempo and are viewable in Grafana. The existing P9 trace
+ * middleware adopts the active OTel trace id, so an `http.request` log line and
+ * its distributed trace share one id (Loki -> Tempo pivot).
+ */
+import { NodeSDK } from '@opentelemetry/sdk-node';
+import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
+
+const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT;
+
+if (endpoint) {
+  const sdk = new NodeSDK({
+    instrumentations: [
+      getNodeAutoInstrumentations({
+        // Noisy and low value: every file read becomes a span.
+        '@opentelemetry/instrumentation-fs': { enabled: false },
+        // k8s probes hit /health constantly; tracing them would swamp Tempo
+        // and bury the real request traces.
+        '@opentelemetry/instrumentation-http': {
+          ignoreIncomingRequestHook: (req) => {
+            const url = req.url ?? '';
+            return ['/health', '/ready', '/healthz', '/readyz', '/metrics'].some((p) =>
+              url.startsWith(p),
+            );
+          },
+        },
+      }),
+    ],
+  });
+
+  sdk.start();
+
+  const shutdown = (): void => {
+    // Flush buffered spans before exit, else the last requests before a
+    // rollout are lost.
+    void sdk.shutdown().finally(() => process.exit(0));
+  };
+  process.on('SIGTERM', shutdown);
+  process.on('SIGINT', shutdown);
+}

packages/iios-service/src/threads/threads.controller.ts

@@ -54,6 +54,19 @@ export class ThreadsController {
     return this.threads.getMessages(id);
   }
 
+  /** Mute / unmute notifications from this thread for the caller. */
+  @Post(':id/mute')
+  @HttpCode(200)
+  async mute(@Param('id') id: string, @Headers('authorization') auth?: string) {
+    return this.messages.muteThread(id, this.principal(auth), true);
+  }
+
+  @Post(':id/unmute')
+  @HttpCode(200)
+  async unmute(@Param('id') id: string, @Headers('authorization') auth?: string) {
+    return this.messages.muteThread(id, this.principal(auth), false);
+  }
+
   /** REST/polling fallback for native send (same write as the socket path). */
   @Post(':id/messages')
   @HttpCode(201)