diff --git a/clients/claude-code/hooks-manifest.test.ts b/clients/claude-code/hooks-manifest.test.ts index 70e3edb..92ccdbd 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. + * + * 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. + * + * 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 = [] 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 () => { @@ -197,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() @@ -230,10 +304,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/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/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/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/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/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/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( diff --git a/packages/zulip/adapter-events.test.ts b/packages/zulip/adapter-events.test.ts index e4e2af0..e08e3c2 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,130 @@ 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 }, +) + +// 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 +// 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.test.ts b/packages/zulip/adapter.test.ts index 54ad8d0..74a4aa5 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -2831,21 +2831,144 @@ 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) + }), +) + +// 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 +// 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..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?" @@ -1747,10 +1773,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 +1799,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([ @@ -1858,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 4537a89..d1ab703 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 @@ -506,7 +531,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 @@ -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,10 +867,19 @@ 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( + ZulipApiError | ParseResult.ParseError | UnboundEphemeralSession + > = 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 @@ -866,7 +911,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 +976,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 +1057,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 = ( diff --git a/packages/zulip/realm.live.test.ts b/packages/zulip/realm.live.test.ts index 8bf126d..fb70498 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,191 @@ 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, + // `/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) + 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, + dont_block: true, + }) + 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.