feat(iios): expose senderId (stable externalId) on MessageDto

The app needs a reliable "is this message mine?" signal. senderActorId worked only
with a lazily-learned actorId, which Supabase token-refresh wipes. senderId is the
sender's externalId (email/username) — always present and comparable to the session
user — so message alignment is deterministic. Verified: senderId round-trips = email.

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

packages/iios-service/prisma/migrations/20260706182400_add_interaction_annotations/migration.sql

@@ -0,0 +1,27 @@
+-- CreateTable
+CREATE TABLE "IiosInteractionAnnotation" (
+    "id" TEXT NOT NULL,
+    "scopeId" TEXT NOT NULL,
+    "targetInteractionId" TEXT NOT NULL,
+    "actorId" TEXT NOT NULL,
+    "annotationType" TEXT NOT NULL,
+    "value" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT "IiosInteractionAnnotation_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE INDEX "IiosInteractionAnnotation_targetInteractionId_idx" ON "IiosInteractionAnnotation"("targetInteractionId");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "IiosInteractionAnnotation_scopeId_targetInteractionId_actor_key" ON "IiosInteractionAnnotation"("scopeId", "targetInteractionId", "actorId", "annotationType", "value");
+
+-- AddForeignKey
+ALTER TABLE "IiosInteractionAnnotation" ADD CONSTRAINT "IiosInteractionAnnotation_scopeId_fkey" FOREIGN KEY ("scopeId") REFERENCES "IiosScope"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "IiosInteractionAnnotation" ADD CONSTRAINT "IiosInteractionAnnotation_targetInteractionId_fkey" FOREIGN KEY ("targetInteractionId") REFERENCES "IiosInteraction"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "IiosInteractionAnnotation" ADD CONSTRAINT "IiosInteractionAnnotation_actorId_fkey" FOREIGN KEY ("actorId") REFERENCES "IiosActorRef"("id") ON DELETE RESTRICT ON UPDATE CASCADE;

packages/iios-service/prisma/migrations/20260706194823_add_mention_inbox_kind/migration.sql

@@ -0,0 +1,2 @@
+-- AlterEnum
+ALTER TYPE "IiosInboxItemKind" ADD VALUE 'MENTION';

packages/iios-service/prisma/schema.prisma

@@ -79,6 +79,7 @@ enum IiosInboxItemKind {
   NEEDS_REPLY
   NEEDS_REVIEW
   NEEDS_APPROVAL
+  MENTION
   SUPPORT_UPDATE
   MEETING_FOLLOWUP
   DIGEST
@@ -303,6 +304,7 @@ model IiosScope {
   channels      IiosChannel[]
   threads       IiosThread[]
   interactions  IiosInteraction[]
+  annotations   IiosInteractionAnnotation[]
   inboxItems    IiosInboxItem[]
   supportQueues IiosSupportQueue[]
   tickets       IiosTicket[]
@@ -348,6 +350,7 @@ model IiosActorRef {
   threadsCreated     IiosThread[]            @relation("ThreadCreatedBy")
   participations     IiosThreadParticipant[]
   interactions       IiosInteraction[]
+  annotations        IiosInteractionAnnotation[]
   receipts           IiosMessageReceipt[]
   unreadCounters     IiosUnreadCounter[]
   inboxItemsOwned    IiosInboxItem[]
@@ -444,6 +447,7 @@ model IiosInteraction {
 
   parts             IiosMessagePart[]
   receipts          IiosMessageReceipt[]
+  annotations       IiosInteractionAnnotation[]
   inboxItems        IiosInboxItem[]
   ticketsCreatedFrom IiosTicket[]
 
@@ -451,6 +455,26 @@ model IiosInteraction {
   @@index([threadId, occurredAt])
 }
 
+/// A generic annotation an actor attaches to an interaction. Both `annotationType`
+/// and `value` are OPAQUE, app-supplied strings the kernel stores and aggregates
+/// but never interprets — e.g. the chat app writes type "reaction" / value "👍".
+/// The same primitive backs pins, saves, flags, tags later. No chat vocabulary here.
+model IiosInteractionAnnotation {
+  id                  String          @id @default(cuid())
+  scopeId             String
+  scope               IiosScope       @relation(fields: [scopeId], references: [id], onDelete: Cascade)
+  targetInteractionId String
+  target              IiosInteraction @relation(fields: [targetInteractionId], references: [id], onDelete: Cascade)
+  actorId             String
+  actor               IiosActorRef    @relation(fields: [actorId], references: [id])
+  annotationType      String
+  value               String
+  createdAt           DateTime        @default(now())
+
+  @@unique([scopeId, targetInteractionId, actorId, annotationType, value])
+  @@index([targetInteractionId])
+}
+
 model IiosMessagePart {
   id             String              @id @default(cuid())
   interactionId  String

packages/iios-service/scripts/smoke-membership.mjs

@@ -0,0 +1,61 @@
+// v1.1 smoke: dev IdP login, policy-enforced DM cap / group membership, listThreads,
+// and threaded replies — all over REST. Requires the service with IIOS_DEV_TOKENS=1.
+import 'dotenv/config';
+
+const SERVICE = process.env.SMOKE_URL ?? 'http://localhost:3200';
+const assert = (c, m) => { if (!c) { console.error('✗', m); process.exit(1); } console.log('✓', m); };
+
+async function login(username, password) {
+  const r = await fetch(`${SERVICE}/v1/dev/login`, {
+    method: 'POST', headers: { 'content-type': 'application/json' },
+    body: JSON.stringify({ username, password }),
+  });
+  return { status: r.status, body: r.ok ? await r.json() : null };
+}
+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();
+}
+
+// 1) dev IdP: correct password issues a token; wrong password is rejected.
+const ok = await login('alice', 'alice');
+assert(ok.body?.token, 'dev IdP: alice logs in with password');
+const bad = await login('alice', 'nope');
+assert(bad.status === 401, 'dev IdP: wrong password → 401');
+const alice = ok.body.token;
+
+// 2) DM is capped at two (policy).
+const dm = await json(alice, '/v1/threads', 'POST', { membership: 'dm' });
+assert(dm.threadId, `alice created a DM (${dm.threadId})`);
+const add2 = await json(alice, `/v1/threads/${dm.threadId}/participants`, 'POST', { userId: 'bob' });
+assert(add2.participantCount === 2, 'added bob → 2 participants');
+const add3 = await call(alice, `/v1/threads/${dm.threadId}/participants`, 'POST', { userId: 'carol' });
+assert(add3.status === 403, `adding a 3rd to a DM → 403 policy denied (${add3.status})`);
+
+// 3) group: creator (ADMIN) can add several.
+const grp = await json(alice, '/v1/threads', 'POST', { membership: 'group', creatorRole: 'ADMIN' });
+await json(alice, `/v1/threads/${grp.threadId}/participants`, 'POST', { userId: 'bob' });
+const g3 = await json(alice, `/v1/threads/${grp.threadId}/participants`, 'POST', { userId: 'carol' });
+assert(g3.participantCount === 3, 'group admin added bob + carol → 3 participants');
+
+// 4) threaded reply round-trips.
+const first = await json(alice, `/v1/threads/${grp.threadId}/messages`, 'POST', { content: 'question?' });
+const reply = await json(alice, `/v1/threads/${grp.threadId}/messages`, 'POST', { content: 'answer', parentInteractionId: first.id });
+assert(reply.parentInteractionId === first.id, 'a reply carries parentInteractionId');
+
+// 5) listThreads shows the caller's threads with membership + last message.
+const mine = await json(alice, '/v1/threads');
+const g = mine.find((t) => t.threadId === grp.threadId);
+assert(g && g.membership === 'group' && g.participantCount === 3, 'GET /v1/threads lists the group with membership + count');
+assert(g.lastMessage === 'answer', 'thread summary carries the last message');
+
+console.log('\nv1.1 membership + replies smoke: PASS');
+process.exit(0);

packages/iios-service/scripts/smoke-realtime-cluster.mjs

@@ -29,10 +29,12 @@ 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.
+  // Alice opens a thread on instance A; membership is governed so Alice ADDS Bob, who
+  // then opens 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 alice.emitWithAck('add_participant', { threadId, userId: 'bob-cluster' });
   await bob.emitWithAck('open_thread', { threadId });
   assert(true, 'Bob joined the same thread on instance B');
 

packages/iios-service/scripts/smoke-realtime.mjs

@@ -44,10 +44,11 @@ const alice = connect('alice');
 const bob = connect('bob');
 
 try {
-  // Alice creates a thread; Bob joins it.
+  // Alice creates a thread; membership is governed, so Alice ADDS Bob (he can't self-join).
   const opened = await alice.emitWithAck('open_thread', {});
   const threadId = opened.threadId;
   assert(!!threadId, `Alice created thread ${threadId}`);
+  await alice.emitWithAck('add_participant', { threadId, userId: 'bob' });
   await bob.emitWithAck('open_thread', { threadId });
 
   // Alice sends; Bob should receive it live.

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

@@ -1,5 +1,5 @@
 import { randomUUID } from 'node:crypto';
-import { BadRequestException, Body, Controller, ForbiddenException, Headers, Param, Post } from '@nestjs/common';
+import { BadRequestException, Body, Controller, ForbiddenException, Get, Headers, Param, Post, UnauthorizedException } from '@nestjs/common';
 import jwt from 'jsonwebtoken';
 import { IIOS_EVENTS } from '@insignia/iios-contracts';
 import { hmacSign } from '@insignia/iios-adapter-sdk';
@@ -27,20 +27,58 @@ export class DevController {
   @Post('token')
   token(@Body() body: { appId: string; userId: string; name?: string; orgId?: string }): { token: string } {
     this.assertDev();
+    return { token: this.signToken(body.appId, body.userId, body.name, body.orgId) };
+  }
+
+  /**
+   * Dev IdP: a credentialed login that mints the SAME JWT claims a real IdP / Session
+   * Broker would issue (sub/name/appId/orgId). Credentials come from DEV_USERS (JSON
+   * `{username: password}`; defaults to a few demo users). SessionVerifier is the stable
+   * verify seam — swapping in a real IdP means issuing these same claims, nothing else.
+   */
+  @Post('login')
+  login(@Body() body: { username: string; password: string; appId?: string }): { token: string; userId: string } {
+    this.assertDev();
+    const appId = body.appId ?? 'portal-demo';
+    let users: Record<string, string>;
+    try {
+      users = JSON.parse(process.env.DEV_USERS ?? '{"alice":"alice","bob":"bob","carol":"carol"}');
+    } catch {
+      users = {};
+    }
+    const expected = users[body.username];
+    if (expected === undefined || expected !== body.password) throw new UnauthorizedException('invalid username or password');
+    return { token: this.signToken(appId, body.username, body.username), userId: body.username };
+  }
+
+  /** Dev directory: the known dev usernames (from DEV_USERS). A real app validates
+   *  add-by-username against its user store / the MDM port; this stands in for it. */
+  @Get('users')
+  users(): { users: string[] } {
+    this.assertDev();
+    let map: Record<string, string>;
+    try {
+      map = JSON.parse(process.env.DEV_USERS ?? '{"alice":"alice","bob":"bob","carol":"carol"}');
+    } catch {
+      map = {};
+    }
+    return { users: Object.keys(map) };
+  }
+
+  private signToken(appId: string, userId: string, name?: string, orgId?: string): string {
     let secrets: Record<string, string>;
     try {
       secrets = JSON.parse(process.env.APP_SECRETS ?? '{}');
     } catch {
       secrets = {};
     }
-    const secret = secrets[body.appId];
-    if (!secret) throw new BadRequestException(`unknown app: ${body.appId}`);
-    const token = jwt.sign(
-      { sub: body.userId, name: body.name ?? body.userId, appId: body.appId, orgId: body.orgId ?? `org_${body.appId}` },
+    const secret = secrets[appId];
+    if (!secret) throw new BadRequestException(`unknown app: ${appId}`);
+    return jwt.sign(
+      { sub: userId, name: name ?? userId, appId, orgId: orgId ?? `org_${appId}` },
       secret,
       { algorithm: 'HS256', expiresIn: '2h' },
     );
-    return { token };
   }
 
   /** Server signs a sample provider payload and feeds it to the inbound pipeline. */

packages/iios-service/src/identity/actor.resolver.ts

@@ -68,10 +68,10 @@ export class ActorResolver {
     );
   }
 
-  async ensureParticipant(threadId: string, actorId: string): Promise<void> {
+  async ensureParticipant(threadId: string, actorId: string, participantRole = 'MEMBER'): Promise<void> {
     await this.prisma.iiosThreadParticipant.upsert({
       where: { threadId_actorId: { threadId, actorId } },
-      create: { threadId, actorId },
+      create: { threadId, actorId, participantRole },
       update: {},
     });
   }

packages/iios-service/src/inbox/inbox.projector.ts

@@ -10,6 +10,7 @@ interface MessageSentData {
   interactionId: string;
   threadId: string;
   senderActorId: string;
+  mentions?: string[]; // opaque app notify-list (userIds); the kernel never parses "@"
 }
 interface MessageReadData {
   threadId: string;
@@ -57,6 +58,60 @@ export class InboxProjector implements OnModuleInit {
       if (p.actorId === data.senderActorId) continue;
       await this.upsertNeedsReply(thread.scopeId, p.actorId, data.threadId, data.interactionId, traceId, thread.subject);
     }
+
+    // Targeted mention notifications from the app-supplied opaque notify-list.
+    const mentions = data.mentions ?? [];
+    if (mentions.length > 0) {
+      const src = await this.prisma.iiosInteraction.findUnique({
+        where: { id: data.interactionId },
+        include: { parts: { where: { kind: 'TEXT' }, take: 1 }, actor: { include: { sourceHandle: true } } },
+      });
+      const senderName = src?.actor?.sourceHandle?.externalId ?? 'Someone';
+      const text = src?.parts[0]?.bodyText ?? undefined;
+      for (const userId of mentions) {
+        await this.createMention(thread.scopeId, userId, data.threadId, data.interactionId, data.senderActorId, senderName, text, traceId);
+      }
+    }
+  }
+
+  /** One MENTION item per (owner, source message) for a mentioned participant (never the sender). */
+  private async createMention(
+    scopeId: string,
+    userId: string,
+    threadId: string,
+    sourceInteractionId: string,
+    senderActorId: string,
+    senderName: string,
+    summary: string | undefined,
+    traceId: string | undefined,
+  ): Promise<void> {
+    const handle = await this.prisma.iiosSourceHandle.findUnique({
+      where: { scopeId_kind_externalId: { scopeId, kind: 'PORTAL_USER', externalId: userId } },
+    });
+    if (!handle) return;
+    const actor = await this.prisma.iiosActorRef.findFirst({ where: { sourceHandleId: handle.id } });
+    if (!actor || actor.id === senderActorId) return; // resolvable, and never notify the sender
+    const isMember = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } });
+    if (!isMember) return; // only thread participants get mention items
+    const existing = await this.prisma.iiosInboxItem.findFirst({ where: { ownerActorId: actor.id, sourceInteractionId, kind: 'MENTION' } });
+    if (existing) return; // idempotent per source message
+
+    const item = await this.prisma.iiosInboxItem.create({
+      data: {
+        scopeId,
+        ownerActorId: actor.id,
+        kind: 'MENTION',
+        title: `${senderName} mentioned you`,
+        summary,
+        priority: 'HIGH',
+        threadId,
+        sourceInteractionId,
+        traceId,
+      },
+    });
+    await this.prisma.iiosInboxItemStateHistory.create({
+      data: { inboxItemId: item.id, toState: 'OPEN', reasonCode: 'mentioned' },
+    });
   }
 
   async onMessageRead(event: CloudEvent): Promise<void> {
@@ -67,21 +122,24 @@ export class InboxProjector implements OnModuleInit {
 
   private async applyMessageRead(event: CloudEvent): Promise<void> {
     const data = event.data as MessageReadData;
-    const item = await this.prisma.iiosInboxItem.findFirst({
+    // Reading the thread clears the reader's activity AND mention items for it.
+    const items = await this.prisma.iiosInboxItem.findMany({
       where: {
         threadId: data.threadId,
         ownerActorId: data.actorId,
-        kind: 'NEEDS_REPLY',
+        kind: { in: ['NEEDS_REPLY', 'MENTION'] },
         state: { in: ['OPEN', 'SNOOZED'] },
       },
     });
-    if (!item) return;
-    await this.prisma.$transaction([
-      this.prisma.iiosInboxItem.update({ where: { id: item.id }, data: { state: 'DONE' } }),
-      this.prisma.iiosInboxItemStateHistory.create({
-        data: { inboxItemId: item.id, fromState: item.state, toState: 'DONE', reasonCode: 'message_read' },
-      }),
-    ]);
+    if (items.length === 0) return;
+    await this.prisma.$transaction(
+      items.flatMap((item) => [
+        this.prisma.iiosInboxItem.update({ where: { id: item.id }, data: { state: 'DONE' } }),
+        this.prisma.iiosInboxItemStateHistory.create({
+          data: { inboxItemId: item.id, fromState: item.state, toState: 'DONE', reasonCode: 'message_read' },
+        }),
+      ]),
+    );
   }
 
   private async upsertNeedsReply(

packages/iios-service/src/inbox/inbox.spec.ts

@@ -88,6 +88,38 @@ describe('InboxProjector (P3 work surface)', () => {
     expect(await itemsFor('bob')).toHaveLength(1);
   });
 
+  it('message.sent with mentions → a MENTION item for the mentioned participant only', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    await m.addParticipant(threadId, alice, 'carol');
+    await m.send(threadId, alice, { content: 'hey @bob look' }, 'k1', undefined, undefined, ['bob']);
+
+    await projector().onMessageSent((await eventsOf(IIOS_EVENTS.messageSent))[0]!);
+
+    const bobMentions = (await itemsFor('bob')).filter((i) => i.kind === 'MENTION');
+    expect(bobMentions).toHaveLength(1);
+    expect(bobMentions[0]?.state).toBe('OPEN');
+    expect(bobMentions[0]?.sourceInteractionId).toBeTruthy();
+    expect((await itemsFor('carol')).filter((i) => i.kind === 'MENTION')).toHaveLength(0); // not mentioned
+    expect((await itemsFor('alice')).filter((i) => i.kind === 'MENTION')).toHaveLength(0); // never the sender
+  });
+
+  it('reading the thread resolves the reader’s MENTION item to DONE', async () => {
+    const m = ms();
+    const { threadId } = await m.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await m.addParticipant(threadId, alice, 'bob');
+    const sent = await m.send(threadId, alice, { content: '@bob ping' }, 'k1', undefined, undefined, ['bob']);
+
+    const proj = projector();
+    await proj.onMessageSent((await eventsOf(IIOS_EVENTS.messageSent))[0]!);
+    expect((await itemsFor('bob')).filter((i) => i.kind === 'MENTION')[0]?.state).toBe('OPEN');
+
+    await m.markRead(threadId, bob, sent.id);
+    await proj.onMessageRead((await eventsOf(IIOS_EVENTS.messageRead))[0]!);
+    expect((await itemsFor('bob')).filter((i) => i.kind === 'MENTION')[0]?.state).toBe('DONE');
+  });
+
   it('message.read → the recipient item goes DONE', async () => {
     const m = ms();
     const { threadId } = await m.openThread(null, alice);

packages/iios-service/src/main.ts

@@ -5,6 +5,7 @@ import { ValidationPipe } from '@nestjs/common';
 import { AppModule } from './app.module';
 import { traceMiddleware } from './observability/trace.middleware';
 import { RedisIoAdapter } from './realtime/redis-io.adapter';
+import { PolicyDeniedFilter } from './platform/policy-denied.filter';
 
 async function bootstrap(): Promise<void> {
   const app = await NestFactory.create(AppModule, { rawBody: true });
@@ -12,6 +13,7 @@ async function bootstrap(): Promise<void> {
   app.useGlobalPipes(
     new ValidationPipe({ whitelist: true, forbidNonWhitelisted: true, transform: true }),
   );
+  app.useGlobalFilters(new PolicyDeniedFilter()); // fail-closed policy denials → 403
   app.enableCors({ origin: true, credentials: true });
 
   // Multi-replica realtime: fan socket.io room emits across instances via Redis.

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

@@ -66,17 +66,32 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
   }
 
   @SubscribeMessage('open_thread')
-  async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string }) {
+  async openThread(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId?: string; membership?: string; creatorRole?: string; subject?: string }) {
     const { principal } = client.data as SocketState;
-    const result = await this.messages.openThread(body?.threadId ?? null, principal);
-    await client.join(result.threadId);
-    return result;
+    try {
+      const result = await this.messages.openThread(body?.threadId ?? null, principal, {
+        membership: body?.membership,
+        creatorRole: body?.creatorRole,
+        subject: body?.subject,
+      });
+      await client.join(result.threadId);
+      return result;
+    } catch (err) {
+      // Fail to an ACK'd error instead of throwing (which never acks → client hangs on "loading").
+      return { error: (err as Error).message ?? 'could not open the conversation' };
+    }
+  }
+
+  @SubscribeMessage('add_participant')
+  async addParticipant(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string; userId: string; role?: string }) {
+    const { principal } = client.data as SocketState;
+    return this.messages.addParticipant(body.threadId, principal, body.userId, body.role);
   }
 
   @SubscribeMessage('send_message')
   async sendMessage(
     @ConnectedSocket() client: Socket,
-    @MessageBody() body: { threadId: string; content: string; contentRef?: string },
+    @MessageBody() body: { threadId: string; content: string; contentRef?: string; parentInteractionId?: string; mentions?: string[] },
   ) {
     const { principal } = client.data as SocketState;
     const msg = await this.messages.send(
@@ -84,12 +99,35 @@ export class MessageGateway implements OnGatewayInit, OnGatewayConnection {
       principal,
       { content: body.content, contentRef: body.contentRef },
       randomUUID(),
+      undefined,
+      body.parentInteractionId,
+      body.mentions,
     );
     this.emittedLocally.add(msg.id);
     this.server.to(body.threadId).emit('message', msg);
     return msg;
   }
 
+  @SubscribeMessage('annotate')
+  async annotate(
+    @ConnectedSocket() client: Socket,
+    @MessageBody() body: { threadId: string; interactionId: string; type: string; value: string },
+  ) {
+    const { principal } = client.data as SocketState;
+    const r = await this.messages.toggleAnnotation(body.interactionId, principal, body.type, body.value);
+    // Broadcast the refreshed user list for this (interaction,type,value) so every client re-renders.
+    this.server.to(r.threadId).emit('annotation', {
+      threadId: r.threadId,
+      interactionId: r.interactionId,
+      type: r.type,
+      value: r.value,
+      op: r.op,
+      users: r.users,
+      userId: principal.userId,
+    });
+    return r;
+  }
+
   @SubscribeMessage('read')
   async read(@ConnectedSocket() client: Socket, @MessageBody() body: { threadId: string; interactionId: string }) {
     const { principal } = client.data as SocketState;

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

@@ -9,16 +9,38 @@ import { ActorResolver, type MessagePrincipal } from '../identity/actor.resolver
 
 export type { MessagePrincipal };
 
+/** A generic annotation aggregate on a message (opaque type/value + who applied it). */
+export interface AnnotationDto {
+  type: string;
+  value: string;
+  users: string[];
+}
+
 export interface MessageDto {
   id: string;
   threadId: string;
   senderActorId: string;
+  senderId: string; // the sender's stable externalId (email/username) — reliable "is this mine?"
+  senderName: string;
   content: string;
   contentRef?: string;
+  parentInteractionId?: string;
+  annotations: AnnotationDto[];
   traceId: string;
   createdAt: Date;
 }
 
+export interface ThreadSummary {
+  threadId: string;
+  subject: string | null;
+  membership?: string;
+  participants: string[];
+  participantCount: number;
+  unread: number;
+  lastMessage?: string;
+  lastAt?: Date;
+}
+
 export interface OpenThreadResult {
   threadId: string;
   status: string;
@@ -39,16 +61,29 @@ export class MessageService {
     private readonly actors: ActorResolver,
   ) {}
 
-  /** Open an existing thread, or create one when no id is given. Joins as participant. */
-  async openThread(threadId: string | null, principal: MessagePrincipal): Promise<OpenThreadResult> {
+  /**
+   * Open an existing thread, or create one when no id is given. On create, an optional
+   * generic `membership` attribute is stamped on the thread's metadata (the app's DM/group
+   * hint — the kernel never branches on it; policy does), and the creator joins as ADMIN
+   * for a group. Opening an existing thread is a GOVERNED join: only an existing member may
+   * re-open it (policy `iios.thread.join`) — new members enter via addParticipant.
+   */
+  async openThread(threadId: string | null, principal: MessagePrincipal, opts?: { membership?: string; creatorRole?: string; subject?: string }): Promise<OpenThreadResult> {
     if (!threadId) {
       await decideOrThrow(this.ports, { action: 'iios.thread.create', scope: principal });
       const scope = await this.actors.resolveScope(principal);
       const actor = await this.actors.resolveActor(scope.id, principal);
+      // `membership`/`creatorRole`/`subject` are generic, app-supplied thread attributes — the
+      // kernel stores/echoes them but never branches on their chat meaning (that lives in policy + app).
       const thread = await this.prisma.iiosThread.create({
-        data: { scopeId: scope.id, createdByActorId: actor.id },
+        data: {
+          scopeId: scope.id,
+          createdByActorId: actor.id,
+          subject: opts?.subject?.trim() || undefined,
+          metadata: opts?.membership ? ({ membership: opts.membership } as Prisma.InputJsonValue) : undefined,
+        },
       });
-      await this.actors.ensureParticipant(thread.id, actor.id);
+      await this.actors.ensureParticipant(thread.id, actor.id, opts?.creatorRole ?? 'MEMBER');
       return { threadId: thread.id, status: thread.status, history: [] };
     }
 
@@ -56,16 +91,108 @@ export class MessageService {
     if (!thread) throw new NotFoundException('thread not found');
     await decideOrThrow(this.ports, { action: 'iios.thread.read', threadId, scopeId: thread.scopeId });
     const actor = await this.actors.resolveActor(thread.scopeId, principal);
+    const alreadyMember =
+      (await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: actor.id } } })) !== null;
+    // Join is governed ONLY for threads that opted into a membership model (chat dm/group);
+    // support/inbox/generic threads (no membership attr) keep open-join, unchanged.
+    const membership = (thread.metadata as { membership?: string } | null)?.membership;
+    await decideOrThrow(this.ports, { action: 'iios.thread.join', threadId, scopeId: thread.scopeId, membership, alreadyMember });
     await this.actors.ensureParticipant(threadId, actor.id);
     return { threadId, status: thread.status, history: await this.history(threadId) };
   }
 
+  /**
+   * Governed thread membership (a member adds another user by userId). Fail-closed via
+   * policy: the DM cap / group-admin rule is enforced by OPA (dev stub now, real later);
+   * the kernel only supplies generic context and adds the participant. The target's actor
+   * is resolved-or-created so you can add someone who hasn't logged in yet.
+   */
+  async addParticipant(threadId: string, principal: MessagePrincipal, targetUserId: string, role = 'MEMBER'): Promise<{ threadId: string; participantCount: number }> {
+    const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
+    if (!thread) throw new NotFoundException('thread not found');
+    const caller = await this.actors.resolveActor(thread.scopeId, principal);
+    const callerP = await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId, actorId: caller.id } } });
+    const participantCount = await this.prisma.iiosThreadParticipant.count({ where: { threadId } });
+    const membership = (thread.metadata as { membership?: string } | null)?.membership;
+
+    await decideOrThrow(this.ports, {
+      action: 'iios.thread.participant.add',
+      threadId,
+      scopeId: thread.scopeId,
+      membership,
+      participantCount,
+      callerRole: callerP?.participantRole,
+      targetUserId,
+      role,
+    });
+
+    const scope = await this.actors.resolveScope(principal);
+    const target = await this.actors.resolveActor(scope.id, {
+      userId: targetUserId,
+      appId: principal.appId,
+      orgId: principal.orgId,
+      tenantId: principal.tenantId,
+      displayName: targetUserId,
+    });
+    await this.actors.ensureParticipant(threadId, target.id, role);
+    return { threadId, participantCount: participantCount + 1 };
+  }
+
+  /** 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 threadIds = memberships.map((m) => m.threadId);
+    if (threadIds.length === 0) return [];
+
+    const [threads, unreads, allParts] = await Promise.all([
+      this.prisma.iiosThread.findMany({ where: { id: { in: threadIds } } }),
+      this.prisma.iiosUnreadCounter.findMany({ where: { threadId: { in: threadIds }, actorId: actor.id } }),
+      this.prisma.iiosThreadParticipant.findMany({
+        where: { threadId: { in: threadIds } },
+        include: { actor: { include: { sourceHandle: true } } },
+      }),
+    ]);
+    const unreadBy = new Map(unreads.map((u) => [u.threadId, u.unreadCount]));
+    const membersBy = new Map<string, string[]>();
+    for (const p of allParts) {
+      const name = p.actor?.sourceHandle?.externalId ?? p.actor?.displayName ?? p.actorId;
+      membersBy.set(p.threadId, [...(membersBy.get(p.threadId) ?? []), name]);
+    }
+
+    const summaries = await Promise.all(
+      threads.map(async (t) => {
+        const last = await this.prisma.iiosInteraction.findFirst({
+          where: { threadId: t.id },
+          orderBy: { occurredAt: 'desc' },
+          include: { parts: { where: { kind: 'TEXT' }, take: 1 } },
+        });
+        const members = membersBy.get(t.id) ?? [];
+        return {
+          threadId: t.id,
+          subject: t.subject,
+          membership: (t.metadata as { membership?: string } | null)?.membership,
+          participants: members,
+          participantCount: members.length,
+          unread: unreadBy.get(t.id) ?? 0,
+          lastMessage: last?.parts[0]?.bodyText ?? undefined,
+          lastAt: last?.occurredAt,
+        };
+      }),
+    );
+    return summaries.sort((a, b) => (b.lastAt?.getTime() ?? 0) - (a.lastAt?.getTime() ?? 0));
+  }
+
   async send(
     threadId: string,
     principal: MessagePrincipal,
     body: { content: string; contentRef?: string },
     idempotencyKey: string,
     traceId: string = randomUUID(),
+    parentInteractionId?: string,
+    mentions?: string[],
   ): Promise<MessageDto> {
     const thread = await this.prisma.iiosThread.findUnique({ where: { id: threadId } });
     if (!thread) throw new NotFoundException('thread not found');
@@ -75,10 +202,15 @@ export class MessageService {
     const actor = await this.actors.resolveActor(thread.scopeId, principal);
     await this.actors.ensureParticipant(threadId, actor.id);
 
+    // A reply links to its parent — but only if the parent is in the same thread (else ignored).
+    const parentRef = parentInteractionId
+      ? (await this.prisma.iiosInteraction.findFirst({ where: { id: parentInteractionId, threadId }, select: { id: true } }))?.id
+      : undefined;
+
     // Idempotency: a repeat key returns the existing message (no re-increment).
     const existing = await this.prisma.iiosInteraction.findUnique({
       where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } },
-      include: { parts: { orderBy: { partIndex: 'asc' } } },
+      include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
     });
     if (existing) return this.toDto(existing, threadId);
 
@@ -93,6 +225,7 @@ export class MessageService {
             idempotencyKey,
             status: 'NORMALIZED',
             traceId,
+            parentInteractionId: parentRef,
           },
         });
 
@@ -114,7 +247,9 @@ export class MessageService {
           datacontenttype: 'application/json',
           traceparent: `00-${traceId.replace(/-/g, '')}-0000000000000000-01`,
           insignia: { scopeSnapshotId: thread.scopeId, correlationId: traceId, idempotencyKey, dataClass: 'internal' },
-          data: { interactionId: created.id, threadId, senderActorId: actor.id },
+          // `mentions` is an OPAQUE app-supplied notify-list (userIds) — the kernel never parses
+          // "@"; the inbox projector generically fans out a notification to those actors.
+          data: { interactionId: created.id, threadId, senderActorId: actor.id, mentions: mentions ?? [] },
         };
         await tx.iiosOutboxEvent.create({
           data: {
@@ -139,7 +274,7 @@ export class MessageService {
 
         return tx.iiosInteraction.findUniqueOrThrow({
           where: { id: created.id },
-          include: { parts: { orderBy: { partIndex: 'asc' } } },
+          include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
         });
       });
 
@@ -148,7 +283,7 @@ export class MessageService {
       if (err instanceof Prisma.PrismaClientKnownRequestError && err.code === 'P2002') {
         const winner = await this.prisma.iiosInteraction.findUniqueOrThrow({
           where: { scopeId_idempotencyKey: { scopeId: thread.scopeId, idempotencyKey } },
-          include: { parts: { orderBy: { partIndex: 'asc' } } },
+          include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
         });
         return this.toDto(winner, threadId);
       }
@@ -220,7 +355,7 @@ export class MessageService {
   async getMessageById(id: string, principal?: MessagePrincipal): Promise<MessageDto | null> {
     const i = await this.prisma.iiosInteraction.findUnique({
       where: { id },
-      include: { parts: { orderBy: { partIndex: 'asc' } } },
+      include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
     });
     if (!i || !i.threadId) return null;
     if (principal) await this.actors.assertOwns(principal, i.scopeId); // tenant fence (KG-02) when called on behalf of a caller
@@ -231,16 +366,140 @@ export class MessageService {
     const interactions = await this.prisma.iiosInteraction.findMany({
       where: { threadId },
       orderBy: { occurredAt: 'asc' },
-      include: { parts: { orderBy: { partIndex: 'asc' } } },
+      include: { parts: { orderBy: { partIndex: 'asc' } }, actor: { include: { sourceHandle: true } } },
     });
-    return interactions.map((i) => this.toDto(i, threadId));
+    const annotations = await this.annotationsByInteraction(interactions.map((i) => i.id));
+    return interactions.map((i) => this.toDto(i, threadId, annotations.get(i.id) ?? []));
+  }
+
+  /**
+   * Toggle a generic annotation (actor attaches/removes an opaque label on an interaction).
+   * `annotationType`/`value` are app-supplied and never interpreted here (the chat app uses
+   * type "reaction" + an emoji value). Governed: only a thread participant may annotate.
+   * Returns the refreshed user list for that (type,value) so callers can broadcast it.
+   */
+  async toggleAnnotation(
+    interactionId: string,
+    principal: MessagePrincipal,
+    annotationType: string,
+    value: string,
+  ): Promise<{ interactionId: string; threadId: string; type: string; value: string; op: 'add' | 'remove'; users: string[] }> {
+    const target = await this.prisma.iiosInteraction.findUnique({
+      where: { id: interactionId },
+      select: { id: true, threadId: true, scopeId: true },
+    });
+    if (!target || !target.threadId) throw new NotFoundException('message not found');
+    const actor = await this.actors.resolveActor(target.scopeId, principal);
+    const isMember =
+      (await this.prisma.iiosThreadParticipant.findUnique({ where: { threadId_actorId: { threadId: target.threadId, actorId: actor.id } } })) !== null;
+    await decideOrThrow(this.ports, { action: 'iios.interaction.annotate', threadId: target.threadId, scopeId: target.scopeId, isMember });
+
+    const key = {
+      scopeId_targetInteractionId_actorId_annotationType_value: {
+        scopeId: target.scopeId,
+        targetInteractionId: interactionId,
+        actorId: actor.id,
+        annotationType,
+        value,
+      },
+    };
+    const existing = await this.prisma.iiosInteractionAnnotation.findUnique({ where: key });
+    let op: 'add' | 'remove';
+    if (existing) {
+      await this.prisma.iiosInteractionAnnotation.delete({ where: { id: existing.id } });
+      op = 'remove';
+    } else {
+      await this.prisma.iiosInteractionAnnotation.create({
+        data: { scopeId: target.scopeId, targetInteractionId: interactionId, actorId: actor.id, annotationType, value },
+      });
+      op = 'add';
+    }
+
+    const users =
+      (await this.annotationsByInteraction([interactionId])).get(interactionId)?.find((a) => a.type === annotationType && a.value === value)?.users ?? [];
+    return { interactionId, threadId: target.threadId, type: annotationType, value, op, users };
+  }
+
+  /**
+   * Generic "my annotated interactions" — every message the caller annotated with a given
+   * type, newest first, with thread context. The app uses type "save" for a personal
+   * bookmarks list (and could use "pin" for a cross-thread pinned view). Generic: the kernel
+   * never interprets the type string.
+   */
+  async listMyAnnotated(
+    principal: MessagePrincipal,
+    annotationType: string,
+  ): Promise<Array<{ message: MessageDto; threadId: string; threadSubject: string | null }>> {
+    const scope = await this.actors.findScope(principal);
+    if (!scope) return [];
+    const actor = await this.actors.resolveActor(scope.id, principal);
+    const anns = await this.prisma.iiosInteractionAnnotation.findMany({
+      where: { actorId: actor.id, annotationType },
+      orderBy: { createdAt: 'desc' },
+      select: { targetInteractionId: true },
+    });
+    const ids = anns.map((a) => a.targetInteractionId);
+    if (ids.length === 0) return [];
+    const [interactions, agg] = await Promise.all([
+      this.prisma.iiosInteraction.findMany({
+        where: { id: { in: ids } },
+        include: {
+          parts: { orderBy: { partIndex: 'asc' } },
+          actor: { include: { sourceHandle: true } },
+          thread: { select: { subject: true } },
+        },
+      }),
+      this.annotationsByInteraction(ids),
+    ]);
+    const byId = new Map(interactions.map((i) => [i.id, i]));
+    return ids
+      .map((id) => byId.get(id))
+      .filter((i): i is NonNullable<typeof i> => Boolean(i && i.threadId))
+      .map((i) => ({
+        message: this.toDto(i, i.threadId as string, agg.get(i.id) ?? []),
+        threadId: i.threadId as string,
+        threadSubject: i.thread?.subject ?? null,
+      }));
   }
 
   // ─── helpers ────────────────────────────────────────────────────
 
+  /** Aggregate annotations for the given interactions into { type, value, users[] } groups. */
+  private async annotationsByInteraction(interactionIds: string[]): Promise<Map<string, AnnotationDto[]>> {
+    const map = new Map<string, AnnotationDto[]>();
+    if (interactionIds.length === 0) return map;
+    const rows = await this.prisma.iiosInteractionAnnotation.findMany({
+      where: { targetInteractionId: { in: interactionIds } },
+      include: { actor: { include: { sourceHandle: true } } },
+      orderBy: { createdAt: 'asc' },
+    });
+    for (const r of rows) {
+      const user = r.actor?.sourceHandle?.externalId ?? r.actor?.displayName ?? r.actorId;
+      const list = map.get(r.targetInteractionId) ?? [];
+      let entry = list.find((e) => e.type === r.annotationType && e.value === r.value);
+      if (!entry) {
+        entry = { type: r.annotationType, value: r.value, users: [] };
+        list.push(entry);
+      }
+      entry.users.push(user);
+      map.set(r.targetInteractionId, list);
+    }
+    for (const list of map.values()) for (const e of list) e.users.sort(); // deterministic order
+    return map;
+  }
+
   private toDto(
-    interaction: { id: string; actorId: string | null; traceId: string | null; occurredAt: Date; parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }> },
+    interaction: {
+      id: string;
+      actorId: string | null;
+      actor?: { displayName: string | null; sourceHandle: { externalId: string } | null } | null;
+      parentInteractionId?: string | null;
+      traceId: string | null;
+      occurredAt: Date;
+      parts: Array<{ kind: string; bodyText: string | null; contentRef: string | null }>;
+    },
     threadId: string,
+    annotations: AnnotationDto[] = [],
   ): MessageDto {
     const text = interaction.parts.find((p) => p.kind === 'TEXT');
     const file = interaction.parts.find((p) => p.contentRef);
@@ -248,8 +507,12 @@ export class MessageService {
       id: interaction.id,
       threadId,
       senderActorId: interaction.actorId ?? '',
+      senderId: interaction.actor?.sourceHandle?.externalId ?? '',
+      senderName: interaction.actor?.displayName ?? interaction.actor?.sourceHandle?.externalId ?? 'unknown',
       content: text?.bodyText ?? '',
       contentRef: file?.contentRef ?? undefined,
+      parentInteractionId: interaction.parentInteractionId ?? undefined,
+      annotations,
       traceId: interaction.traceId ?? '',
       createdAt: interaction.occurredAt,
     };

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

@@ -1,15 +1,20 @@
 import { describe, it, expect, beforeAll, afterAll, beforeEach } from 'vitest';
 import { PrismaClient } from '@prisma/client';
+import { PolicyDeniedError, type IiosPlatformPorts } from '@insignia/iios-contracts';
 import { resetDb } from '../test-utils/reset-db';
 import { makeFakePorts } from '@insignia/iios-testkit';
 import { MessageService, type MessagePrincipal } from './message.service';
 import { ActorResolver } from '../identity/actor.resolver';
+import { DevOpaPort } from '../platform/dev-opa.port';
 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 svc = () => new MessageService(asService, makeFakePorts(), new ActorResolver(asService));
+// A service whose OPA actually enforces the membership rules (real dev policy plane).
+const gov = () =>
+  new MessageService(asService, { ...makeFakePorts(), opa: new DevOpaPort() } as IiosPlatformPorts, new ActorResolver(asService));
 
 const alice: MessagePrincipal = { userId: 'alice', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Alice' };
 const bob: MessagePrincipal = { userId: 'bob', orgId: 'org_demo', appId: 'portal-demo', displayName: 'Bob' };
@@ -101,3 +106,124 @@ describe('MessageService (P2 native messaging)', () => {
     expect(ce.insignia.correlationId).toBe(msg.traceId);
   });
 });
+
+describe('Governed membership + replies (v1.1, policy-enforced)', () => {
+  it('a direct message is capped at two people (OPA policy)', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'dm' });
+    expect((await s.addParticipant(threadId, alice, 'bob')).participantCount).toBe(2);
+    await expect(s.addParticipant(threadId, alice, 'carol')).rejects.toBeInstanceOf(PolicyDeniedError);
+    expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(2); // unchanged
+  });
+
+  it('group: the creator is ADMIN and can add; a plain member cannot', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await s.addParticipant(threadId, alice, 'bob'); // admin adds a member
+    expect(await prisma.iiosThreadParticipant.count({ where: { threadId } })).toBe(2);
+    await expect(s.addParticipant(threadId, bob, 'carol')).rejects.toBeInstanceOf(PolicyDeniedError); // bob is MEMBER
+  });
+
+  it('self-join is governed: a non-member cannot open a thread by id; after being added, they can', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await expect(s.openThread(threadId, bob)).rejects.toBeInstanceOf(PolicyDeniedError);
+    await s.addParticipant(threadId, alice, 'bob');
+    expect((await s.openThread(threadId, bob)).threadId).toBe(threadId);
+  });
+
+  it('listThreads returns the caller’s threads with membership, count, last message + unread', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await s.addParticipant(threadId, alice, 'bob');
+    await s.send(threadId, alice, { content: 'hello team' }, 'k1');
+    const mine = await s.listThreads(alice);
+    expect(mine).toHaveLength(1);
+    expect(mine[0]).toMatchObject({ membership: 'group', participantCount: 2, lastMessage: 'hello team' });
+    expect((await s.listThreads(bob))[0]?.unread).toBe(1);
+  });
+
+  it('a reply stores + returns parentInteractionId; a cross-thread parent is ignored', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'dm' });
+    const first = await s.send(threadId, alice, { content: 'question?' }, 'k1');
+    const reply = await s.send(threadId, alice, { content: 'answer' }, 'k2', undefined, first.id);
+    expect(reply.parentInteractionId).toBe(first.id);
+
+    const { threadId: other } = await s.openThread(null, alice, { membership: 'dm' });
+    const cross = await s.send(other, alice, { content: 'x' }, 'k3', undefined, first.id);
+    expect(cross.parentInteractionId).toBeUndefined(); // parent not in this thread
+  });
+});
+
+describe('Interaction annotations (generic reactions primitive)', () => {
+  it('toggleAnnotation adds then removes (toggle) and aggregates users into history', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    await s.addParticipant(threadId, alice, 'bob');
+    await s.openThread(threadId, bob);
+    const msg = await s.send(threadId, alice, { content: 'ship it' }, 'k1');
+
+    const add = await s.toggleAnnotation(msg.id, bob, 'reaction', '👍');
+    expect(add.op).toBe('add');
+    expect(add.users).toEqual(['bob']);
+
+    const add2 = await s.toggleAnnotation(msg.id, alice, 'reaction', '👍'); // alice also 👍
+    expect(add2.op).toBe('add');
+    expect(add2.users).toEqual(['alice', 'bob']); // sorted, deterministic
+
+    const rem = await s.toggleAnnotation(msg.id, bob, 'reaction', '👍'); // bob toggles off
+    expect(rem.op).toBe('remove');
+    expect(rem.users).toEqual(['alice']);
+
+    const hist = await s.history(threadId);
+    const m = hist.find((x) => x.id === msg.id)!;
+    expect(m.annotations).toEqual([{ type: 'reaction', value: '👍', users: ['alice'] }]);
+  });
+
+  it('different values coexist for the same actor (👍 and 🎉 both stick)', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    const msg = await s.send(threadId, alice, { content: 'hi' }, 'k1');
+    await s.toggleAnnotation(msg.id, alice, 'reaction', '👍');
+    await s.toggleAnnotation(msg.id, alice, 'reaction', '🎉');
+    const m = (await s.history(threadId)).find((x) => x.id === msg.id)!;
+    expect(m.annotations).toEqual(
+      expect.arrayContaining([
+        { type: 'reaction', value: '👍', users: ['alice'] },
+        { type: 'reaction', value: '🎉', users: ['alice'] },
+      ]),
+    );
+  });
+
+  it('a non-member cannot annotate (governed by policy)', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN' });
+    const msg = await s.send(threadId, alice, { content: 'secret' }, 'k1');
+    await expect(s.toggleAnnotation(msg.id, bob, 'reaction', '👍')).rejects.toBeInstanceOf(PolicyDeniedError);
+  });
+
+  it('listMyAnnotated returns the caller’s saved messages with thread context; others see none', async () => {
+    const s = gov();
+    const { threadId } = await s.openThread(null, alice, { membership: 'group', creatorRole: 'ADMIN', subject: 'Design' });
+    await s.addParticipant(threadId, alice, 'bob');
+    const msg = await s.send(threadId, alice, { content: 'save this' }, 'k1');
+    await s.toggleAnnotation(msg.id, alice, 'save', ''); // save is a personal annotation
+
+    const saved = await s.listMyAnnotated(alice, 'save');
+    expect(saved).toHaveLength(1);
+    expect(saved[0]).toMatchObject({ threadId, threadSubject: 'Design' });
+    expect(saved[0]?.message.content).toBe('save this');
+    expect(await s.listMyAnnotated(bob, 'save')).toHaveLength(0); // bob saved nothing
+  });
+
+  it('send carries an OPAQUE mentions[] into the message event (kernel never parses @)', async () => {
+    const s = svc();
+    const { threadId } = await s.openThread(null, alice);
+    await s.openThread(threadId, bob);
+    await s.send(threadId, alice, { content: 'hi @bob' }, 'k1', undefined, undefined, ['bob']);
+    const ev = await prisma.iiosOutboxEvent.findFirstOrThrow({ where: { eventType: 'com.insignia.iios.message.sent.v1' } });
+    const ce = ev.cloudEvent as { data: { mentions?: string[] } };
+    expect(ce.data.mentions).toEqual(['bob']);
+  });
+});

packages/iios-service/src/platform/dev-opa.port.spec.ts

@@ -0,0 +1,45 @@
+import { describe, it, expect } from 'vitest';
+import { DevOpaPort } from './dev-opa.port';
+
+const opa = new DevOpaPort();
+
+describe('DevOpaPort (dev policy plane — membership rules)', () => {
+  it('allows every existing action by default (thread create/read, message send)', async () => {
+    for (const action of ['iios.thread.create', 'iios.thread.read', 'iios.message.send', undefined]) {
+      expect((await opa.decide({ action })).allow).toBe(true);
+    }
+  });
+
+  it('DM is capped at two: adding a 2nd is allowed, a 3rd is denied', async () => {
+    const add = (participantCount: number) =>
+      opa.decide({ action: 'iios.thread.participant.add', membership: 'dm', callerRole: 'MEMBER', participantCount });
+    expect((await add(1)).allow).toBe(true); // 1 → adding the 2nd
+    const third = await add(2); // 2 → adding a 3rd
+    expect(third.allow).toBe(false);
+    expect(third.obligations[0]?.reason).toMatch(/two people/);
+  });
+
+  it('adding a participant requires the caller be a member', async () => {
+    const d = await opa.decide({ action: 'iios.thread.participant.add', membership: 'group', callerRole: 'NONE', participantCount: 3 });
+    expect(d.allow).toBe(false);
+    expect(d.obligations[0]?.reason).toMatch(/member/);
+  });
+
+  it('group add/remove requires ADMIN', async () => {
+    const asMember = await opa.decide({ action: 'iios.thread.participant.add', membership: 'group', callerRole: 'MEMBER', participantCount: 3 });
+    expect(asMember.allow).toBe(false);
+    expect(asMember.obligations[0]?.reason).toMatch(/admin/);
+    const asAdmin = await opa.decide({ action: 'iios.thread.participant.add', membership: 'group', callerRole: 'ADMIN', participantCount: 3 });
+    expect(asAdmin.allow).toBe(true);
+  });
+
+  it('self-join is governed only on membership threads', async () => {
+    // generic / support thread (no membership attr) → open join, unchanged
+    expect((await opa.decide({ action: 'iios.thread.join', alreadyMember: false })).allow).toBe(true);
+    // a membership (chat) thread → existing member allowed, stranger denied
+    expect((await opa.decide({ action: 'iios.thread.join', membership: 'group', alreadyMember: true })).allow).toBe(true);
+    const stranger = await opa.decide({ action: 'iios.thread.join', membership: 'group', alreadyMember: false });
+    expect(stranger.allow).toBe(false);
+    expect(stranger.obligations[0]?.reason).toMatch(/not a member/);
+  });
+});

packages/iios-service/src/platform/dev-opa.port.ts

@@ -0,0 +1,49 @@
+import type { PolicyDecision } from '@insignia/iios-contracts';
+
+/**
+ * Dev stand-in for the real OPA policy plane. It evaluates a small **policy table**
+ * against the generic `{ action, ...context }` the kernel passes to `opa.decide`.
+ * Everything defaults to ALLOW (so existing actions are unchanged); only the rules
+ * below deny. This is the *policy plane*, NOT the kernel — the chat meaning of "dm"
+ * vs "group" lives here as policy, and a real OPA drops in behind the same interface.
+ */
+export interface OpaInput {
+  action?: string;
+  membership?: string; // app-set generic thread attribute: 'dm' | 'group'
+  participantCount?: number;
+  callerRole?: string; // 'MEMBER' | 'ADMIN'
+  alreadyMember?: boolean;
+  isMember?: boolean;
+  [k: string]: unknown;
+}
+
+const allow = (): PolicyDecision => ({ decisionRef: 'dev-allow', allow: true, obligations: [], ttlSeconds: 60 });
+const deny = (reason: string): PolicyDecision => ({ decisionRef: 'dev-deny', allow: false, obligations: [{ kind: 'AUDIT', reason }], ttlSeconds: 0 });
+
+export class DevOpaPort {
+  async decide(input: unknown): Promise<PolicyDecision> {
+    const i = (input ?? {}) as OpaInput;
+    switch (i.action) {
+      case 'iios.thread.participant.add': {
+        const count = i.participantCount ?? 0;
+        if (i.membership === 'dm' && count >= 2) return deny('a direct message is limited to two people');
+        if (i.callerRole !== 'MEMBER' && i.callerRole !== 'ADMIN') return deny('only a member can add participants');
+        if (i.membership === 'group' && i.callerRole !== 'ADMIN') return deny('only a group admin can add or remove members');
+        return allow();
+      }
+      case 'iios.thread.join': {
+        // Only threads that opted into a membership model (chat dm/group) are governed;
+        // generic/support threads (no membership attr) keep open-join. For a governed
+        // thread, new members must be added first (governed add-participant).
+        if (!i.membership) return allow();
+        return i.alreadyMember ? allow() : deny('you are not a member of this thread');
+      }
+      case 'iios.interaction.annotate': {
+        // Annotating (e.g. reacting to) a message requires being a participant of its thread.
+        return i.isMember ? allow() : deny('you are not a member of this thread');
+      }
+      default:
+        return allow();
+    }
+  }
+}

packages/iios-service/src/platform/platform-ports.ts

@@ -1,10 +1,10 @@
 import type {
   IiosPlatformPorts,
   PlatformPrincipal,
-  PolicyDecision,
   ContextDecisionBundle,
   SourceHandleResolution,
 } from '@insignia/iios-contracts';
+import { DevOpaPort } from './dev-opa.port';
 
 /** DI token for the platform ports (Session/OPA/CMP/MDM/CRRE/SAS/Capability). */
 export const PLATFORM_PORTS = Symbol('PLATFORM_PORTS');
@@ -22,11 +22,8 @@ export class LocalDevPorts implements IiosPlatformPorts {
       return { principalRef: 'local-dev', orgId: 'org_demo', appId: 'portal-demo', assurance: 'service' };
     },
   };
-  opa = {
-    async decide(_input: unknown): Promise<PolicyDecision> {
-      return { decisionRef: 'local-allow', allow: true, obligations: [], ttlSeconds: 60 };
-    },
-  };
+  // Dev policy plane: default-allow + the membership rules (real OPA swaps in here).
+  opa = new DevOpaPort();
   cmp = {
     async checkPurpose(_input: unknown): Promise<{ receiptRef?: string; status: 'ALLOW' | 'DENY' | 'NOT_REQUIRED' }> {
       return { status: 'NOT_REQUIRED' };

packages/iios-service/src/platform/policy-denied.filter.ts

@@ -0,0 +1,16 @@
+import { ArgumentsHost, Catch, HttpStatus, type ExceptionFilter } from '@nestjs/common';
+import { PolicyDeniedError } from '@insignia/iios-contracts';
+import type { Response } from 'express';
+
+/**
+ * Maps a fail-closed `PolicyDeniedError` (from `decideOrThrow`) to HTTP 403 instead of a
+ * generic 500, so callers can tell "policy denied" (with the reason) from a server fault.
+ * Applies to every policy-gated REST endpoint (e.g. the DM-cap / membership rules).
+ */
+@Catch(PolicyDeniedError)
+export class PolicyDeniedFilter implements ExceptionFilter {
+  catch(err: PolicyDeniedError, host: ArgumentsHost): void {
+    const res = host.switchToHttp().getResponse<Response>();
+    res.status(HttpStatus.FORBIDDEN).json({ statusCode: 403, code: err.code, message: err.message });
+  }
+}

packages/iios-service/src/platform/session.verifier.spec.ts

@@ -0,0 +1,75 @@
+import { describe, it, expect, beforeAll, afterAll, vi } from 'vitest';
+import { generateKeyPairSync, type KeyObject } from 'node:crypto';
+import jwt from 'jsonwebtoken';
+import { SessionVerifier } from './session.verifier';
+
+// Two independent fake IdPs (each its own EC signing key + kid).
+function makeIssuer(kid: string) {
+  const { publicKey, privateKey } = generateKeyPairSync('ec', { namedCurve: 'P-256' });
+  const jwk = { ...(publicKey.export({ format: 'jwk' }) as Record<string, unknown>), kid, use: 'sig', alg: 'ES256' };
+  return { privateKey, jwks: { keys: [jwk] } };
+}
+const chat = makeIssuer('chat-kid');
+const support = makeIssuer('support-kid');
+const CHAT_URL = 'https://chat-proj.example.co';
+const SUPPORT_URL = 'https://support-proj.example.co';
+
+function sat(priv: KeyObject, url: string, kid: string, email: string, name: string): string {
+  return jwt.sign(
+    { email, user_metadata: { full_name: name }, role: 'authenticated' },
+    priv,
+    { algorithm: 'ES256', issuer: `${url}/auth/v1`, audience: 'authenticated', subject: 'uuid-x', keyid: kid, expiresIn: '1h' },
+  );
+}
+
+describe('SessionVerifier — multi-issuer (OIDC/JWKS registry)', () => {
+  let verifier: SessionVerifier;
+
+  beforeAll(async () => {
+    process.env.AUTH_ISSUERS = JSON.stringify([
+      { url: CHAT_URL, appId: 'portal-demo' },
+      { url: SUPPORT_URL, appId: 'support-app' },
+    ]);
+    // Serve each issuer's JWKS from its own well-known URL.
+    vi.stubGlobal(
+      'fetch',
+      vi.fn(async (u: string) => {
+        const s = String(u);
+        const body = s.includes('chat-proj') ? chat.jwks : s.includes('support-proj') ? support.jwks : { keys: [] };
+        return { ok: true, json: async () => body } as unknown as Response;
+      }),
+    );
+    verifier = new SessionVerifier();
+    await verifier.onModuleInit();
+  });
+
+  afterAll(() => {
+    vi.unstubAllGlobals();
+    delete process.env.AUTH_ISSUERS;
+  });
+
+  it('routes a token to the appId of its own issuer', () => {
+    expect(verifier.verify(sat(chat.privateKey, CHAT_URL, 'chat-kid', 'Alice@x.com', 'Alice'))).toMatchObject({
+      userId: 'alice@x.com',
+      appId: 'portal-demo',
+      displayName: 'Alice',
+    });
+  });
+
+  it('a second issuer maps to a DIFFERENT appId — isolated scope', () => {
+    expect(verifier.verify(sat(support.privateKey, SUPPORT_URL, 'support-kid', 'Bob@x.com', 'Bob'))).toMatchObject({
+      userId: 'bob@x.com',
+      appId: 'support-app',
+    });
+  });
+
+  it('rejects a token from an untrusted issuer', () => {
+    const rogue = makeIssuer('rogue-kid');
+    expect(() => verifier.verify(sat(rogue.privateKey, 'https://evil.example.co', 'rogue-kid', 'e@x.com', 'E'))).toThrow();
+  });
+
+  it('rejects a forgery: signed by issuer B but claiming issuer A + A’s kid', () => {
+    const forged = sat(support.privateKey, CHAT_URL, 'chat-kid', 'x@x.com', 'X'); // wrong key for the claimed issuer
+    expect(() => verifier.verify(forged)).toThrow();
+  });
+});

packages/iios-service/src/platform/session.verifier.ts

@@ -1,30 +1,148 @@
-import { Injectable, UnauthorizedException } from '@nestjs/common';
+import { createPublicKey } from 'node:crypto';
+import { Injectable, OnModuleInit, UnauthorizedException } from '@nestjs/common';
 import jwt from 'jsonwebtoken';
 import type { MessagePrincipal } from '../messaging/message.service';
 
+/** One trusted OIDC issuer (a Supabase project / IdP) → the app scope it maps to. */
+interface IssuerEntry {
+  issuer: string; // the token `iss` claim, e.g. https://<ref>.supabase.co/auth/v1
+  jwksUrl: string;
+  appId: string;
+  orgId: string;
+  audience: string;
+  kidToPem: Map<string, string>;
+}
+
 /**
- * The real `session` port for P2: verifies a host app's HS256 token (reuses the
- * support-service AppTokenVerifier pattern). Per-app secrets come from the
- * APP_SECRETS env (JSON map keyed by appId). Expected claims: { sub, name?,
- * appId, orgId?, tenantId? }.
+ * The `session` port. Verifies whatever token a caller presents:
+ *
+ *  1. OIDC / JWKS (real IdPs) — a REGISTRY of trusted issuers (env `AUTH_ISSUERS`,
+ *     or the single `SUPABASE_URL` shorthand). A token is routed by its `iss` claim
+ *     to that issuer's entry, verified against that issuer's public JWKS (ES256, no
+ *     secret), and stamped with that entry's `appId`/`orgId`. So two projects/IdPs
+ *     map to two isolated app scopes on one IIOS — app A's tokens can't reach app B.
+ *
+ *  2. App token (dev / HS256) — legacy per-app secrets from `APP_SECRETS`, keyed by
+ *     the `appId` claim. Unchanged; used by the dev IdP + tests.
+ *
+ * A Session Broker would later collapse case 1 to a single issuer (the Broker's PAT).
  */
 @Injectable()
-export class SessionVerifier {
-  private readonly secrets: Record<string, string>;
+export class SessionVerifier implements OnModuleInit {
+  private readonly appSecrets: Record<string, string>;
+  private readonly issuers = new Map<string, IssuerEntry>(); // keyed by issuer string
 
   constructor() {
     try {
-      this.secrets = JSON.parse(process.env.APP_SECRETS ?? '{}');
+      this.appSecrets = JSON.parse(process.env.APP_SECRETS ?? '{}');
     } catch {
-      this.secrets = {};
+      this.appSecrets = {};
+    }
+    for (const cfg of this.readIssuerConfig()) {
+      const url = this.normalize(cfg.url);
+      const appId = cfg.appId?.trim() || 'portal-demo';
+      const issuer = cfg.issuer?.trim() || `${url}/auth/v1`;
+      const jwksUrl = cfg.jwksUrl?.trim() || `${url}/auth/v1/.well-known/jwks.json`;
+      this.issuers.set(issuer, {
+        issuer,
+        jwksUrl,
+        appId,
+        orgId: cfg.orgId?.trim() || `org_${appId}`,
+        audience: cfg.audience?.trim() || 'authenticated',
+        kidToPem: new Map(),
+      });
+    }
+  }
+
+  async onModuleInit(): Promise<void> {
+    await Promise.all([...this.issuers.values()].map((e) => this.refreshJwks(e)));
+  }
+
+  /** Assemble the issuer registry from AUTH_ISSUERS (multi) or SUPABASE_URL (single, back-compat). */
+  private readIssuerConfig(): Array<{ url: string; appId?: string; orgId?: string; issuer?: string; jwksUrl?: string; audience?: string }> {
+    const out: Array<{ url: string; appId?: string; orgId?: string; issuer?: string; jwksUrl?: string; audience?: string }> = [];
+    try {
+      const raw = process.env.AUTH_ISSUERS;
+      if (raw) {
+        const parsed = JSON.parse(raw) as Array<{ url: string; appId?: string; orgId?: string; issuer?: string; jwksUrl?: string; audience?: string }>;
+        if (Array.isArray(parsed)) out.push(...parsed.filter((e) => e && (e.url || e.issuer)));
+      }
+    } catch {
+      /* ignore malformed AUTH_ISSUERS */
+    }
+    const single = process.env.SUPABASE_URL?.trim();
+    if (single && !out.length) {
+      out.push({ url: single, appId: process.env.SUPABASE_APP_ID?.trim(), orgId: process.env.SUPABASE_ORG_ID?.trim() });
+    }
+    return out;
+  }
+
+  /** Fetch one issuer's JWKS and cache each key as PEM (so verify() stays synchronous). */
+  private async refreshJwks(entry: IssuerEntry): Promise<void> {
+    try {
+      const res = await fetch(entry.jwksUrl);
+      if (!res.ok) return;
+      const { keys } = (await res.json()) as { keys: Array<Record<string, unknown>> };
+      const next = new Map<string, string>();
+      for (const jwk of keys ?? []) {
+        const kid = jwk.kid as string | undefined;
+        if (!kid) continue;
+        try {
+          next.set(kid, createPublicKey({ key: jwk as never, format: 'jwk' }).export({ type: 'spki', format: 'pem' }) as string);
+        } catch {
+          /* skip a key we can't import */
+        }
+      }
+      if (next.size) entry.kidToPem = next;
+    } catch {
+      /* keep whatever keys we already have */
     }
   }
 
   verify(token: string): MessagePrincipal {
-    const decoded = jwt.decode(token) as jwt.JwtPayload | null;
-    const appId = decoded?.appId ? String(decoded.appId) : undefined;
+    const decoded = jwt.decode(token, { complete: true }) as { header?: { alg?: string; kid?: string }; payload?: jwt.JwtPayload } | null;
+    if (!decoded?.payload) throw new UnauthorizedException('invalid token');
+
+    if (this.issuers.size && decoded.header?.alg === 'ES256') {
+      const iss = decoded.payload.iss ? String(decoded.payload.iss) : '';
+      const entry = this.issuers.get(iss);
+      if (!entry) throw new UnauthorizedException(`untrusted issuer: ${iss || '(none)'}`);
+      return this.verifyOidc(token, decoded.header.kid, entry);
+    }
+    return this.verifyAppToken(token, decoded.payload);
+  }
+
+  /** Verify an ES256 SAT against its issuer's JWKS key, mapping to that issuer's app scope. */
+  private verifyOidc(token: string, kid: string | undefined, entry: IssuerEntry): MessagePrincipal {
+    const pem = kid ? entry.kidToPem.get(kid) : undefined;
+    if (!pem) {
+      void this.refreshJwks(entry); // rotated / not loaded — pull fresh for next time
+      throw new UnauthorizedException('unknown signing key — retry');
+    }
+    let payload: jwt.JwtPayload;
+    try {
+      payload = jwt.verify(token, pem, { algorithms: ['ES256'], issuer: entry.issuer, audience: entry.audience }) as jwt.JwtPayload;
+    } catch {
+      throw new UnauthorizedException('invalid token');
+    }
+    const email = payload.email ? String(payload.email).toLowerCase() : undefined;
+    const meta = (payload.user_metadata ?? {}) as { full_name?: string; name?: string };
+    // userId = email (stable + human-readable → mentions/directory work); RealMDM canonicalises `sub` later.
+    const userId = email ?? String(payload.sub);
+    return {
+      userId,
+      appId: entry.appId,
+      orgId: entry.orgId,
+      tenantId: undefined,
+      displayName: meta.full_name ?? meta.name ?? email ?? userId,
+    };
+  }
+
+  /** Legacy per-app HS256 token (dev IdP / host apps). */
+  private verifyAppToken(token: string, decodedPayload: jwt.JwtPayload): MessagePrincipal {
+    const appId = decodedPayload.appId ? String(decodedPayload.appId) : undefined;
     if (!appId) throw new UnauthorizedException('missing appId claim');
-    const secret = this.secrets[appId];
+    const secret = this.appSecrets[appId];
     if (!secret) throw new UnauthorizedException(`unknown app: ${appId}`);
 
     let payload: jwt.JwtPayload;
@@ -43,4 +161,9 @@ export class SessionVerifier {
       displayName: payload.name ? String(payload.name) : undefined,
     };
   }
+
+  /** Accept a project URL in any shape (…/rest/v1, …/auth/v1, trailing slash). */
+  private normalize(url: string): string {
+    return url.trim().replace(/\/+$/, '').replace(/\/(rest|auth)\/v1$/, '');
+  }
 }

packages/iios-service/src/threads/send-message.dto.ts

@@ -3,4 +3,5 @@ import { IsOptional, IsString } from 'class-validator';
 export class SendMessageDto {
   @IsString() content!: string;
   @IsOptional() @IsString() contentRef?: string;
+  @IsOptional() @IsString() parentInteractionId?: string;
 }

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

@@ -8,6 +8,7 @@ import {
   HttpCode,
   Param,
   Post,
+  Query,
 } from '@nestjs/common';
 import { ThreadsService } from './threads.service';
 import { MessageService } from '../messaging/message.service';
@@ -22,6 +23,32 @@ export class ThreadsController {
     private readonly session: SessionVerifier,
   ) {}
 
+  /** Generic "my threads" — every thread the caller participates in (last message + unread). */
+  @Get()
+  async listThreads(@Headers('authorization') auth?: string) {
+    return this.messages.listThreads(this.principal(auth));
+  }
+
+  /** Generic "my annotated messages" (e.g. ?type=save for a personal bookmarks list). */
+  @Get('my-annotations')
+  async myAnnotations(@Headers('authorization') auth?: string, @Query('type') type = 'save') {
+    return this.messages.listMyAnnotated(this.principal(auth), type);
+  }
+
+  /** Create a thread; `membership`/`creatorRole`/`subject` are opaque, app-supplied attributes the kernel stores but never interprets. */
+  @Post()
+  @HttpCode(201)
+  async createThread(@Body() body: { membership?: string; creatorRole?: string; subject?: string }, @Headers('authorization') auth?: string) {
+    return this.messages.openThread(null, this.principal(auth), { membership: body?.membership, creatorRole: body?.creatorRole, subject: body?.subject });
+  }
+
+  /** Governed membership: add a user (by userId) to a thread — policy enforces DM cap / roles. */
+  @Post(':id/participants')
+  @HttpCode(201)
+  async addParticipant(@Param('id') id: string, @Body() body: { userId: string; role?: string }, @Headers('authorization') auth?: string) {
+    return this.messages.addParticipant(id, this.principal(auth), body.userId, body.role);
+  }
+
   @Get(':id/messages')
   async listMessages(@Param('id') id: string) {
     return this.threads.getMessages(id);
@@ -36,14 +63,19 @@ export class ThreadsController {
     @Headers('authorization') authorization?: string,
     @Headers('idempotency-key') idempotencyKey?: string,
   ) {
-    const token = (authorization ?? '').replace(/^Bearer\s+/i, '');
-    if (!token) throw new BadRequestException('Authorization bearer token is required');
-    const principal = this.session.verify(token);
     return this.messages.send(
       id,
-      principal,
+      this.principal(authorization),
       { content: body.content, contentRef: body.contentRef },
       idempotencyKey ?? randomUUID(),
+      undefined,
+      body.parentInteractionId,
     );
   }
+
+  private principal(authorization?: string) {
+    const token = (authorization ?? '').replace(/^Bearer\s+/i, '');
+    if (!token) throw new BadRequestException('Authorization bearer token is required');
+    return this.session.verify(token);
+  }
 }

scripts/vitest-global-setup.mjs

@@ -0,0 +1,20 @@
+import { execSync } from 'node:child_process';
+
+// Tests TRUNCATE between cases, so they MUST NOT touch the dev database. Run them
+// against an isolated `iios_test` DB (created + migrated here, once, before the suite).
+const TEST_URL = 'postgresql://iios:iios@localhost:5434/iios_test?schema=public';
+
+export default function setup() {
+  try {
+    execSync(`docker exec iios-db psql -U iios -d postgres -c "CREATE DATABASE iios_test"`, { stdio: 'pipe' });
+  } catch (e) {
+    const msg = String(e.stderr ?? e.stdout ?? e);
+    if (!/already exists/i.test(msg)) {
+      console.warn(`[vitest] could not create iios_test (is the iios-db container up?): ${msg.slice(0, 160)}`);
+    }
+  }
+  execSync('pnpm --filter @insignia/iios-service exec prisma migrate deploy', {
+    stdio: 'inherit',
+    env: { ...process.env, DATABASE_URL: TEST_URL },
+  });
+}

vitest.config.ts

@@ -15,9 +15,13 @@ export default defineConfig({
   },
   test: {
     include: ['packages/**/src/**/*.{test,spec}.ts', 'test/**/*.{test,spec}.ts'],
-    // DB-backed specs share one Postgres and TRUNCATE between tests. Run every
-    // file in a single worker process, sequentially, so there is no cross-file
-    // race on the shared database.
+    // DB-backed specs TRUNCATE between tests, so they run against an ISOLATED
+    // `iios_test` database (created + migrated by the global setup) — never the dev
+    // DB. This env overrides any DATABASE_URL from the shell.
+    env: { DATABASE_URL: 'postgresql://iios:iios@localhost:5434/iios_test?schema=public' },
+    globalSetup: ['./scripts/vitest-global-setup.mjs'],
+    // Run every file in a single worker process, sequentially, so there is no
+    // cross-file race on the shared database.
     fileParallelism: false,
     sequence: { concurrent: false },
     pool: 'forks',