From 457e2ba37720d95be4870ffd6bca4604d971e425 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:14:32 +0100 Subject: [PATCH 1/6] Subscribe and register the events queue under the seat's own principal An event queue and a subscription row are both state the realm holds on the agent's behalf, so they belong under the agent's own principal rather than the shared minter's (docs/agent-experience.md principle 5). The two move together because Zulip couples them. A channel message is delivered only to the users the channel's subscription rows name (zerver/actions/message_send.py builds the recipient set from those rows, and a mention does not widen it), so a seat-owned queue over minter-held subscriptions would receive nothing at all. Queue ownership and subscription ownership are one unit, not two sequenced steps. inbox.subscribe / inbox.unsubscribe therefore reach through boundHttp, which makes reaching for a bound credential the declaration that the realm is about to hold state -- the same seam the write verbs use, with no second list to keep in agreement. A bind failure flows out untouched rather than being flattened into an InboxError, matching publish. Boot-time subscribe supplies the seat's naming inputs by POLLING the shared session-id deferred, never awaiting it. The boot-env feeder completes that deferred earlier in boot, so the only zero-action source has already fired by the time registration runs and there is no race left to lose; an await here would park the BOOT fiber, which is a hung MCP child rather than a merely deaf one. A seat that cannot bind logs what it lost and carries on serving. The hooks manifest guard now traces adapter.inbox.* alongside adapter.publisher.*. Tracing only the publisher receiver is exactly how a guard filed in May 2026 against this drift stayed green for eight months while a binding verb sat outside its compared set. subscribe and unsubscribe are unstamped by the PreToolUse matcher, so they are recorded in a named list the assertion holds to -- widening that matcher reverses 0f0e755 (PR #126) and is held pending ratification. comms-g5zh.2 comms-g5zh.3 Claude-Session: https://claude.ai/code/session_01NdgPHEUtz1xqbFEiN3cxCB --- clients/claude-code/hooks-manifest.test.ts | 101 +++++++++++- packages/core/ports.ts | 12 +- packages/mcp/bootstrap.ts | 16 +- packages/mcp/server.ts | 64 +++++++- packages/mcp/subscription-restore.ts | 6 +- packages/zulip/adapter.test.ts | 94 ++++++++++-- packages/zulip/adapter.ts | 170 +++++++++++++-------- packages/zulip/events.ts | 2 +- 8 files changed, 361 insertions(+), 104 deletions(-) diff --git a/clients/claude-code/hooks-manifest.test.ts b/clients/claude-code/hooks-manifest.test.ts index 70e3edb..1ab0fe6 100644 --- a/clients/claude-code/hooks-manifest.test.ts +++ b/clients/claude-code/hooks-manifest.test.ts @@ -53,9 +53,44 @@ const BOUND_HTTP_CALLERS = [ 'react', 'setChannelDescription', 'setThreadResolved', + 'subscribe', 'unreact', + 'unsubscribe', ] as const +/** + * Inbox verbs whose adapter implementation reaches `boundHttp` (comms-g5zh.2 / + * .3). Declaring interest writes realm state under the seat's own principal — + * a subscription row and the event queue that delivers against it — so these + * bind for the same reason the publisher verbs do. + * + * Held apart from {@link BOUND_VERBS} because the tool-side trace resolves them + * through a different receiver (`adapter.inbox.*`, not `adapter.publisher.*`). + * That distinction is the whole reason the pre-existing guard could not see + * them: it compared over publisher verbs alone, so a binding inbox verb sat + * outside the compared set entirely and the suite stayed green. + */ +const BOUND_INBOX_VERBS = ['subscribe', 'unsubscribe'] as const + +/** + * Tools that reach `boundHttp` through an inbox verb while sitting outside the + * matcher, so the hook never stamps them and the bind seam sees no session id. + * + * These are NOT a `comms-tww6`-style attribution accident — they cannot inherit + * an earlier call's seat, because `boundHttp` consults the binder on every call + * and refuses outright when the context carries no session id + * (`adapter.ts`, the "may THIS CALLER use it?" comment). An unstamped + * `subscribe` therefore FAILS rather than binding to the wrong seat. + * + * Emptying this list means widening the `hooks.json` matcher to the seven tools + * that declare `session_id`. That change reverses commit `0f0e755` (PR #126), + * which deliberately declined it while the shared minter still owned + * subscriptions — so it is a ratification call, not a lint fix, and is held + * pending that ruling. The list exists so the gap is machine-visible in the + * meantime rather than resting on prose. + */ +const G5ZH3_MATCHER_PENDING = ['subscribe', 'unsubscribe'] as const + /** * Tools that reach `boundHttp` while declaring no `session_id` and sitting * outside the matcher — the open P1 `comms-tww6`. They run under whatever seat @@ -91,18 +126,25 @@ function adapterVerbsReachingBoundHttp(source: string): ReadonlySet { interface ToolFacts { readonly verbs: ReadonlySet + readonly inboxVerbs: ReadonlySet readonly declaresSessionId: boolean } -/** Per-tool: which publisher verbs its handler calls, and whether it declares `session_id`. */ +/** + * Per-tool: which publisher verbs and which inbox verbs its handler calls, and + * whether it declares `session_id`. + */ function toolFactsFromToolsSource(source: string): ReadonlyMap { - const facts = new Map; declaresSessionId: boolean }>() + const facts = new Map< + string, + { verbs: Set; inboxVerbs: Set; declaresSessionId: boolean } + >() let current: string | undefined for (const line of source.split('\n')) { const named = line.match(/^ {6}name: '([a-z_]+)',$/)?.[1] if (named !== undefined) { current = named - facts.set(named, { verbs: new Set(), declaresSessionId: false }) + facts.set(named, { verbs: new Set(), inboxVerbs: new Set(), declaresSessionId: false }) } const entry = current === undefined ? undefined : facts.get(current) if (entry === undefined) continue @@ -110,9 +152,18 @@ function toolFactsFromToolsSource(source: string): ReadonlyMap [name, { ...e, verbs: e.verbs }])) + return new Map( + [...facts].map(([name, e]) => [name, { ...e, verbs: e.verbs, inboxVerbs: e.inboxVerbs }]), + ) } function alternationToolsFromMatcher(matcher: string): ReadonlySet { @@ -175,6 +226,25 @@ test('every tool whose adapter path reaches boundHttp is in the PreToolUse match expect(missing).toEqual([]) }) +// The same rule, stated over the INBOX verbs that began binding with +// comms-g5zh.2/.3. Kept as its own assertion with its own named list so the +// publisher-side rule above cannot go green on a set that no longer covers +// every binding path — the failure mode comms-65nj recorded, where a guard +// filed in May 2026 against exactly this drift stayed green for eight months +// because its compared set was scoped one level too low. +test('the tools that bind via an inbox verb but are unstamped are exactly the recorded set', async () => { + const facts = toolFactsFromToolsSource(await toolsSource()) + const matched = alternationToolsFromMatcher(injectSessionIdMatcher(hooksManifest)) + const unstamped = [...facts] + .filter(([, f]) => + [...f.inboxVerbs].some((v) => (BOUND_INBOX_VERBS as ReadonlyArray).includes(v)), + ) + .map(([name]) => name) + .filter((name) => !matched.has(name)) + .sort() + expect(unstamped).toEqual([...G5ZH3_MATCHER_PENDING]) +}) + // The rule stated over ALL bound verbs, including the two helper-backed ones // the tool layer never stamps. This is the assertion `comms-tww6` closes. test('comms-tww6: the known unstamped bound-path tools are exactly the recorded exceptions', async () => { @@ -230,10 +300,29 @@ test('toolFactsFromToolsSource attributes verbs and session_id to the enclosing handler: async () => { await run(adapter.history.readChannel(channel)) }, + name: 'gamma', + handler: async () => { + await run(adapter.inbox.subscribe(target)) + }, ` const facts = toolFactsFromToolsSource(synthetic) - expect(facts.get('alpha')).toEqual({ verbs: new Set(['post']), declaresSessionId: true }) - expect(facts.get('beta')).toEqual({ verbs: new Set(), declaresSessionId: false }) + expect(facts.get('alpha')).toEqual({ + verbs: new Set(['post']), + inboxVerbs: new Set(), + declaresSessionId: true, + }) + expect(facts.get('beta')).toEqual({ + verbs: new Set(), + inboxVerbs: new Set(), + declaresSessionId: false, + }) + // A read through the inbox is still traced as an inbox verb here; whether it + // BINDS is decided by `BOUND_INBOX_VERBS`, not by the receiver. + expect(facts.get('gamma')).toEqual({ + verbs: new Set(), + inboxVerbs: new Set(['subscribe']), + declaresSessionId: false, + }) }) test('alternationToolsFromMatcher splits the trailing parenthesised group', () => { diff --git a/packages/core/ports.ts b/packages/core/ports.ts index 00dc4b9..cd72012 100644 --- a/packages/core/ports.ts +++ b/packages/core/ports.ts @@ -663,8 +663,16 @@ export interface MessageInbox { * * and trust that the post will be observable on the stream. */ - subscribe(target: SubscriptionTarget): Effect.Effect - unsubscribe(target: SubscriptionTarget): Effect.Effect + /** + * Declaring interest writes realm state under the agent's own principal — + * a subscription row, and the event queue that delivers against it — so + * these carry {@link BindError} for the same reason the write verbs do. + * A seat that cannot bind is refused rather than silently falling back to + * a shared principal, which would leave it reading a surface that is not + * its own. + */ + subscribe(target: SubscriptionTarget): Effect.Effect + unsubscribe(target: SubscriptionTarget): Effect.Effect /** * Effect-native Stream of inbound events. Adapters drive this from * their substrate's event mechanism (Zulip's events queue, Discord diff --git a/packages/mcp/bootstrap.ts b/packages/mcp/bootstrap.ts index a193d6b..87706ab 100644 --- a/packages/mcp/bootstrap.ts +++ b/packages/mcp/bootstrap.ts @@ -1,4 +1,4 @@ -import type { BotName, InboxError, MessageInbox } from '@commy/core/ports' +import type { BindError, BotName, InboxError, MessageInbox } from '@commy/core/ports' import { decodeBotNameSync } from '@commy/core/ports' import type { ZulipAdapter } from '@commy/zulip/adapter' import { zulipAdapter } from '@commy/zulip/adapter' @@ -720,16 +720,20 @@ export const ZulipAdapterLive: Layer.Layer< * to tee only matching events to the MCP host. * 2. `inbox.subscribe` keeps the substrate side wired so the * adapter actually receives events. For Zulip this calls - * `/users/me/subscriptions` against the minter, ensuring the - * stream is in the minter's queue. The boot-time reconciler - * covers most streams; this per-session call still - * handles streams created after the plugin booted. + * `/users/me/subscriptions` under the SEAT's own principal and + * registers the seat's event queue, so the stream lands in a queue + * the seat owns rather than a shared one. + * + * Because that is realm state under the agent's own principal, this + * binds — hence the `BindError` in the error channel. A seat with no + * way to obtain an identity cannot hold subscriptions at all, and is + * refused here rather than silently seeded onto someone else's. */ export const subscribeFromEnv = ( inbox: MessageInbox, narrowSet: NarrowSet, parsed: ParsedEnv, -): Effect.Effect, SubscribeTokenError | InboxError> => { +): Effect.Effect, SubscribeTokenError | BindError | InboxError> => { if (parsed.subscribe === undefined) return Effect.succeed([]) return Effect.forEach(parsed.subscribe.split(','), (raw) => parseSubscribeTarget(raw.trim()).pipe( diff --git a/packages/mcp/server.ts b/packages/mcp/server.ts index 6d49f92..2dadb76 100644 --- a/packages/mcp/server.ts +++ b/packages/mcp/server.ts @@ -1,7 +1,13 @@ import { basename, join } from 'node:path' import { stderrLoggerLayer } from '@commy/core/logging' -import type { AcquiredIdentity, AgentComms, InboxError, MessageInbox } from '@commy/core/ports' -import { decodeChannelName, decodeThreadName } from '@commy/core/ports' +import type { + AcquiredIdentity, + AgentComms, + BindError, + InboxError, + MessageInbox, +} from '@commy/core/ports' +import { decodeChannelName, decodeThreadName, isBindError } from '@commy/core/ports' import { CommandExecutor, FetchHttpClient, FileSystem, type HttpClient } from '@effect/platform' import { NodeContext, NodeRuntime } from '@effect/platform-node' import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js' @@ -52,6 +58,7 @@ import { SessionBinderLive, SessionBinder as SessionBinderTag, } from './session-binder.ts' +import { withSessionContext } from './session-context.ts' import { SessionIdLive, SessionId as SessionIdTag } from './session-id.ts' import type { SubscribeIntent, SubscribeTokenError } from './subscribe-parser.ts' import { intentToTarget, intentToToken } from './subscribe-parser.ts' @@ -177,7 +184,7 @@ const createType2DefaultsOnAcquire = ( narrowSet: NarrowSet, inbox: MessageInbox, ): ((project: ProjectSlug | undefined) => Effect.Effect) => { - const registerIntent = (intent: SubscribeIntent): Effect.Effect => + const registerIntent = (intent: SubscribeIntent): Effect.Effect => Effect.sync(() => narrowSet.add(intent)).pipe( Effect.zipRight(inbox.subscribe(intentToTarget(intent))), ) @@ -508,10 +515,22 @@ export const makeProgram = ( // and is swallowed rather than stranding the session. Ephemeral mode only: // a persistent COMMY_BOT_NAME pane gets a new session_id every launch, so // its store is always absent → the fresh path → COMMY_SUBSCRIBE-only. + // + // Restoring re-subscribes, and a subscription is realm state under the + // seat's own principal, so this needs the seat's naming inputs in context + // for the bind to resolve. Awaiting the id here is the same wait the + // store's own `read` already performs, and it is safe for the same reason + // the fork exists: nothing downstream of a forked fiber is waiting on it. const restoreOnResume: Effect.Effect = parsed.botName !== undefined ? Effect.void - : restoreSubscriptions({ subscriptionStore, narrowSet, inbox: adapter.inbox }).pipe( + : Deferred.await(sessionIdDeferred).pipe( + Effect.flatMap((sessionId) => + withSessionContext( + restoreSubscriptions({ subscriptionStore, narrowSet, inbox: adapter.inbox }), + { sessionId, project: parsed.project }, + ), + ), Effect.catchAll((err) => Effect.logError( `commy plugin: subscription restore failed: ${Predicate.isError(err) ? err.message : String(err)}`, @@ -699,7 +718,42 @@ export const makeProgram = ( // Ephemeral mode runs a periodic idle sweep (forked into this scope). const runsIdleSweep = parsed.botName === undefined - const subscribedIntents = yield* subscribeFromEnv(adapter.inbox, narrowSet, parsed) + // Boot-time subscribe now binds: a subscription and its events queue are + // realm state under the seat's own principal, so the seam needs this + // seat's naming inputs in the fiber-local context the bind reads. + // + // POLLED, NEVER AWAITED, and the ordering is what makes that sound rather + // than lucky: the boot-env feeder completes this deferred above (the + // `readBootSessionId` step), so by the time boot reaches here the only + // zero-action source has already fired. There is no race left to lose, so + // an await would buy nothing — and would cost everything, because this + // runs on the BOOT fiber. Parking here would leave the MCP child never + // finishing boot: a hang, not a seat that is merely deaf. + const bootSessionId = yield* Deferred.poll(sessionIdDeferred).pipe( + Effect.flatMap( + Option.match({ + onNone: () => Effect.succeedNone, + onSome: (awaitId) => Effect.asSome(awaitId), + }), + ), + ) + // A seat with no way to bind cannot hold subscriptions at all — the + // ephemeral bot name is derived from the session id, so there is no name + // to mint under. Log what was lost and carry on serving: deaf is the + // accepted outcome here (the residual gap Graeme's 2026-07-05 ruling + // names), a dead MCP child is not. + const subscribedIntents = yield* withSessionContext( + subscribeFromEnv(adapter.inbox, narrowSet, parsed), + { sessionId: Option.getOrUndefined(bootSessionId), project: parsed.project }, + ).pipe( + Effect.catchIf(isBindError, (cause) => + Effect.logWarning( + `commy plugin: boot-time subscribe could not bind an identity, so no ` + + `subscriptions were applied — this seat will not receive channel traffic. ` + + `${Predicate.isError(cause) ? cause.message : String(cause)}`, + ).pipe(Effect.provide(loggerLayer), Effect.as>([])), + ), + ) // Leave a positive trace of what the boot-time subscribe set actually // resolved to. This is the diagnostic whose absence let a clobbered diff --git a/packages/mcp/subscription-restore.ts b/packages/mcp/subscription-restore.ts index 670f959..e007254 100644 --- a/packages/mcp/subscription-restore.ts +++ b/packages/mcp/subscription-restore.ts @@ -1,4 +1,4 @@ -import type { InboxError, MessageInbox } from '@commy/core/ports' +import type { BindError, InboxError, MessageInbox } from '@commy/core/ports' import type { PlatformError } from '@effect/platform/Error' import { Effect, Option, type ParseResult } from 'effect' import type { ProjectSlug } from './bootstrap.ts' @@ -37,7 +37,7 @@ export interface SubscriptionRestoreDeps { const applyRestored = ( deps: Pick, intents: ReadonlyArray, -): Effect.Effect => +): Effect.Effect => Effect.sync(() => deps.narrowSet.load(Option.some(intents))).pipe( Effect.zipRight( Effect.forEach(intents, (intent) => deps.inbox.subscribe(intentToTarget(intent)), { @@ -96,7 +96,7 @@ export const seedDefaultsIfFresh = ( */ export const restoreSubscriptions = ( deps: Pick, -): Effect.Effect => +): Effect.Effect => deps.subscriptionStore.read().pipe( Effect.flatMap( Option.match({ diff --git a/packages/zulip/adapter.test.ts b/packages/zulip/adapter.test.ts index 54ad8d0..2ddf6fd 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -2831,21 +2831,85 @@ effectTest('directory.presence runs pre-acquire and routes via minter creds', () }), ) -effectTest( - 'inbox.subscribe runs pre-acquire and routes /users/me/subscriptions via minter creds', - () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* seedSubscribeOk(stub, 'general') - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - yield* adapter.inbox.subscribe(generalChannel.name) - const subReq = yield* findRequest(stub, 'POST', '/api/v1/users/me/subscriptions') - expect(decodeBasicAuth(subReq.headers.get('Authorization'))).toEqual(minterAuth) - // The /register that arms the events queue must also be minter-creds — - // the queue belongs to the minter so lurking sessions share it. - const regReq = yield* findRequest(stub, 'POST', '/api/v1/register') - expect(decodeBasicAuth(regReq.headers.get('Authorization'))).toEqual(minterAuth) - }), +// ─── receiving runs on the seat's own principal ────────────────── + +// The credentials `buildAdapter`'s acquire binds: HERMES's delivery email +// with the key `seedRegenerate` hands back. Distinct from `minterAuth` in +// both fields, so an assertion cannot pass by accident on a shared value. +const seatAuth = { email: 'hermes-agent-bot@example.com', apiKey: 'fresh-key' } + +// Was: "inbox.subscribe runs pre-acquire and routes /users/me/subscriptions +// via minter creds". Inverted by comms-g5zh.2/.3. Receiving is state the realm +// holds on the agent's behalf (principle 5), so a subscription belongs under +// the seat's own principal and the events queue is registered against it. +// +// Both halves move together because Zulip couples them: a queue delivers a +// channel message only to users the channel's subscription rows name +// (zerver/actions/message_send.py builds the recipient set from those rows), +// so a seat-owned queue over minter-held subscriptions receives nothing. +effectTest('inbox.subscribe writes the subscription under the seat own principal', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedSubscribeOk(stub, 'general') + const adapter = yield* buildAdapter(stub) + yield* adapter.inbox.subscribe(generalChannel.name) + const subReq = yield* findRequest(stub, 'POST', '/api/v1/users/me/subscriptions') + expect(decodeBasicAuth(subReq.headers.get('Authorization'))).toEqual(seatAuth) + }), +) + +effectTest('inbox.subscribe registers the events queue under the seat own principal', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedSubscribeOk(stub, 'general') + const adapter = yield* buildAdapter(stub) + yield* adapter.inbox.subscribe(generalChannel.name) + const regReq = yield* findRequest(stub, 'POST', '/api/v1/register') + expect(decodeBasicAuth(regReq.headers.get('Authorization'))).toEqual(seatAuth) + }), +) + +effectTest('inbox.unsubscribe deletes the subscription under the seat own principal', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedSubscribeOk(stub, 'general') + yield* stub.respond('DELETE', '/api/v1/users/me/subscriptions', { + body: { result: 'success', subscribed: {}, already_subscribed: {}, unauthorized: [] }, + }) + const adapter = yield* buildAdapter(stub) + yield* adapter.inbox.subscribe(generalChannel.name) + yield* adapter.inbox.unsubscribe(generalChannel.name) + const delReq = yield* findRequest(stub, 'DELETE', '/api/v1/users/me/subscriptions') + expect(decodeBasicAuth(delReq.headers.get('Authorization'))).toEqual(seatAuth) + }), +) + +// A seat that cannot bind must not fall back to the minter for receiving, for +// the same reason `publisher.post` must not: the fallback is invisible and +// leaves the seat reading a surface that is not its own. The refusal is the +// typed BindError, surfaced rather than swallowed. +effectTest('inbox.subscribe refuses rather than falling back to minter creds', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedUsers(stub, []) + yield* seedSubscribeOk(stub, 'general') + const adapter = yield* zulipAdapter(stub, { + ...(yield* makeConfig()), + bindOnDemand: Effect.fail( + new UnboundEphemeralSession({ message: 'commy: ephemeral mode requires a session_id' }), + ), + }) + const exit = yield* Effect.exit(adapter.inbox.subscribe(generalChannel.name)) + expect(Exit.isFailure(exit)).toBe(true) + const reqs = yield* stub.captured + expect( + reqs.filter( + (r) => + r.url.pathname === '/api/v1/users/me/subscriptions' || + r.url.pathname === '/api/v1/register', + ), + ).toHaveLength(0) + }), ) // Was: "pre-acquire call dies on the 'not acquired' invariant". The invariant diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index 8f8233c..b10ac82 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -1747,10 +1747,23 @@ export const zulipAdapter = ( // registration serves every subscription state and later subscribes // reuse it. // - // The queue is registered against the minter, not the per-session - // bot — the inbox is a minter-side surface so lurking - // sessions can receive events before any acquire happens. - const ensureQueueRegistered = (): Effect.Effect => + // The queue is registered against THE SEAT, not the minter. An event queue + // is state the realm holds on this agent's behalf, so it belongs under the + // agent's own principal (docs/agent-experience.md principle 5). + // + // The seat's subscriptions move with it, and they have to: Zulip delivers a + // channel message only to the users the channel's subscription rows name + // (`zerver/actions/message_send.py` builds the recipient set from those + // rows), so a seat-owned queue over minter-held subscriptions would receive + // nothing. Queue ownership and subscription ownership are one unit. + // + // The caller passes the bound client in rather than this binding for + // itself, so the bind round-trip happens OUTSIDE the `inboxRef` lock — a + // mint held under that lock would block every concurrent subscribe on a + // network call. + const ensureQueueRegistered = ( + http: BotHttp, + ): Effect.Effect => // Atomic read-decide-register-write: the lock is held across registerQueue // so two concurrent subscribe() calls can't both read registration=None // and double-register the events queue. The snapshot is @@ -1760,77 +1773,102 @@ export const zulipAdapter = ( if (Option.isSome(state.registration)) { return Effect.succeed([undefined, state] as const) } - return registerQueue(minterHttp, config.queueIdleTimeoutSecs).pipe( + return registerQueue(http, config.queueIdleTimeoutSecs).pipe( Effect.tap((q) => config.onQueueRegister?.(q) ?? Effect.void), Effect.map((q) => [undefined, { ...state, registration: Option.some(q) }] as const), ) }) const inbox: MessageInbox = { + // Bind first, then mutate. A subscription is realm state under the seat's + // own principal, so reaching for a bound credential IS the declaration + // that the realm is about to hold state on this agent's behalf — the same + // seam the write verbs use, with no second list deciding it. + // + // A bind failure is not a subscribe failure — it is raised before any + // narrow is recorded or any call is attempted — so it flows out untouched + // rather than being flattened into an InboxError the caller cannot act + // on. Same treatment `publish` gives it. subscribe: (target) => - Effect.suspend(() => { - const channel = channelOf(target) - // Record the narrow first, snapshotting whether the channel was - // already listened to under any narrow — that decides whether the - // remote /users/me/subscriptions call is needed. - return SynchronizedRef.modify(inboxRef, (state) => { - const wasListening = streamIsListening(state, channel) - const next: InboxState = Predicate.hasProperty(target, 'kind') - ? { - ...state, - newTopicsChannels: HashSet.add(state.newTopicsChannels, channel), - } - : { - ...state, - subscribedChannels: HashSet.add(state.subscribedChannels, channel), - } - return [wasListening, next] - }).pipe( - Effect.flatMap((wasListening) => { - const subscribeRemote = wasListening - ? Effect.void - : minterHttp - .post('/users/me/subscriptions', subscriptionsResponseSchema, { - subscriptions: JSON.stringify([{ name: channel }]), - }) - .pipe(Effect.asVoid) - // /users/me/subscriptions is "me" = minter. The - // boot-time reconciler covers the universal-listener backstop; - // this per-session call still matters for streams created - // *after* the plugin booted. - return subscribeRemote.pipe(Effect.andThen(ensureQueueRegistered())) - }), - ) - }).pipe(Effect.mapError((cause) => new InboxError({ operation: 'subscribe', cause }))), + boundHttp().pipe( + Effect.flatMap((http) => + Effect.suspend(() => { + const channel = channelOf(target) + // Record the narrow first, snapshotting whether the channel was + // already listened to under any narrow — that decides whether the + // remote /users/me/subscriptions call is needed. + return SynchronizedRef.modify(inboxRef, (state) => { + const wasListening = streamIsListening(state, channel) + const next: InboxState = Predicate.hasProperty(target, 'kind') + ? { + ...state, + newTopicsChannels: HashSet.add(state.newTopicsChannels, channel), + } + : { + ...state, + subscribedChannels: HashSet.add(state.subscribedChannels, channel), + } + return [wasListening, next] + }).pipe( + Effect.flatMap((wasListening) => { + // "me" is the SEAT. Zulip's `principals` defaults to self + // (`zerver/views/streams.py` add_subscriptions_backend), so a + // bot subscribing itself is the plain unprivileged case — and + // each seat then lands in its own per-user rate-limit bucket + // instead of contending for the minter's single one. + const subscribeRemote = wasListening + ? Effect.void + : http + .post('/users/me/subscriptions', subscriptionsResponseSchema, { + subscriptions: JSON.stringify([{ name: channel }]), + }) + .pipe(Effect.asVoid) + return subscribeRemote.pipe(Effect.andThen(ensureQueueRegistered(http))) + }), + ) + }).pipe(Effect.mapError((cause) => new InboxError({ operation: 'subscribe', cause }))), + ), + ), unsubscribe: (target) => - Effect.suspend(() => { - const channel = channelOf(target) - // Drop the narrow, snapshotting whether the channel is still - // listened to afterward — if so, the minter stays subscribed. - return SynchronizedRef.modify(inboxRef, (state) => { - const next: InboxState = Predicate.hasProperty(target, 'kind') - ? { - ...state, - newTopicsChannels: HashSet.remove(state.newTopicsChannels, channel), - seenTopicsByChannel: HashMap.remove(state.seenTopicsByChannel, channel), - } - : { - ...state, - subscribedChannels: HashSet.remove(state.subscribedChannels, channel), - } - return [streamIsListening(next, channel), next] - }).pipe( - Effect.flatMap((stillListening) => - stillListening - ? Effect.void - : minterHttp - .delete('/users/me/subscriptions', subscriptionsResponseSchema, { - subscriptions: JSON.stringify([channel]), - }) - .pipe(Effect.asVoid), + boundHttp().pipe( + Effect.flatMap((http) => + Effect.suspend(() => { + const channel = channelOf(target) + // Drop the narrow, snapshotting whether the channel is still + // listened to afterward — if so, the seat stays subscribed. + return SynchronizedRef.modify(inboxRef, (state) => { + const next: InboxState = Predicate.hasProperty(target, 'kind') + ? { + ...state, + newTopicsChannels: HashSet.remove(state.newTopicsChannels, channel), + seenTopicsByChannel: HashMap.remove(state.seenTopicsByChannel, channel), + } + : { + ...state, + subscribedChannels: HashSet.remove(state.subscribedChannels, channel), + } + return [streamIsListening(next, channel), next] + }).pipe( + Effect.flatMap((stillListening) => + stillListening + ? Effect.void + : // Deletes THIS SEAT's subscription. Under the minter this + // call deafened every other seat sharing it — the + // cross-seat bug comms-g5zh.3 deletes rather than + // mitigates, since there is no longer a shared + // subscription to unwind. + http + .delete('/users/me/subscriptions', subscriptionsResponseSchema, { + subscriptions: JSON.stringify([channel]), + }) + .pipe(Effect.asVoid), + ), + ) + }).pipe( + Effect.mapError((cause) => new InboxError({ operation: 'unsubscribe', cause })), ), - ) - }).pipe(Effect.mapError((cause) => new InboxError({ operation: 'unsubscribe', cause }))), + ), + ), events: () => Stream.unwrap( Effect.all([ diff --git a/packages/zulip/events.ts b/packages/zulip/events.ts index 4537a89..0491bfb 100644 --- a/packages/zulip/events.ts +++ b/packages/zulip/events.ts @@ -506,7 +506,7 @@ export const reactionToInboundEvent = ( export const MAX_QUEUE_TIMEOUT_SECS = 604800 export const registerQueue = ( - http: ZulipHttp, + http: Pick, idleTimeoutSecs?: number, ): Effect.Effect => { // `realm` carries the realm-wide setting changes that move a consumer's From 0b91f3a45866a63be1d35a40b0d5312254c776b3 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:20:37 +0100 Subject: [PATCH 2/6] Poll and re-register the events queue through the seat that owns it Registering the queue under the seat was only half the move. The producer polls with the same client it registers with (events.ts), and the adapter was still handing it the minter's -- so a seat-owned queue would have been polled by a principal Zulip rejects outright (zerver/tornado/event_queue.py access_client_descriptor raises BadEventQueueIdError when the caller is not the queue's user). EventsConfig now separates the two clients by what they touch. The reads a batch performs -- rendered content, reaction targets, the directory -- stay on the minter, because a read leaves no realm-visible trace and belongs to whoever wrote the state. GET /events and the re-register after a dead queue go out under the queue's owner. That owner is read PASSIVELY, not through the bind seam. Asking which credential owns a queue this seat already registered is not a declaration that the realm is about to hold state -- the declaration happened at subscribe, which bound. Reading through the binding that registration established keeps polling off the mint path, which is what lets it run on the pump's own daemon fiber where no tool-call session context exists. An unbound seat owns no queue and must not mint one here, so the step idles and returns empty rather than failing or registering. Failing would enter the never-give-up retry and spin against the realm; latching would reproduce comms-9iro, where a seat that lost one early race stayed deaf for the pump's whole lifetime. An empty step keeps the unfold alive, so the next step re-reads currentRegistration and adopts a queue a later subscribe registers. The anti-latch property comes from removing the one-shot, not from adding a park. comms-g5zh.2 Claude-Session: https://claude.ai/code/session_01NdgPHEUtz1xqbFEiN3cxCB --- packages/zulip/adapter-events.test.ts | 84 +++++++++++++++++++++ packages/zulip/adapter.ts | 32 ++++++++ packages/zulip/events.ts | 102 +++++++++++++++++++++----- 3 files changed, 200 insertions(+), 18 deletions(-) diff --git a/packages/zulip/adapter-events.test.ts b/packages/zulip/adapter-events.test.ts index e4e2af0..2fefe59 100644 --- a/packages/zulip/adapter-events.test.ts +++ b/packages/zulip/adapter-events.test.ts @@ -196,6 +196,20 @@ const eventQueue = ( return queue }) +const decodeBasicAuth = (header: string | null): { email: string; apiKey: string } => { + if (header === null || !header.startsWith('Basic ')) { + throw new Error(`expected Basic auth, got ${header ?? ''}`) + } + const decoded = Buffer.from(header.slice('Basic '.length), 'base64').toString('utf-8') + const idx = decoded.indexOf(':') + if (idx < 0) throw new Error(`malformed basic auth payload: ${decoded}`) + return { email: decoded.slice(0, idx), apiKey: decoded.slice(idx + 1) } +} + +const minterAuth = { email: 'minter@example.com', apiKey: 'minter-key' } +/** The credentials `buildAdapter`'s acquire binds — distinct from the minter's in both fields. */ +const seatAuth = { email: 'hermes-agent-bot@example.com', apiKey: 'fresh-key' } + const isEventsPoll = (r: CapturedHttpRequest): boolean => r.method === 'GET' && r.url.pathname === '/api/v1/events' @@ -320,6 +334,76 @@ effectTest( { layer: TestContext.TestContext }, ) +// comms-g5zh.2. An event queue is a capability handle bound to its owner: +// Zulip's `access_client_descriptor` rejects a poll whose caller is not the +// queue's user, so a queue registered by the seat and polled by the minter +// would fail on every step. Registration and polling must name the same +// principal, and both must be the seat's — asserted together here so the pair +// cannot drift apart. +effectTest( + 'inbox.events registers and polls the queue under the seat own principal', + () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedRegister(stub) + yield* seedSubscribeOk(stub) + yield* stub.respondSequence('GET', '/api/v1/events', [ + { + body: { + result: 'success', + events: [messageEvent(5, aZulipMessage({ content: 'first' }))], + }, + }, + { hang: true }, + ]) + yield* adapter.inbox.subscribe(homeChannel.name) + const queue = yield* eventQueue(adapter) + yield* Queue.take(queue) + const registers = yield* registerPosts(stub) + const polls = yield* eventPolls(stub) + expect(registers).not.toHaveLength(0) + expect(polls).not.toHaveLength(0) + for (const req of [...registers, ...polls]) { + expect(decodeBasicAuth(req.headers.get('Authorization'))).toEqual(seatAuth) + } + }), + { layer: TestContext.TestContext }, +) + +// The reads a batch needs — rendered content, reaction targets — stay on the +// minter. A read leaves no realm-visible trace, so it needs no principal of +// its own, and routing it through the seat would spend the seat's rate-limit +// budget on state that belongs to whoever wrote it. +effectTest( + 'inbox.events resolves the directory through the minter, not the seat', + () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + const adapter = yield* buildAdapter(stub) + yield* seedRegister(stub) + yield* stub.respondSequence('GET', '/api/v1/events', [ + { + body: { + result: 'success', + events: [messageEvent(5, aZulipMessage({ content: 'first' }))], + }, + }, + { hang: true }, + ]) + const queue = yield* eventQueue(adapter) + yield* Queue.take(queue) + const userReads = (yield* stub.captured).filter( + (r) => r.method === 'GET' && r.url.pathname === '/api/v1/users', + ) + expect(userReads).not.toHaveLength(0) + for (const req of userReads) { + expect(decodeBasicAuth(req.headers.get('Authorization'))).toEqual(minterAuth) + } + }), + { layer: TestContext.TestContext }, +) + effectTest( 'eager subscribe-time register carries idle_queue_timeout and fires onQueueRegister', () => diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index b10ac82..d33d6dd 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -635,6 +635,32 @@ export const zulipAdapter = ( ), ) + // The binding as it stands, WITHOUT consulting the binder — the passive + // counterpart to `boundHttp`. Asking which credential owns the queue this + // seat already registered is not a declaration that the realm is about to + // hold state, so it does not mint and does not need a caller's session + // context. That is what lets the event pump, which runs on its own daemon + // fiber with no tool call behind it, poll the queue it owns. + // + // Refuses rather than falling back to the minter: a poll issued under the + // wrong principal would be rejected by Zulip anyway, and silently reading a + // surface that is not this seat's is the failure this whole change removes. + const ownerHttp = (): Effect.Effect => + SynchronizedRef.get(boundRef).pipe( + Effect.flatMap( + Option.match({ + onNone: (): Effect.Effect => + Effect.fail( + new UnboundEphemeralSession({ + message: + 'commy: this seat holds no identity, so it owns no events queue to poll.', + }), + ), + onSome: (b: BoundState) => Effect.succeed(b.http), + }), + ), + ) + const boundHttp = (): Effect.Effect => // The binder is consulted on EVERY write, not only when `boundRef` is // empty. A short-circuit on `boundRef` would make "is something bound?" @@ -1896,7 +1922,13 @@ export const zulipAdapter = ( return reportAbsentResume.pipe( Effect.as( inboxEvents({ + // Reads only — rendered content and reaction targets. A + // read leaves no realm-visible trace, so it stays on the + // minter and costs the seat nothing. http: minterHttp, + // The queue itself is the seat's, so polling and + // re-registering it go out under the seat's credential. + queueHttp: ownerHttp(), permalinkBase: base, resolveDirectory: buildDirectoryLookup, // Live registration read. A seat that had no queue when diff --git a/packages/zulip/events.ts b/packages/zulip/events.ts index 0491bfb..5e03d9c 100644 --- a/packages/zulip/events.ts +++ b/packages/zulip/events.ts @@ -22,6 +22,7 @@ import type { MessageRef, RealmSettings, Timestamp, + UnboundEphemeralSession, } from '@commy/core/ports' import { decodeChannelId, @@ -130,7 +131,31 @@ export const createWatermarkStore = (): Effect.Effect => ) export interface EventsConfig { + /** + * Minter-credentialled client for the READS this producer performs while + * mapping a batch — rendered content and reaction targets. A read leaves no + * realm-visible trace, so it needs no principal of its own. + */ readonly http: ZulipHttp + /** + * Client for the two calls that touch THE QUEUE ITSELF — `GET /events` and + * the re-register after a dead queue. An event queue is a capability handle + * bound to its owner: Zulip refuses a poll from anyone else + * (`zerver/tornado/event_queue.py` `access_client_descriptor` raises + * `BadEventQueueIdError` when the caller is not the queue's user), so this + * must be the same principal the queue was registered under. + * + * A PASSIVE read of the existing binding, not the bind seam. Polling a queue + * you already own is not a declaration that the realm is about to hold state + * — the declaration happened at `subscribe`, which bound. Reading through the + * binding that registration established keeps this off the mint path + * entirely, which is what lets it run on the pump's own fiber where no + * tool-call session context exists. + * + * Omit for a standalone producer with no seat behind it: {@link http} is used + * for these calls too, which is the pre-seat-ownership behaviour. + */ + readonly queueHttp?: Effect.Effect, UnboundEphemeralSession> /** * Human-facing realm origin for narrow permalinks. The adapter * resolves it once from its config (public host when a Host-header override @@ -606,6 +631,17 @@ export const defaultRetrySchedule: Schedule.Schedule<[Duration.Duration, number] */ export const RESUME_VERDICT_FALLBACK: Duration.Duration = Duration.seconds(60) +/** + * How long the producer waits before re-checking when the seat holds no + * binding, and so owns no queue to poll. + * + * This is a re-check interval, not a timeout: the wait ends in another read of + * the inbox's registration, so a seat that binds later is picked up rather than + * written off. Long enough that an unbound seat costs nothing while it waits, + * short enough that reception starts promptly once a `subscribe` binds. + */ +export const UNBOUND_IDLE_INTERVAL: Duration.Duration = Duration.seconds(5) + type EventEnvelope = { readonly id: number readonly type: string @@ -831,22 +867,32 @@ export const inboxEvents = (config: EventsConfig): Stream.Stream = ), ) + // The queue's owning principal, resolved per use rather than captured + // once: a seat that rebinds (a fresh conversation on the same MCP child) + // must not go on polling through the previous bot's credential. + const queueHttp: Effect.Effect< + Pick, + UnboundEphemeralSession + > = config.queueHttp ?? Effect.succeed(config.http) + const registerFreshQueue: Effect.Effect< EventQueueCursor, - ZulipApiError | ParseResult.ParseError - > = registerQueue(config.http, config.queueIdleTimeoutSecs).pipe( - Effect.tap((q) => config.onQueueRegister?.(q) ?? Effect.void), - // A (re-)register is the one moment a seat can silently lose its - // backlog: the new queue starts at the server's current - // last_event_id, so anything that arrived while no queue existed - // is gone. Recording it means a gap in the record has a visible - // cause rather than looking like a quiet realm. - Effect.tap((q) => - Effect.logInfo( - `commy zulip events: registered queue_id=${q.queueId} last_event_id=${q.lastEventId}`, + ZulipApiError | ParseResult.ParseError | UnboundEphemeralSession + > = queueHttp + .pipe(Effect.flatMap((http) => registerQueue(http, config.queueIdleTimeoutSecs))) + .pipe( + Effect.tap((q) => config.onQueueRegister?.(q) ?? Effect.void), + // A (re-)register is the one moment a seat can silently lose its + // backlog: the new queue starts at the server's current + // last_event_id, so anything that arrived while no queue existed + // is gone. Recording it means a gap in the record has a visible + // cause rather than looking like a quiet realm. + Effect.tap((q) => + Effect.logInfo( + `commy zulip events: registered queue_id=${q.queueId} last_event_id=${q.lastEventId}`, + ), ), - ), - ) + ) const readRegistration: Effect.Effect> = config.currentRegistration ?? Effect.succeedNone @@ -866,7 +912,10 @@ export const inboxEvents = (config: EventsConfig): Stream.Stream = */ const handleBadQueue = ( state: ProducerState, - ): Effect.Effect => + ): Effect.Effect< + StepResult, + ZulipApiError | ParseResult.ParseError | UnboundEphemeralSession + > => Effect.gen(function* () { yield* reportResume(false) const since = yield* watermark.get() @@ -928,10 +977,14 @@ export const inboxEvents = (config: EventsConfig): Stream.Stream = ), }) const currentQueue: EventQueueCursor = observed.queue ?? (yield* registerFreshQueue) - const res = yield* config.http.get('/events', eventsResponseSchema, { - queue_id: currentQueue.queueId, - last_event_id: currentQueue.lastEventId, - }) + const res = yield* queueHttp.pipe( + Effect.flatMap((http) => + http.get('/events', eventsResponseSchema, { + queue_id: currentQueue.queueId, + last_event_id: currentQueue.lastEventId, + }), + ), + ) // The poll returned, so the queue this step polled is live. On the // first poll that is the resume verdict: the surviving queue is // replaying the backlog — report it so history catch-up stands down. @@ -1005,6 +1058,19 @@ export const inboxEvents = (config: EventsConfig): Stream.Stream = } return Effect.fail(e) }), + // The seat holds no binding, so it owns no queue to poll and must not + // mint one here — registering belongs to `subscribe`, which binds. + // + // Idle rather than fail, and idle rather than latch. Failing would + // enter the never-give-up retry and hot-spin against the realm; + // latching would reproduce comms-9iro, where a seat that lost one + // early race stayed deaf for the pump's whole lifetime with no + // recovery. Returning an empty step keeps the unfold alive so the + // NEXT step re-reads `currentRegistration` — a subscribe that binds + // later registers a queue and this producer adopts it. + Effect.catchTag('UnboundEphemeralSession', () => + Effect.sleep(UNBOUND_IDLE_INTERVAL).pipe(Effect.as([[], state] as StepResult)), + ), ) const stepWithRetry = ( From bb46ce7a80a90cc7fb1d9156a9cc904d274a6c0a Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:23:55 +0100 Subject: [PATCH 3/6] Pin the cross-seat unsubscribe and the absent latch Two regression tests for the bugs this rework deletes rather than fixes. comms-g5zh.3's bug: under the shared minter there was ONE subscription row for the whole fleet, so seat B's unsubscribe issued a DELETE against the row seat A was receiving through, and deafened A. Nothing refcounted it -- streamIsListening counts only within one adapter instance across narrow kinds, and inboxRef is per-process -- so there was no cross-seat unwinding to get wrong. With each seat holding its own row the bug has no shape to take. The test pins the structural fact that makes that true: every subscription write in a two-seat exchange goes out under the seat that issued it, and the minter issues none. comms-9iro's bug: the producer consulted the session once at materialisation, got nothing, and latched, leaving a seat that lost that race deaf for the pump's whole lifetime. The replacement property is not an await -- it is the absence of a latch. The test starts the pump on an unbound seat, binds afterwards, and requires the event to arrive. Mutation-checked: replacing the idle step with a park makes it hang and fail, so it genuinely observes the latch rather than the happy path. The behavioural half of the cross-seat proof needs two real principals in a realm and belongs in the live suite; this is the structural half. comms-g5zh.2 comms-g5zh.3 Claude-Session: https://claude.ai/code/session_01NdgPHEUtz1xqbFEiN3cxCB --- packages/zulip/adapter-events.test.ts | 54 ++++++++++++++++++++++++ packages/zulip/adapter.test.ts | 59 +++++++++++++++++++++++++++ packages/zulip/events.ts | 27 ++++++------ 3 files changed, 126 insertions(+), 14 deletions(-) diff --git a/packages/zulip/adapter-events.test.ts b/packages/zulip/adapter-events.test.ts index 2fefe59..e08e3c2 100644 --- a/packages/zulip/adapter-events.test.ts +++ b/packages/zulip/adapter-events.test.ts @@ -371,6 +371,60 @@ effectTest( { layer: TestContext.TestContext }, ) +// comms-9iro, dissolved rather than mitigated. That bug was a ONE-SHOT: the +// producer consulted the session once at materialisation, got nothing, and +// latched — a seat that lost that race stayed deaf for the pump's entire +// lifetime, with no retry and no way to notice from inside. +// +// The property that replaces it is not an await. It is the absence of a latch: +// a producer that starts unbound keeps its unfold alive and re-reads the +// inbox's registration each step, so a seat that binds LATER is picked up. This +// test starts the pump on an unbound seat, binds afterwards, and requires the +// event to arrive. It fails if any step in that path gives up permanently. +effectTest( + 'a producer that starts unbound adopts the queue a later subscribe registers', + () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedUsers(stub, [HERMES]) + yield* seedRegenerate(stub, HERMES.user_id) + yield* seedRegister(stub) + yield* seedSubscribeOk(stub) + yield* stub.respondSequence('GET', '/api/v1/events', [ + { + body: { + result: 'success', + events: [messageEvent(5, aZulipMessage({ content: 'after binding' }))], + }, + }, + { hang: true }, + ]) + // Deliberately NOT acquired: the producer materialises against a seat + // that owns no queue and cannot register one. + const adapter = yield* zulipAdapter({ + realmUrl: yield* RealmUrl(REALM_URL).pipe(Effect.orDie), + minterEmail: yield* BotEmail('minter@example.com').pipe(Effect.orDie), + minterApiKey: Redacted.make(yield* ApiKey('minter-key').pipe(Effect.orDie)), + }).pipe(Effect.provideService(HttpClient.HttpClient, stub.client)) + const queue = yield* eventQueue(adapter) + + // Let the unbound producer idle through several re-checks. Under a latch + // these are the steps during which it would have given up for good. + yield* TestClock.adjust(Duration.seconds(30)) + expect(yield* eventPolls(stub)).toHaveLength(0) + expect(yield* registerPosts(stub)).toHaveLength(0) + + yield* adapter.identity.acquire(decodeBotNameSync('hermes-agent')) + yield* adapter.inbox.subscribe(homeChannel.name) + yield* TestClock.adjust(Duration.seconds(30)) + + const event = yield* Queue.take(queue) + expect(event.kind).toBe('message-posted') + expect(yield* eventPolls(stub)).not.toHaveLength(0) + }), + { layer: TestContext.TestContext }, +) + // The reads a batch needs — rendered content, reaction targets — stay on the // minter. A read leaves no realm-visible trace, so it needs no principal of // its own, and routing it through the seat would spend the seat's rate-limit diff --git a/packages/zulip/adapter.test.ts b/packages/zulip/adapter.test.ts index 2ddf6fd..74a4aa5 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -2884,6 +2884,65 @@ effectTest('inbox.unsubscribe deletes the subscription under the seat own princi }), ) +// comms-g5zh.3's acceptance, and the bug it DELETES rather than mitigates. +// +// Under the shared minter there was one subscription row for the whole fleet, +// so seat B's unsubscribe issued a DELETE against the row seat A was receiving +// through — and deafened A. Nothing refcounted it: `streamIsListening` counts +// only within one adapter instance across narrow kinds, and `inboxRef` is +// per-process, so no cross-seat unwinding existed to get wrong. +// +// With each seat holding its own row the bug has no shape to take: B's DELETE +// names B's principal, and A's row is not reachable from it. This asserts the +// structural fact that makes that true — every write in the exchange goes out +// under the seat that issued it, and the minter issues none. The behavioural +// half (A keeps receiving) needs two real principals in a realm and lives in +// the live suite. +effectTest('one seat unsubscribing writes only under its own principal, never a shared one', () => + Effect.gen(function* () { + const stub = yield* makeStubHttpClient + yield* seedUsers(stub, [HERMES, RIQ]) + yield* seedRegenerate(stub, HERMES.user_id, 'hermes-key') + yield* seedRegenerate(stub, RIQ.user_id, 'riq-key') + yield* seedRegisterOk(stub) + yield* stub.respond('POST', '/api/v1/users/me/subscriptions', { + body: { result: 'success', subscribed: {}, already_subscribed: {}, unauthorized: [] }, + }) + yield* stub.respond('DELETE', '/api/v1/users/me/subscriptions', { + body: { result: 'success', subscribed: {}, already_subscribed: {}, unauthorized: [] }, + }) + + const config = yield* makeConfig() + const seatA = yield* zulipAdapter(stub, config) + yield* seatA.identity.acquire(decodeBotNameSync('hermes-agent')) + const seatB = yield* zulipAdapter(stub, config) + yield* seatB.identity.acquire(decodeBotNameSync('riq6r230')) + + const authA = { email: 'hermes-agent-bot@example.com', apiKey: 'hermes-key' } + const authB = { email: 'riq-bot@example.com', apiKey: 'riq-key' } + + yield* seatA.inbox.subscribe(generalChannel.name) + yield* seatB.inbox.subscribe(generalChannel.name) + yield* seatB.inbox.unsubscribe(generalChannel.name) + + const subscriptionWrites = (yield* stub.captured).filter( + (r) => r.url.pathname === '/api/v1/users/me/subscriptions', + ) + const byAuth = subscriptionWrites.map((r) => ({ + method: r.method, + auth: decodeBasicAuth(r.headers.get('Authorization')), + })) + expect(byAuth).toEqual([ + { method: 'POST', auth: authA }, + { method: 'POST', auth: authB }, + { method: 'DELETE', auth: authB }, + ]) + // The load-bearing negative: no subscription write in the exchange went out + // as the minter. A single minter-issued DELETE here is the whole bug. + expect(byAuth.filter((w) => w.auth.email === minterAuth.email)).toEqual([]) + }), +) + // A seat that cannot bind must not fall back to the minter for receiving, for // the same reason `publisher.post` must not: the fallback is invisible and // leaves the seat reading a surface that is not its own. The refusal is the diff --git a/packages/zulip/events.ts b/packages/zulip/events.ts index 5e03d9c..d1ab703 100644 --- a/packages/zulip/events.ts +++ b/packages/zulip/events.ts @@ -878,21 +878,20 @@ export const inboxEvents = (config: EventsConfig): Stream.Stream = const registerFreshQueue: Effect.Effect< EventQueueCursor, ZulipApiError | ParseResult.ParseError | UnboundEphemeralSession - > = queueHttp - .pipe(Effect.flatMap((http) => registerQueue(http, config.queueIdleTimeoutSecs))) - .pipe( - Effect.tap((q) => config.onQueueRegister?.(q) ?? Effect.void), - // A (re-)register is the one moment a seat can silently lose its - // backlog: the new queue starts at the server's current - // last_event_id, so anything that arrived while no queue existed - // is gone. Recording it means a gap in the record has a visible - // cause rather than looking like a quiet realm. - Effect.tap((q) => - Effect.logInfo( - `commy zulip events: registered queue_id=${q.queueId} last_event_id=${q.lastEventId}`, - ), + > = queueHttp.pipe( + Effect.flatMap((http) => registerQueue(http, config.queueIdleTimeoutSecs)), + Effect.tap((q) => config.onQueueRegister?.(q) ?? Effect.void), + // A (re-)register is the one moment a seat can silently lose its + // backlog: the new queue starts at the server's current + // last_event_id, so anything that arrived while no queue existed + // is gone. Recording it means a gap in the record has a visible + // cause rather than looking like a quiet realm. + Effect.tap((q) => + Effect.logInfo( + `commy zulip events: registered queue_id=${q.queueId} last_event_id=${q.lastEventId}`, ), - ) + ), + ) const readRegistration: Effect.Effect> = config.currentRegistration ?? Effect.succeedNone From f3f5253d69aead9db884b942e37a694ddd9c8e81 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:41:33 +0100 Subject: [PATCH 4/6] Verify per-seat receiving on a real realm, and correct the record The two facts this rework turns on are facts about Zulip, not about our code: an event queue is a capability bound to its owner and refuses anyone else, and a channel message is delivered only to principals the channel's subscription rows name. A stub is blind to both -- it will hand the minter a seat's queue, and deliver to a queue whose owner subscribes to nothing. That is the shape of failure comms-hsym recorded: a suite green against a configuration nobody deploys. Three live tests, gated on the existing live-channel env var: Ownership is proved by REFUSAL, not by assertion. Registering under the seat only means something if the minter genuinely cannot use the queue, so the discriminating check is that the minter's poll is rejected with BAD_EVENT_QUEUE_ID while the seat's poll of the same queue succeeds. A queue still registered under the minter passes the second half and fails the first. Delivery is the consequence that makes the move worth anything: a channel message must reach the seat's own queue, which a deaf seat cannot fake. The cross-seat test is comms-g5zh.3's acceptance with two real principals, the only way to show the deafening is gone. Docs corrected where they now describe a fixed bug as live. agent-experience.md's worked example stated the cross-seat deafening in the present tense; it is now recorded as deleted rather than fixed, with the distinction spelled out -- nothing counts seats today either, and nothing needs to, because there is no shared subscription left to unwind. What has NOT moved (the minter's blanket subscription, client-side narrowing) is named so the correction cannot read as more than it is. self-hosting.md gains the operator-facing behaviour change as one entry for the host class it affects: an ephemeral non-CC host that injects no session id can no longer receive, because it cannot name an identity to mint. It was previously carried by the minter's queue; retiring that crutch makes the gap visible rather than masked. Both supported ways out are named. Claude Code seats are unaffected. comms-g5zh.2 comms-g5zh.3 Claude-Session: https://claude.ai/code/session_01NdgPHEUtz1xqbFEiN3cxCB --- docs/agent-experience.md | 44 ++++--- docs/self-hosting.md | 28 +++++ packages/zulip/realm.live.test.ts | 198 +++++++++++++++++++++++++++++- 3 files changed, 254 insertions(+), 16 deletions(-) diff --git a/docs/agent-experience.md b/docs/agent-experience.md index 5c939fb..5dfc9ca 100644 --- a/docs/agent-experience.md +++ b/docs/agent-experience.md @@ -140,13 +140,18 @@ Places the current implementation fails this reference. How a small optimisation becomes a large architecture. -An ephemeral session does not mint a bot until its first attribution- -producing call, so that a session which never uses commy costs the realm +An ephemeral session did not mint a bot until its first attribution- +producing call, so that a session which never used commy cost the realm nothing. But a session that has not yet minted still needs to receive — so -something must listen on its behalf. That something is the minter, +something must listen on its behalf. That something was the minter, subscribed to every public stream, with the event queue registered against it rather than the per-session bot. +The example is told in the past tense because the first two steps of the +chain have since been taken apart: the queue and the subscriptions now +belong to the seat. The rest of the chain is still standing, and the +paragraphs below say which parts. + Everything else follows from that one deferral. One shared subscriber means per-agent narrowing cannot be a realm subscription, so it becomes a client-side filter. A client-side filter over a shared account needs @@ -156,17 +161,28 @@ a persistent store. That store has no realm principal to key on, so it keys on `session_id`. Each step is locally reasonable; the sum is not. The refcounting step is the one the architecture never took, and its absence -is live today. `streamIsListening` (`packages/zulip/adapter.ts:506-508`) does -refcount — but over the narrow *kinds* one seat holds on a channel -(`channel:X` against `new-topics:X`), within a single `InboxState`. That -state lives behind an `inboxRef` constructed inside the adapter -(`adapter.ts:1576`), so its scope is one adapter instance: one process, one -seat. Nothing counts seats. So `unsubscribe` reaches "nobody is listening" -on the strength of one seat's own narrows and issues -`DELETE /users/me/subscriptions` (`adapter.ts:1712-1730`) — where "me" is -the shared minter. One agent unsubscribing from a channel deafens every -other agent on it, until some unrelated seat's boot reconciler happens to -resubscribe. +was live. `streamIsListening` does refcount — but over the narrow *kinds* one +seat holds on a channel (`channel:X` against `new-topics:X`), within a single +`InboxState`. That state lives behind an `inboxRef` constructed inside the +adapter, so its scope is one adapter instance: one process, one seat. Nothing +counted seats. So `unsubscribe` reached "nobody is listening" on the strength +of one seat's own narrows and issued `DELETE /users/me/subscriptions` — where +"me" was the shared minter. One agent unsubscribing from a channel deafened +every other agent on it, until some unrelated seat's boot reconciler happened +to resubscribe. + +That bug is now **deleted rather than fixed**, and the distinction is the +point: nothing counts seats today either. The queue and the subscriptions moved +to the seat's own principal, so "me" is the seat, and one seat's `DELETE` +cannot name another's row. The refcounting the architecture never built is not +owed — there is no shared subscription left to unwind. Queue and subscription +had to move together, because Zulip builds a channel message's recipient set +from the channel's subscription rows: a seat-owned queue over minter-held +subscriptions would have received nothing at all. + +What has not moved yet: the minter still holds its blanket public-stream +subscription, and narrowing is still a client-side filter. Those are the next +steps of the same unwinding, not exemptions. Principle 5 catches it at the first step. Principle 3 catches the store. Principle 1 catches `session_id` reaching the tool surface. diff --git a/docs/self-hosting.md b/docs/self-hosting.md index 9414dff..3254ff9 100644 --- a/docs/self-hosting.md +++ b/docs/self-hosting.md @@ -191,6 +191,34 @@ For a container, pin the published package version at image build time — e.g. command `commy-mcp` (or `npx @codeforbreakfast/commy-mcp`), so boot resolves the already-present bundle and never reaches the network. +## An ephemeral non-CC host must supply a session id to receive + +**Behaviour change.** Receiving is state the realm holds on the agent's behalf, +so a seat's event queue and its subscriptions now live under **its own** +principal rather than the shared minter's. A seat therefore has to be able to +obtain an identity before it can subscribe or receive at all — and an ephemeral +identity is named from the session id (`cc-[-]`), so no +session id means no name to mint under, and no reception. + +This affects one host class: an **ephemeral** host (no `COMMY_BOT_NAME`) that +injects no session id. Such a seat previously received channel traffic through +the minter's queue, since the minter was subscribed to every public stream and +listened on behalf of un-minted seats. That crutch is being retired, so the +seat's inability to identify itself is now visible instead of masked: its +boot-time subscribe is refused, it registers no queue, and it logs a warning +naming what it lost. It keeps serving; it does not receive. + +Two supported ways to avoid it, both already documented above: + +- Set `COMMY_BOT_NAME` for a persistent identity — the session id is irrelevant + in that mode, and the bot subscribes under its own stable principal. +- Pass a UUID `session_id` in the tool-call arguments, which is the binding an + ephemeral non-CC host has (`docs/claude-channel-inbound-contract.md`). + +Claude Code seats are unaffected: the plugin injects `CLAUDE_CODE_SESSION_ID` +into the MCP child's environment at spawn, so the id is known before the seat's +first subscribe. + ## Inbound is host work A standalone MCP client on the open pipe physically receives inbound events (each diff --git a/packages/zulip/realm.live.test.ts b/packages/zulip/realm.live.test.ts index 8bf126d..94395dc 100644 --- a/packages/zulip/realm.live.test.ts +++ b/packages/zulip/realm.live.test.ts @@ -37,7 +37,13 @@ */ import { describe, expect, test } from 'bun:test' -import type { BotName, Credentials, DisplayName, ReleaseOpts } from '@commy/core/ports' +import type { + BotName, + Credentials, + DisplayName, + EventQueueCursor, + ReleaseOpts, +} from '@commy/core/ports' import { decodeBotNameSync, decodeChannelNameSync, @@ -46,7 +52,7 @@ import { decodeThreadNameSync, } from '@commy/core/ports' import { FetchHttpClient, HttpClient, HttpClientRequest } from '@effect/platform' -import { Duration, Effect, Encoding, Option, Redacted, Schema } from 'effect' +import { Duration, Effect, Encoding, Option, Redacted, Schema, Stream } from 'effect' import type { ZulipAdapter } from './adapter.ts' import { zulipAdapter } from './adapter.ts' import { ApiKey, BotEmail, makeZulipHttp, RealmUrl, ZulipApiError, type ZulipHttp } from './http.ts' @@ -145,6 +151,15 @@ const sendMessageSchema = Schema.Struct({ id: Schema.Int, }) +/** + * Just enough of GET /events to tell "the realm accepted this poll" from "the + * realm refused it". The events themselves are irrelevant here — the assertion + * is about WHO may poll the queue, so the payload stays unmodelled. + */ +const anyEventsSchema = Schema.Struct({ + result: Schema.Literal('success'), +}) + /** Just enough of GET /messages to prove a topic is empty. */ const messagesInTopicSchema = Schema.Struct({ result: Schema.Literal('success'), @@ -491,6 +506,185 @@ describeLiveChannel('zulip live resolve-then-post — zulip.example.com', () => ) }) +/** + * Per-seat receiving (comms-g5zh.2 / .3), on a real realm because nothing else + * can settle it. + * + * A stub answers whatever it is told to. The two facts this rework turns on are + * facts about ZULIP, not about our code: that an event queue is a capability + * bound to its owner and refuses anyone else, and that a channel message is + * delivered only to principals the channel's subscription rows name. Both are + * invisible to a fake — a stub will happily hand the minter a seat's queue, and + * will happily deliver to a queue whose owner subscribes to nothing. That is + * exactly the shape of failure this suite exists to catch: 1182 green tests + * against a configuration nobody deploys. + */ +describeLiveChannel('zulip live per-seat receiving — zulip.example.com', () => { + // Ownership, proved by REFUSAL rather than by assertion. Registering under + // the seat is only meaningful if the minter genuinely cannot use the queue — + // so the discriminating check is that the minter's poll is REJECTED while the + // seat's succeeds. A queue still registered under the minter passes the + // second half and fails the first. + test( + 'the queue a subscribe registers belongs to the seat: the minter cannot poll it', + () => + Effect.runPromise( + Effect.gen(function* () { + const e = liveEnv() + const channel = decodeChannelNameSync(liveChannelName ?? '') + const registered: EventQueueCursor[] = [] + const adapter = yield* Effect.provideService( + zulipAdapter({ + realmUrl: yield* RealmUrl(e.site), + minterEmail: yield* BotEmail(e.minterEmail), + minterApiKey: Redacted.make(yield* ApiKey(e.minterApiKey)), + onQueueRegister: (q) => Effect.sync(() => void registered.push(q)), + }).pipe(Effect.orDie), + HttpClient.HttpClient, + httpClient, + ) + yield* Effect.acquireUseRelease( + pacedAcquire(adapter, decodeBotNameSync(uniqueName('queue-owner'))), + (acquired) => + Effect.gen(function* () { + yield* adapter.inbox.subscribe(channel) + const queue = registered[0] + expect(queue).toBeDefined() + if (queue === undefined) return + + // The minter is a real, live, privileged principal on this + // realm — it just is not this queue's owner. Zulip answers + // BAD_EVENT_QUEUE_ID for a queue belonging to another user + // (zerver/tornado/event_queue.py access_client_descriptor). + const asMinter = yield* minterHttp(e) + const refusal = yield* Effect.flip( + asMinter.get('/events', anyEventsSchema, { + queue_id: queue.queueId, + last_event_id: queue.lastEventId, + }), + ) + expect(refusal).toBeInstanceOf(ZulipApiError) + expect((refusal as ZulipApiError).code).toBe('BAD_EVENT_QUEUE_ID') + + // ...and the seat's own credential is accepted for the same + // queue, so the refusal above is about WHO asked, not about the + // queue being dead. + const asSeat = yield* botHttp(e, credentialsOf(acquired.credentials)) + const ok = yield* asSeat.get('/events', anyEventsSchema, { + queue_id: queue.queueId, + last_event_id: queue.lastEventId, + }) + expect(ok.result).toBe('success') + }), + () => pacedRelease(adapter), + ) + }), + ), + 45_000, + ) + + // The consequence that makes the queue move worth anything. A seat-owned + // queue over minter-held subscriptions receives NOTHING from channels — the + // recipient set for a channel message is built from the channel's + // subscription rows — so this is the assertion that a deaf seat cannot pass. + test( + 'a channel message reaches the seat own queue, not just its DMs', + () => + Effect.runPromise( + Effect.gen(function* () { + const e = liveEnv() + const channel = decodeChannelNameSync(liveChannelName ?? '') + const thread = decodeThreadNameSync( + `cc-live-perseat-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`, + ) + const body = `per-seat delivery probe ${Math.random().toString(36).slice(2, 10)}` + const adapter = yield* buildAdapter() + yield* Effect.acquireUseRelease( + pacedAcquire(adapter, decodeBotNameSync(uniqueName('receiver'))), + () => + Effect.gen(function* () { + // subscribe resolving is the readiness contract: the queue + // exists before we post, so the message cannot race ahead of it. + yield* adapter.inbox.subscribe(channel) + const asMinter = yield* minterHttp(e) + yield* asMinter.post('/messages', sendMessageSchema, { + type: 'channel', + to: channel, + topic: thread, + content: body, + }) + const seen = yield* adapter.inbox.events().pipe( + Stream.filter( + (event) => event.kind === 'message-posted' && event.message.body === body, + ), + Stream.runHead, + Effect.timeout(Duration.seconds(25)), + ) + expect(Option.isSome(seen)).toBe(true) + }), + () => pacedRelease(adapter), + ) + }), + ), + 45_000, + ) + + // comms-g5zh.3's acceptance, behaviourally. Under the shared minter this was + // a live bug: one subscription row served every seat, so B's unsubscribe + // DELETEd the row A was receiving through and A went silently deaf. Two real + // principals are the only way to show it is gone — with a row each, B's + // unsubscribe cannot reach A's. + test( + 'one seat unsubscribing does not deafen another seat on the same channel', + () => + Effect.runPromise( + Effect.gen(function* () { + const e = liveEnv() + const channel = decodeChannelNameSync(liveChannelName ?? '') + const thread = decodeThreadNameSync( + `cc-live-twoseat-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`, + ) + const body = `two-seat probe ${Math.random().toString(36).slice(2, 10)}` + const seatA = yield* buildAdapter() + const seatB = yield* buildAdapter() + yield* Effect.acquireUseRelease( + pacedAcquire(seatA, decodeBotNameSync(uniqueName('stayer'))), + () => + Effect.acquireUseRelease( + pacedAcquire(seatB, decodeBotNameSync(uniqueName('leaver'))), + () => + Effect.gen(function* () { + yield* seatA.inbox.subscribe(channel) + yield* seatB.inbox.subscribe(channel) + // The act that used to deafen A. + yield* seatB.inbox.unsubscribe(channel) + + const asMinter = yield* minterHttp(e) + yield* asMinter.post('/messages', sendMessageSchema, { + type: 'channel', + to: channel, + topic: thread, + content: body, + }) + const seen = yield* seatA.inbox.events().pipe( + Stream.filter( + (event) => event.kind === 'message-posted' && event.message.body === body, + ), + Stream.runHead, + Effect.timeout(Duration.seconds(25)), + ) + expect(Option.isSome(seen)).toBe(true) + }), + () => pacedRelease(seatB), + ), + () => pacedRelease(seatA), + ) + }), + ), + 60_000, + ) +}) + describeLive('zulip live upload round-trip — zulip.example.com', () => { // The unit tests prove uploadRaw shapes a multipart request; only a real // round-trip proves Django actually accepts that body and the bytes survive. From c9de952ae4e65efd5f2c84578696148d619bd42a Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 08:55:42 +0100 Subject: [PATCH 5/6] Stamp subscribe and unsubscribe, and make the fakes honour the bind seam RATIFIED REVERSAL OF 0f0e755 (PR #126). That commit chose id-blind subscribe and explicitly declined to add subscribe/unsubscribe to the PreToolUse matcher, on the grounds that "subscribe/unsubscribe need not carry session_id". That was correct for the architecture it served: a subscription was written under the SHARED MINTER, so the seat needed no identity of its own to declare interest. comms-g5zh.3 retires that architecture. A subscription is realm state under the seat's own principal, so subscribe MINTS -- and a tool whose adapter path reaches the bind seam must carry a session id or be refused. The premise #126 rested on is gone, so the matcher widens to the seven tools that declare session_id. Graeme ratified this reversal on 2026-07-31; PR #191 attempted the same change in July and was closed because it did not cite #126. Naming it is the point: this is a authorized reversal of a considered decision, not a fourth uncited pass at the same file. THE REASON THIS WAS INVISIBLE. The memory adapter's inbox.subscribe did not bind, so every rig that boots the server against it exercised a seam that differs from the one that ships -- the exact failure comms-hsym recorded. Making the fake honour the seam turned a green suite red and surfaced three real defects that would otherwise have shipped: 1. A REENTRANT BIND DEADLOCK, which hung the first action of every seat with a project slug. Post binds -> acquire -> onAcquire seeds subscriptions -> inbox.subscribe binds -> re-enters the same single-flight while its state still says `pending` -> awaits the Deferred its own caller must complete. Fixed by recording the binding BEFORE running post-acquire work, which is also the honest model: the identity exists the moment acquire returns it, and what follows is work done AS that identity. The documented failure contract is kept by hand, via onError rather than tapError so a hook that THROWS still drops the binding. 2. STICKY ENGAGEMENT ESCAPED THE SESSION CONTEXT. It ran its subscribe at the bare runtime edge rather than through the caller's run, so the bind saw no session id and refused a caller who had done nothing wrong -- they posted; the subscribe is our inference from that post. 3. The deployed-wiring rig reported subscribe/unsubscribe as refused-for-want-of-identity, which is precisely the rig doing its job. Two guards re-aimed rather than relaxed. hooks.test.ts used `subscribe` as its example of a tool the matcher must NOT capture; the guarantee now stands on `read_thread`, which genuinely never binds. Its per-tool list was also missing edit_message, so it is restated against all seven. comms-g5zh.2 comms-g5zh.3 Claude-Session: https://claude.ai/code/session_01NdgPHEUtz1xqbFEiN3cxCB --- clients/claude-code/hooks-manifest.test.ts | 32 ++++++++------- clients/claude-code/hooks/hooks.json | 2 +- clients/claude-code/hooks/hooks.test.ts | 21 +++++++++- packages/mcp/ensure-bound.ts | 37 +++++++++++++++++ packages/mcp/identity-cache.ts | 40 ++++++++---------- packages/mcp/subscription-resume.test.ts | 48 ++++++++++++++++------ packages/mcp/tools.ts | 12 +++++- packages/memory/adapter.ts | 21 +++++++++- 8 files changed, 157 insertions(+), 56 deletions(-) diff --git a/clients/claude-code/hooks-manifest.test.ts b/clients/claude-code/hooks-manifest.test.ts index 1ab0fe6..92ccdbd 100644 --- a/clients/claude-code/hooks-manifest.test.ts +++ b/clients/claude-code/hooks-manifest.test.ts @@ -76,20 +76,20 @@ const BOUND_INBOX_VERBS = ['subscribe', 'unsubscribe'] as const * Tools that reach `boundHttp` through an inbox verb while sitting outside the * matcher, so the hook never stamps them and the bind seam sees no session id. * - * These are NOT a `comms-tww6`-style attribution accident — they cannot inherit - * an earlier call's seat, because `boundHttp` consults the binder on every call - * and refuses outright when the context carries no session id - * (`adapter.ts`, the "may THIS CALLER use it?" comment). An unstamped - * `subscribe` therefore FAILS rather than binding to the wrong seat. + * EMPTY, and it has to stay that way. An unstamped inbox verb is not a + * `comms-tww6`-style attribution accident — it cannot inherit an earlier + * call's seat, because `boundHttp` consults the binder on every call and + * refuses outright when the context carries no session id. It simply FAILS. * - * Emptying this list means widening the `hooks.json` matcher to the seven tools - * that declare `session_id`. That change reverses commit `0f0e755` (PR #126), - * which deliberately declined it while the shared minter still owned - * subscriptions — so it is a ratification call, not a lint fix, and is held - * pending that ruling. The list exists so the gap is machine-visible in the - * meantime rather than resting on prose. + * The matcher was widened to the seven tools that declare `session_id`, which + * REVERSES commit `0f0e755` (PR #126) — that commit chose id-blind subscribe + * and explicitly declined to add these two. The reversal is deliberate and + * ratified: `#126`'s choice served the shared-minter architecture, where + * subscribe wrote under the minter and needed no identity of its own. + * `comms-g5zh.3` retires that architecture — subscribe now mints — so the + * premise `#126` rested on is gone. */ -const G5ZH3_MATCHER_PENDING = ['subscribe', 'unsubscribe'] as const +const G5ZH3_MATCHER_PENDING = [] as const /** * Tools that reach `boundHttp` while declaring no `session_id` and sitting @@ -267,9 +267,13 @@ test('the matcher carries no tool that never reaches boundHttp and never binds', .filter((name) => name !== 'current_identity') .filter((name) => { const f = facts.get(name) + if (f === undefined) return true + // Either receiver counts. A tool binds through the publisher verbs or + // through the inbox verbs; asking only about the first would call a + // legitimately-stamped `subscribe` an orphan. return ( - f === undefined || - ![...f.verbs].some((v) => (BOUND_VERBS as ReadonlyArray).includes(v)) + ![...f.verbs].some((v) => (BOUND_VERBS as ReadonlyArray).includes(v)) && + ![...f.inboxVerbs].some((v) => (BOUND_INBOX_VERBS as ReadonlyArray).includes(v)) ) }) .sort() diff --git a/clients/claude-code/hooks/hooks.json b/clients/claude-code/hooks/hooks.json index 520c65a..21e9ec3 100644 --- a/clients/claude-code/hooks/hooks.json +++ b/clients/claude-code/hooks/hooks.json @@ -2,7 +2,7 @@ "hooks": { "PreToolUse": [ { - "matcher": "mcp__plugin_commy_commy__(post|edit_message|react|unreact|current_identity)", + "matcher": "mcp__plugin_commy_commy__(post|edit_message|react|unreact|current_identity|subscribe|unsubscribe)", "hooks": [ { "type": "command", diff --git a/clients/claude-code/hooks/hooks.test.ts b/clients/claude-code/hooks/hooks.test.ts index 09cfc65..0f1c64c 100644 --- a/clients/claude-code/hooks/hooks.test.ts +++ b/clients/claude-code/hooks/hooks.test.ts @@ -4,7 +4,20 @@ import hooks from './hooks.json' const PLUGIN_SLUG = 'commy' const EXPECTED_PREFIX = `mcp__plugin_${PLUGIN_SLUG}_${PLUGIN_SLUG}__` -const ATTRIBUTION_TOOLS = [['post'], ['react'], ['unreact'], ['current_identity']] as const +// Every tool the matcher must stamp. `edit_message` was missing here for as +// long as the list was hand-kept; `subscribe` / `unsubscribe` join it because +// declaring interest now mints (comms-g5zh.3). The set-level invariant — +// matcher == the tools whose schema declares session_id — is asserted in +// `hooks-manifest.test.ts`; this list is the per-name spelling check. +const ATTRIBUTION_TOOLS = [ + ['post'], + ['edit_message'], + ['react'], + ['unreact'], + ['current_identity'], + ['subscribe'], + ['unsubscribe'], +] as const interface HookEntry { readonly matcher: string @@ -52,7 +65,11 @@ test('matcher does not match arbitrary other MCP tools (no over-broad capture)', const matcher = preToolUse[0]?.matcher ?? '' const re = new RegExp(`^${matcher}$`) expect(re.test('mcp__plugin_discord_discord__reply')).toBe(false) - expect(re.test(`${EXPECTED_PREFIX}subscribe`)).toBe(false) + // `subscribe` used to stand here as the example of a tool the matcher must + // NOT capture. It is now stamped deliberately (comms-g5zh.3 — subscribe + // mints), so the guarantee is restated against a tool that genuinely never + // binds: `read_thread` is a read, and a read never mints. + expect(re.test(`${EXPECTED_PREFIX}read_thread`)).toBe(false) expect(re.test(`${EXPECTED_PREFIX}list_agents`)).toBe(false) expect(re.test(`${EXPECTED_PREFIX}list_channels`)).toBe(false) }) diff --git a/packages/mcp/ensure-bound.ts b/packages/mcp/ensure-bound.ts index 4a75d09..86d0520 100644 --- a/packages/mcp/ensure-bound.ts +++ b/packages/mcp/ensure-bound.ts @@ -16,6 +16,28 @@ export interface EnsureBoundDeps { * `composeBotName`. */ readonly name: BotName + /** + * Post-acquire work, sequenced into the caller's call but run AFTER the + * state machine has recorded the binding. + * + * The ordering is load-bearing and the reason this is a separate hook rather + * than something `acquire` wraps. Post-acquire work RE-ENTERS the bind seam: + * seeding a seat's subscriptions calls `inbox.subscribe`, which reaches for a + * bound credential like any other state-holding call. Run while the state + * still says `pending`, that re-entry awaits the very `Deferred` its own + * caller is responsible for completing — a self-deadlock that hangs the + * seat's first action. + * + * Recording the binding first is also the honest model: the identity exists + * the moment `acquire` returns it. What follows is work done AS that + * identity, not part of obtaining it. + * + * The caller still waits for this, so the "subscriptions are seeded before + * the tool result returns" contract holds. What changes is that a CONCURRENT + * caller is released as soon as the identity exists rather than waiting out + * an unrelated caller's seeding. + */ + readonly afterAcquire?: (acquired: AcquiredIdentity) => Effect.Effect } /** @@ -100,6 +122,21 @@ export const createEnsureBound = (deps: EnsureBoundDeps): Effect.Effect + deps.afterAcquire === undefined + ? Effect.void + : // `onError`, not `tapError`: a hook that THROWS dies rather than + // failing, and a defect must drop the binding for the same reason + // a typed failure does — the post-acquire work did not complete. + deps + .afterAcquire(acquired) + .pipe(Effect.onError(() => Ref.set(stateRef, { kind: 'idle' as const }))), + ), ) }) diff --git a/packages/mcp/identity-cache.ts b/packages/mcp/identity-cache.ts index 3ae4483..2ba02a4 100644 --- a/packages/mcp/identity-cache.ts +++ b/packages/mcp/identity-cache.ts @@ -194,33 +194,26 @@ export const createEphemeralIdentityCache = ( ...(project !== undefined ? { project } : {}), }) - const wrapWithOnAcquire = + // `onAcquire` is handed to `createEnsureBound` as its `afterAcquire` hook + // rather than wrapped around `acquire` here. It re-enters the bind seam — + // seeding subscriptions calls `inbox.subscribe`, which reaches for a bound + // credential — so it has to run after the binding is recorded, or it awaits + // the Deferred its own caller must complete. See `EnsureBoundDeps`. + const afterAcquireFor = (project: ProjectSlug | undefined, sessionId: SessionId) => - (n: BotName): Effect.Effect => - deps - .acquire(n) - .pipe( - Effect.tap((acquired) => - deps.onAcquire !== undefined - ? deps.onAcquire(acquired, project, sessionId) - : Effect.void, - ), - ) + (acquired: AcquiredIdentity): Effect.Effect => + deps.onAcquire !== undefined ? deps.onAcquire(acquired, project, sessionId) : Effect.void // Capture the prior slot's release into the new slot's first acquire: // the release-then-acquire only fires if the prior identity actually // bound. `priorEnsure.current()` is read when the new acquire runs, so a // prior still mid-acquire (current() === undefined) skips release. const acquireForTransition = - ( - priorEnsure: EnsureBound, - project: ProjectSlug | undefined, - sessionId: SessionId, - ) => + (priorEnsure: EnsureBound) => (n: BotName): Effect.Effect => priorEnsure.current() !== undefined - ? deps.release().pipe(Effect.zipRight(wrapWithOnAcquire(project, sessionId)(n))) - : wrapWithOnAcquire(project, sessionId)(n) + ? deps.release().pipe(Effect.zipRight(deps.acquire(n))) + : deps.acquire(n) const mintSlot = ( sessionId: SessionId, @@ -229,11 +222,12 @@ export const createEphemeralIdentityCache = ( prior: Slot | undefined, ): Effect.Effect, Slot]> => { const name = deriveBotName(sessionId, project) - const acquire = - prior !== undefined - ? acquireForTransition(prior.ensureBound, project, sessionId) - : wrapWithOnAcquire(project, sessionId) - return createEnsureBound({ acquire, name }).pipe( + const acquire = prior !== undefined ? acquireForTransition(prior.ensureBound) : deps.acquire + return createEnsureBound({ + acquire, + name, + afterAcquire: afterAcquireFor(project, sessionId), + }).pipe( Effect.map( (ensureBound) => [ensureBound, { sessionId, ensureBound, lastUsedMs: nowMs }] as const, ), diff --git a/packages/mcp/subscription-resume.test.ts b/packages/mcp/subscription-resume.test.ts index acaf15f..860495a 100644 --- a/packages/mcp/subscription-resume.test.ts +++ b/packages/mcp/subscription-resume.test.ts @@ -339,7 +339,15 @@ const buildPersistRig = (): Effect.Effect => return { client, store, session } }) -test('subscribe carrying no session_id persists the snapshot when the id is already known', () => +// Both of these tests used to assert the opposite, and the premise they rested +// on has been withdrawn. They read "the matcher never stamps subscribe, so this +// is the real live shape" — true under commit 0f0e755 (PR #126), which chose +// id-blind subscribe because a subscription was written under the SHARED MINTER +// and needed no identity of its own. comms-g5zh.3 retires that: a subscription +// is realm state under the seat's own principal, so subscribe mints, and the +// matcher now stamps it. The reversal is deliberate and ratified (Graeme, +// 2026-07-31); see the commit message. +test('subscribe carries a stamped session_id and persists the snapshot under it', () => Effect.runPromise( Effect.scoped( Effect.gen(function* () { @@ -348,34 +356,50 @@ test('subscribe carrying no session_id persists the snapshot when the id is alre // shared deferred — the fleet's real state (CC injects the env at boot). yield* Deferred.succeed(rig.session, asSessionId(SID_RESUME)) - // A subscribe that carries NO session_id in args: the matcher never - // stamps it on subscribe, so this is the real live shape. yield* Effect.promise(() => - rig.client.callTool({ name: 'subscribe', arguments: { target: 'other' } }), + rig.client.callTool({ + name: 'subscribe', + arguments: { target: 'other', session_id: SID_RESUME }, + }), ) - // Persist fired id-blind: the session-keyed snapshot now holds the new - // intent, so a later resume restores it. + // The session-keyed snapshot now holds the new intent, so a later + // resume restores it. const persisted = yield* rig.store.read() expect(persisted).toEqual(Option.some([channelOtherIntent])) }), ), )) -test('subscribe carrying no session_id with an unfed deferred returns promptly and does not park', () => +test('subscribe carrying no session_id is refused promptly rather than parking', () => Effect.runPromise( Effect.scoped( Effect.gen(function* () { const rig = yield* buildPersistRig() - // Deferred deliberately UNFED: a subscribe-first seat whose id no source - // has delivered. An unconditional persist would park on the store's - // `Deferred.await`; the poll-guard must no-op and let the call return. + // No stamped id and an UNFED deferred: a non-CC host that supplies + // neither. Declaring interest needs a principal to hold the + // subscription, and this seat cannot name one — so the seam refuses. + // + // The load-bearing half is the SHAPE of that refusal. It must be a + // prompt, typed error and never a park: the whole reason the bind seam + // reads its session context rather than awaiting a deferred is that a + // caller must never hang waiting for an identity that may never arrive. const outcome = yield* Effect.promise(() => - rig.client.callTool({ name: 'subscribe', arguments: { target: 'other' } }), + rig.client + .callTool({ name: 'subscribe', arguments: { target: 'other' } }) + .then(() => 'resolved' as const) + .catch((err: unknown) => (Predicate.isError(err) ? err.message : String(err))), ).pipe(Effect.timeoutOption('2 seconds')) - // Handler returned rather than hanging on the unfed deferred. expect(Option.isSome(outcome)).toBe(true) + expect(Option.getOrElse(outcome, () => '')).toMatch(/requires a session_id/) + + // Nothing was written on the way to refusing — no half-applied intent + // left behind for a resume to restore. The store is id-keyed and its + // read awaits the shared deferred, so the id is revealed only now, + // AFTER the refusal, purely to make the store readable. + yield* Deferred.succeed(rig.session, asSessionId(SID_RESUME)) + expect(yield* rig.store.read()).toEqual(Option.none()) }), ), )) diff --git a/packages/mcp/tools.ts b/packages/mcp/tools.ts index e601204..24fdd6a 100644 --- a/packages/mcp/tools.ts +++ b/packages/mcp/tools.ts @@ -561,7 +561,14 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA // id-blind and best-effort (see the subscribe handler): the post and react // handlers both reach here via ensure-bound, so the session-id deferred is // already fed and the store already seeded or restored by this point. + // Takes the CALLER'S `run`, not the bare runtime edge. Subscribing is a + // state-holding call and binds like any other, so it needs the calling + // session's context; running it at the edge instead would put it on a fiber + // with no session id, where the seam correctly refuses. That refusal would + // land on a caller who did nothing wrong — they posted, and the sticky + // subscribe is our inference from that post. const stickyThreadEngagement = async ( + run: (effect: Effect.Effect) => Promise, channel: ChannelRef, threadName: Option.Option, ): Promise => { @@ -572,7 +579,7 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA threadName: threadName.value, } narrowSet.add(intent) - await runEdge( + await run( adapter.inbox .subscribe(intentToTarget(intent)) .pipe(Effect.zipRight(deps.persistSessionSubscriptions ?? Effect.void)), @@ -735,6 +742,7 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA ) cache.rememberMessage(ref) await stickyThreadEngagement( + run, ref.channel, Option.map(ref.thread, (t) => t.name), ) @@ -834,7 +842,7 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA }), ) cache.rememberMessage(ref) - await stickyThreadEngagement(ref.channel, threadName) + await stickyThreadEngagement(run, ref.channel, threadName) return {} }, }, diff --git a/packages/memory/adapter.ts b/packages/memory/adapter.ts index 171bd19..6011a42 100644 --- a/packages/memory/adapter.ts +++ b/packages/memory/adapter.ts @@ -781,8 +781,25 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect Ref.update(subscriptions, HashSet.add(subscriptionKey(target))), - unsubscribe: (target) => Ref.update(subscriptions, HashSet.remove(subscriptionKey(target))), + // Declaring interest BINDS, in step with the Zulip adapter (comms-g5zh.2 + // / .3): a subscription is state the realm holds on the agent's behalf, + // and the event queue that delivers against it is registered under the + // same principal. + // + // This adapter holds no realm, so nothing here would break if it skipped + // the bind — which is exactly why it must not. Every rig that boots the + // server against this adapter would then exercise a seam that differs + // from the one that ships, and a seat that cannot bind would look + // perfectly healthy in tests while receiving nothing in production. That + // divergence is the failure comms-hsym recorded, not a saving. + subscribe: (target) => + requireBound().pipe( + Effect.zipRight(Ref.update(subscriptions, HashSet.add(subscriptionKey(target)))), + ), + unsubscribe: (target) => + requireBound().pipe( + Effect.zipRight(Ref.update(subscriptions, HashSet.remove(subscriptionKey(target)))), + ), events: () => Stream.asyncPush((emit) => Effect.acquireRelease( From 9ea4170b7983ce039779fe80947d947463a8f3a4 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 09:05:35 +0100 Subject: [PATCH 6/6] Ask the ownership probe not to long-poll GET /events holds ~50s waiting for an event that never comes on an idle queue, so the seat's half of the ownership check timed out rather than answering. The probe asks whether the realm accepts the caller, not what is on the queue, so dont_block is the right question to ask. Verified against the realm: the minter's poll of a seat-registered queue is refused with BAD_EVENT_QUEUE_ID, the seat's poll of the same queue succeeds, a channel message reaches the seat's own queue, and one seat unsubscribing leaves another receiving. comms-g5zh.2 Claude-Session: https://claude.ai/code/session_01NdgPHEUtz1xqbFEiN3cxCB --- packages/zulip/realm.live.test.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/zulip/realm.live.test.ts b/packages/zulip/realm.live.test.ts index 94395dc..fb70498 100644 --- a/packages/zulip/realm.live.test.ts +++ b/packages/zulip/realm.live.test.ts @@ -561,6 +561,11 @@ describeLiveChannel('zulip live per-seat receiving — zulip.example.com', () => asMinter.get('/events', anyEventsSchema, { queue_id: queue.queueId, last_event_id: queue.lastEventId, + // `/events` LONG-POLLS by default, holding ~50s for an + // event that will never come on an idle queue. This probe + // is about whether the realm accepts the caller, not about + // events, so ask it to answer now. + dont_block: true, }), ) expect(refusal).toBeInstanceOf(ZulipApiError) @@ -573,6 +578,7 @@ describeLiveChannel('zulip live per-seat receiving — zulip.example.com', () => const ok = yield* asSeat.get('/events', anyEventsSchema, { queue_id: queue.queueId, last_event_id: queue.lastEventId, + dont_block: true, }) expect(ok.result).toBe('success') }),