Narrow-set filters the seat's own queue, so a receiver always has an identity (comms-g5zh.6)

packages/mcp/event-pump.ts

@@ -56,7 +56,7 @@ export interface EventPumpDeps {
   /**
    * Narrow-filter predicate. The pump tees an event to the notifier
    * only when this returns true. Production wires this to
-   * `NarrowSet.matches(event, getBotIdentityId())`. Omit for
+   * `NarrowSet.matches` under the seat's bound identity. Omit for
    * "deliver everything" — useful in tests that don't exercise the
    * narrow filter.
    */

packages/mcp/narrow-set.test.ts

@@ -125,13 +125,17 @@ const newTopicsIntent = (channel: string): SubscribeIntent => ({
   channelName: decodeChannelNameSync(channel),
 })
 
-const noBot: IdentityIdType | undefined = undefined
+// A receiving seat always holds an identity — the events queue is registered
+// against its own principal, so an unbound seat produces no events to filter.
+// Tests whose subject is not mentions still supply one; this seat is never in
+// any message's mention list, so it never widens a match on its own.
+const seat: IdentityIdType = buildIdentity('bot:seat', 'seat').id
 
 test('empty narrow set matches nothing', () => {
   const set = createNarrowSet()
   expect(set.size()).toBe(0)
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
-  expect(set.matches(buildReactionAdded('home'), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
+  expect(set.matches(buildReactionAdded('home'), seat)).toBe(false)
 })
 
 // comms-n1my. The one exception to "empty narrow → nothing delivered", and the
@@ -153,45 +157,45 @@ test('empty narrow set still matches a mention of the bound bot', () => {
 test('channel narrow matches message-posted on that channel', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('other', undefined), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('other', undefined), seat)).toBe(false)
 })
 
 test('channel narrow matches message-posted in any thread of that channel', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
-  expect(set.matches(buildMessagePosted('home', 'payments'), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'payments'), seat)).toBe(true)
 })
 
 test('thread narrow matches only the named thread within the channel', () => {
   const set = createNarrowSet()
   set.add(threadIntent('home', 'payments'))
-  expect(set.matches(buildMessagePosted('home', 'payments'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('home', 'breakfast'), noBot)).toBe(false)
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
-  expect(set.matches(buildMessagePosted('other', 'payments'), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', 'payments'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'breakfast'), seat)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
+  expect(set.matches(buildMessagePosted('other', 'payments'), seat)).toBe(false)
 })
 
 test('thread narrow does not alias a distinct channel/thread pair sharing a slash boundary', () => {
   const set = createNarrowSet()
   set.add(threadIntent('a', 'b/c'))
-  expect(set.matches(buildMessagePosted('a', 'b/c'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('a/b', 'c'), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('a', 'b/c'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('a/b', 'c'), seat)).toBe(false)
 })
 
 test('channel narrow matches reaction-added/removed on that channel', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
-  expect(set.matches(buildReactionAdded('home'), noBot)).toBe(true)
-  expect(set.matches(buildReactionRemoved('home'), noBot)).toBe(true)
-  expect(set.matches(buildReactionAdded('other'), noBot)).toBe(false)
+  expect(set.matches(buildReactionAdded('home'), seat)).toBe(true)
+  expect(set.matches(buildReactionRemoved('home'), seat)).toBe(true)
+  expect(set.matches(buildReactionAdded('other'), seat)).toBe(false)
 })
 
 test('thread narrow matches reactions on a message in that thread', () => {
   const set = createNarrowSet()
   set.add(threadIntent('home', 'payments'))
-  expect(set.matches(buildReactionAdded('home', 'payments'), noBot)).toBe(true)
-  expect(set.matches(buildReactionAdded('home', 'other'), noBot)).toBe(false)
+  expect(set.matches(buildReactionAdded('home', 'payments'), seat)).toBe(true)
+  expect(set.matches(buildReactionAdded('home', 'other'), seat)).toBe(false)
 })
 
 test('a mention matches message-posted only when the bot is among the mentions', () => {
@@ -202,12 +206,6 @@ test('a mention matches message-posted only when the bot is among the mentions',
   expect(set.matches(buildMessagePosted('home', undefined, [other]), bot.id)).toBe(false)
 })
 
-test('a mention does not match pre-acquire (no bot identity)', () => {
-  const set = createNarrowSet()
-  const bot = buildIdentity('bot:me', 'me')
-  expect(set.matches(buildMessagePosted('home', undefined, [bot]), noBot)).toBe(false)
-})
-
 test('a mention matches mention-received events whenever the bot is in mentions', () => {
   const set = createNarrowSet()
   const bot = buildIdentity('bot:me', 'me')
@@ -228,10 +226,10 @@ test('combined channel + thread narrows widen the match window alongside mention
 test('remove drops an intent so subsequent events no longer match', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(true)
   set.remove(channelIntent('home'))
   expect(set.size()).toBe(0)
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
 })
 
 test('remove on an unknown intent is a no-op', () => {
@@ -260,53 +258,53 @@ test('thread and channel narrows for the same channel coexist as distinct entrie
 test('new-topics narrow matches the first message of a topic on that channel', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('elsewhere', 'fresh-topic'), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('elsewhere', 'fresh-topic'), seat)).toBe(false)
 })
 
 test('new-topics narrow does not match a topic-less message', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
 })
 
 test('new-topics narrow does not match a second message in an already-seen topic', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), seat)).toBe(false)
 })
 
 test('new-topics narrow tracks each topic independently', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildMessagePosted('home', 'topic-a'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('home', 'topic-b'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('home', 'topic-a'), noBot)).toBe(false)
-  expect(set.matches(buildMessagePosted('home', 'topic-b'), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', 'topic-a'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'topic-b'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'topic-a'), seat)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', 'topic-b'), seat)).toBe(false)
 })
 
 test('new-topics first-per-topic is scoped per channel, not channel-wide', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
   set.add(newTopicsIntent('elsewhere'))
-  expect(set.matches(buildMessagePosted('home', 'shared-name'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('elsewhere', 'shared-name'), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'shared-name'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('elsewhere', 'shared-name'), seat)).toBe(true)
 })
 
 test('channel narrow still matches every message in a topic, even after new-topics has seen it', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), noBot)).toBe(true)
-  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), seat)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), seat)).toBe(true)
 })
 
 test('new-topics narrow does not match reactions on the channel', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildReactionAdded('home', 'fresh-topic'), noBot)).toBe(false)
-  expect(set.matches(buildReactionRemoved('home', 'fresh-topic'), noBot)).toBe(false)
+  expect(set.matches(buildReactionAdded('home', 'fresh-topic'), seat)).toBe(false)
+  expect(set.matches(buildReactionRemoved('home', 'fresh-topic'), seat)).toBe(false)
 })
 
 // The new-topics narrow is first-message-per-topic and says nothing about
@@ -334,7 +332,7 @@ test('new-topics narrow remove drops only the new-topics entry', () => {
   set.add(newTopicsIntent('home'))
   set.remove(newTopicsIntent('home'))
   expect(set.size()).toBe(1)
-  expect(set.matches(buildReactionAdded('home'), noBot)).toBe(true)
+  expect(set.matches(buildReactionAdded('home'), seat)).toBe(true)
 })
 
 const sortIntents = (intents: ReadonlyArray<SubscribeIntent>): ReadonlyArray<SubscribeIntent> =>
@@ -367,7 +365,7 @@ test('intents() excludes a removed intent', () => {
 test('intents() does not leak the seen-topics ledger after a new-topics match', () => {
   const set = createNarrowSet()
   set.add(newTopicsIntent('home'))
-  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', 'fresh-topic'), seat)).toBe(true)
   expect(set.intents()).toEqual([newTopicsIntent('home')])
 })
 
@@ -377,16 +375,16 @@ test('load(Some) sets the base on an empty set', () => {
   expect(sortIntents(set.intents())).toEqual(
     sortIntents([channelIntent('home'), newTopicsIntent('general')]),
   )
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(true)
 })
 
 test('load(Some) replaces the prior base — the old set no longer matches', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
   set.load(Option.some([channelIntent('work')]))
   expect(set.intents()).toEqual([channelIntent('work')])
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
-  expect(set.matches(buildMessagePosted('work', undefined), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
+  expect(set.matches(buildMessagePosted('work', undefined), seat)).toBe(true)
 })
 
 test('load(Some([])) drops every narrow — matches nothing', () => {
@@ -396,15 +394,15 @@ test('load(Some([])) drops every narrow — matches nothing', () => {
   set.load(Option.some([]))
   expect(set.intents()).toEqual([])
   expect(set.size()).toBe(0)
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
 })
 
 test('load(None) keeps the current base — the fresh-session fallback', () => {
   const set = createNarrowSet()
   set.add(channelIntent('home'))
   set.load(Option.none())
   expect(set.intents()).toEqual([channelIntent('home')])
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(true)
 })
 
 test('a buffered subscribe replays onto the loaded base', () => {
@@ -425,7 +423,7 @@ test('a buffered unsubscribe of a base member is applied after the base loads',
   set.remove(channelIntent('home'))
   set.load(Option.some([channelIntent('home'), newTopicsIntent('general')]))
   expect(set.intents()).toEqual([newTopicsIntent('general')])
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
 })
 
 test('an env seed applied before buffering is not resurrected by load(Some) — dropped stays dropped', () => {
@@ -436,7 +434,7 @@ test('an env seed applied before buffering is not resurrected by load(Some) —
   // Persisted set unsubscribed the env default; nothing re-adds it.
   set.load(Option.some([]))
   expect(set.intents()).toEqual([])
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(false)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(false)
 })
 
 test('load(None) keeps the env seed and applies buffered deltas', () => {
@@ -454,5 +452,5 @@ test('a delta applies live during buffering — matches is not blacked out befor
   const set = createNarrowSet()
   set.beginBuffering()
   set.add(channelIntent('home'))
-  expect(set.matches(buildMessagePosted('home', undefined), noBot)).toBe(true)
+  expect(set.matches(buildMessagePosted('home', undefined), seat)).toBe(true)
 })

packages/mcp/narrow-set.ts

@@ -2,7 +2,6 @@ import type {
   ChannelName,
   IdentityId,
   InboundEvent,
-  Mention,
   MessageRef,
   ThreadName,
 } from '@commy/core/ports'
@@ -11,29 +10,31 @@ import { Array as Arr, Data, HashSet, Match, Option } from 'effect'
 import type { SubscribeIntent } from './subscribe-parser.ts'
 
 /**
- * Plugin-layer narrow filter for inbound events.
+ * Plugin-layer narrow filter for the seat's own event queue.
  *
- * The commy Zulip adapter ships the minter's full event stream
- * — every public stream the minter is subscribed to. The pump tees
- * an event to the MCP host only if at least one narrow intent in this
- * set matches. Empty narrow → no events delivered, matching today's
- * "you only see what you subscribed to" semantics.
+ * The queue belongs to the seat's binding, so this filters only events the
+ * seat's own principal received. The pump tees an event to the MCP host only
+ * if at least one narrow intent in this set matches. Empty narrow → no events
+ * delivered, matching today's "you only see what you subscribed to" semantics.
  *
- * Adding / removing intents is local-only; nothing here touches the
- * realm. Substrate-side minter-to-stream subscription is owned by the
- * boot-time reconciler plus the per-session substrate POST
- * inside `inbox.subscribe()` for streams created after the plugin
- * booted.
+ * What remains client-side is what the substrate cannot express: Zulip has no
+ * per-topic delivery primitive, so a thread narrow is filtered here rather
+ * than declared to the realm. Channel-level intent is realm state — a
+ * subscription under the seat's own principal — not a line in this set.
  *
  * Mentions are the one exception to "empty narrow → nothing delivered": a bot
  * always receives its own, with nothing to subscribe to and no way to opt out.
- * They match only once the bot identity is known (i.e. post-acquire) —
- * pre-acquire there is nothing to compare a message's mentions against.
  */
 export interface NarrowSet {
   add(intent: SubscribeIntent): void
   remove(intent: SubscribeIntent): void
-  matches(event: InboundEvent, botIdentityId: IdentityId | undefined): boolean
+  /**
+   * A receiving seat always holds an identity: the events queue is registered
+   * against the seat's own principal, so an unbound seat owns no queue to poll
+   * and produces no events at all (`adapter.ts` `ownerHttp`, `events.ts`'s
+   * `UnboundEphemeralSession` idle). `botIdentityId` is therefore required.
+   */
+  matches(event: InboundEvent, botIdentityId: IdentityId): boolean
   size(): number
   /**
    * Snapshot the current subscription intents (the membership set, not the
@@ -137,19 +138,6 @@ const refMatches = (ref: MessageRef, intents: HashSet.HashSet<IntentKey>): boole
   HashSet.has(intents, channelKey(ref)) ||
   Option.exists(threadKey(ref), (tk) => HashSet.has(intents, tk))
 
-/**
- * Unconditional: a bot always receives its own mentions, so this consults no
- * intent. That is what makes an empty narrow set survivable — a seat that has
- * subscribed to nothing still hears its own name.
- *
- * Still gated on the bot being bound. Pre-acquire there is no identity to
- * compare a message's mentions against, so nothing can match.
- */
-const mentionsMatches = (
-  mentions: ReadonlyArray<Mention>,
-  botIdentityId: IdentityId | undefined,
-): boolean => botIdentityId !== undefined && mentionsIdentity(mentions, botIdentityId)
-
 export const createNarrowSet = (): NarrowSet => {
   let intents = HashSet.empty<IntentKey>()
   let seenTopics = HashSet.empty<IntentKey>()
@@ -221,14 +209,17 @@ export const createNarrowSet = (): NarrowSet => {
       pending.length = 0
       recording = false
     },
+    // `mentionsIdentity` consults no intent: a bot always receives its own
+    // mentions. That is what makes an empty narrow set survivable — a seat
+    // that has subscribed to nothing still hears its own name.
     matches: (event, botIdentityId): boolean =>
       Match.value(event).pipe(
         Match.discriminatorsExhaustive('kind')({
           'message-posted': (e) =>
             messagePostedMatches(e.message.ref) ||
-            mentionsMatches(e.message.mentions, botIdentityId),
+            mentionsIdentity(e.message.mentions, botIdentityId),
           'mention-received': (e) =>
-            refMatches(e.message.ref, intents) || mentionsMatches(e.mentions, botIdentityId),
+            refMatches(e.message.ref, intents) || mentionsIdentity(e.mentions, botIdentityId),
           'reaction-added': (e) => refMatches(e.target, intents),
           'reaction-removed': (e) => refMatches(e.target, intents),
         }),

packages/mcp/server.ts

@@ -1042,7 +1042,15 @@ export const makeProgram = (
         inbox: adapter.inbox,
         notifier,
         getBotIdentityId,
-        match: (event) => narrowSet.matches(event, getBotIdentityId()),
+        // No binding means the seat owns no events queue, so an event in hand
+        // is not this seat's to filter — the same statement the producer makes
+        // one layer down by idling instead of polling when `ownerHttp` refuses
+        // (zulip/events.ts, `UnboundEphemeralSession`). Reached only if a
+        // release lands between an event being produced and dispatched.
+        match: (event) => {
+          const botIdentityId = getBotIdentityId()
+          return botIdentityId !== undefined && narrowSet.matches(event, botIdentityId)
+        },
         // Populate the tools-side identity cache from inbound events so
         // `presence` / `post` mentions / `react` can resolve ids only ever
         // seen via a notification.