diff --git a/packages/mcp/event-pump.ts b/packages/mcp/event-pump.ts index e77e8c9..4e16f89 100644 --- a/packages/mcp/event-pump.ts +++ b/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. */ diff --git a/packages/mcp/narrow-set.test.ts b/packages/mcp/narrow-set.test.ts index e782fdd..126e6e3 100644 --- a/packages/mcp/narrow-set.test.ts +++ b/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): ReadonlyArray => @@ -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,7 +375,7 @@ 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', () => { @@ -385,8 +383,8 @@ test('load(Some) replaces the prior base — the old set no longer matches', () 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,7 +394,7 @@ 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', () => { @@ -404,7 +402,7 @@ test('load(None) keeps the current base — the fresh-session fallback', () => { 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) }) diff --git a/packages/mcp/narrow-set.ts b/packages/mcp/narrow-set.ts index 0e7ea5f..1ac1264 100644 --- a/packages/mcp/narrow-set.ts +++ b/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): 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, - botIdentityId: IdentityId | undefined, -): boolean => botIdentityId !== undefined && mentionsIdentity(mentions, botIdentityId) - export const createNarrowSet = (): NarrowSet => { let intents = HashSet.empty() let seenTopics = HashSet.empty() @@ -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), }), diff --git a/packages/mcp/server.ts b/packages/mcp/server.ts index 78f8fc0..a4d10e6 100644 --- a/packages/mcp/server.ts +++ b/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. diff --git a/packages/mcp/subscription-resume.test.ts b/packages/mcp/subscription-resume.test.ts index d3d6845..b97e27d 100644 --- a/packages/mcp/subscription-resume.test.ts +++ b/packages/mcp/subscription-resume.test.ts @@ -46,6 +46,9 @@ const DECISIONS_CHANNEL = 'commy' const DECISIONS_THREAD = 'ref-types-split-decisions' const SID_RESUME = '61b08d76-0000-4000-8000-000000000001' const SID_FRESH = 'f9e5f9e5-0000-4000-8000-000000000002' +// The seat receiving the events these tests match. It is never mentioned in +// them — what's under test is narrow membership, not the mention path. +const SEAT_ID = decodeIdentityIdSync('bot:cc-61b08d76') const generalTopicsIntent: SubscribeIntent = { kind: 'new-topics-in-channel', @@ -166,7 +169,7 @@ test('restore rehydrates the persisted narrow set when the shared deferred is fi ) // Rebooted, deaf: nothing restored while the id is unknown. - expect(narrowSet.matches(reactionOnDecisionsThread(), undefined)).toBe(false) + expect(narrowSet.matches(reactionOnDecisionsThread(), SEAT_ID)).toBe(false) // Fill the shared deferred ALONE — no callTool. yield* Deferred.succeed(session, asSessionId(SID_RESUME)) @@ -174,7 +177,7 @@ test('restore rehydrates the persisted narrow set when the shared deferred is fi // Restored: the human's :one: reaction on the decisions thread now matches, // and the thread was re-wired on the substrate. - expect(narrowSet.matches(reactionOnDecisionsThread(), undefined)).toBe(true) + expect(narrowSet.matches(reactionOnDecisionsThread(), SEAT_ID)).toBe(true) expect(subscribes).toContainEqual(intentToTarget(decisionsThreadIntent)) expect(sortIntents(narrowSet.intents())).toEqual( sortIntents([decisionsThreadIntent, generalTopicsIntent]), @@ -266,7 +269,7 @@ test('a fresh session (no record) rebuilds from the realm and replays buffered d // Nothing re-declared on the substrate: this process already subscribed // `env-default` on the way in, so the rebuild has no round-trip to make. expect(subscribes).toEqual([]) - expect(narrowSet.matches(reactionOnDecisionsThread(), undefined)).toBe(false) + expect(narrowSet.matches(reactionOnDecisionsThread(), SEAT_ID)).toBe(false) }), )) @@ -295,7 +298,7 @@ test('a seat with no realm rows and no topic record comes back empty', () => yield* Fiber.join(fiber) expect(narrowSet.intents()).toEqual([]) - expect(narrowSet.matches(reactionOnDecisionsThread(), undefined)).toBe(false) + expect(narrowSet.matches(reactionOnDecisionsThread(), SEAT_ID)).toBe(false) }), ))