diff --git a/README.md b/README.md index b2ba9ce..4870196 100644 --- a/README.md +++ b/README.md @@ -74,9 +74,11 @@ minter user that owns your agents' bot identities (see | `ZULIP_MINTER_EMAIL` | yes | Email of the minter user that owns every agent bot. Must be in the realm's `can_create_bots_group`. | | `ZULIP_MINTER_API_KEY` | yes | The minter's API key. Stored in the system keychain — never in `settings.json`. | -There's also an optional `COMMY_SUBSCRIBE` (comma-separated auto-subscribe -tokens, e.g. `my-project`) for agents that should already be -listening the moment they boot. Mentions of the bot need no token — they +There's also an optional `COMMY_SUBSCRIBE` (comma-separated subscribe tokens, +e.g. `my-project`) for agents that should already be listening the moment they +boot. It bootstraps a bot's subscriptions when the bot is created and is not +read again — after that the agent owns them and changes them with the +`subscribe` / `unsubscribe` tools. Mentions of the bot need no token — they always arrive. To set any of these non-interactively, repeat `--config KEY=value` on the `install` line. diff --git a/clients/claude-code/README.md b/clients/claude-code/README.md index 1bff6f5..9920498 100644 --- a/clients/claude-code/README.md +++ b/clients/claude-code/README.md @@ -43,7 +43,7 @@ the env block in `.mcp.json`. | `ZULIP_SITE` | yes | no | absolute URL, e.g. `https://zulip.example.com` | Base of the Zulip realm the plugin operates against. | | `ZULIP_MINTER_EMAIL` | yes | no | email-shaped string | Delivery email of the shared minter user (human-type Zulip user) that owns all bots managed through this plugin. Must be a member of the realm's `can_create_bots_group`. | | `ZULIP_MINTER_API_KEY` | yes | yes | opaque token | Minter user's API key. Used to mint or regenerate bot credentials at boot. | -| `COMMY_SUBSCRIBE` | no | no | comma-separated tokens: ``, `/`, `new-topics:` | Pre-loaded inbox subscriptions. A bare path is scope, a prefix is a delivery rule — `new-topics:` is the only prefix left. Mentions of the bot need no token — they arrive unconditionally; the retired `channel:` / `thread:` prefixes and the retired `mentions` token are rejected as config errors. Applied at MCP-child boot before tools are announced, in both eager and lazy modes — the minter is the universal listener; this just teaches the plugin which events to surface. Two supply paths, and they compose: this prompted value is written to `COMMY_SUBSCRIBE_USER_CONFIG`, and the plugin falls back to an inherited `COMMY_SUBSCRIBE` when that is empty. The plugin therefore never overwrites a value you set by other means — a systemd unit, a pane env, a nix module — which it did while it wrote the substitution to the bare name. Leaving both unset is not a boot failure; it means no subscriptions. | +| `COMMY_SUBSCRIBE` | no | no | comma-separated tokens: ``, `/`, `new-topics:` | Pre-loaded inbox subscriptions. A bare path is scope, a prefix is a delivery rule — `new-topics:` is the only prefix left. Mentions of the bot need no token — they arrive unconditionally; the retired `channel:` / `thread:` prefixes and the retired `mentions` token are rejected as config errors. Applied ONCE, when the bot is created, in both eager and lazy modes; after that the bot owns its subscriptions and changes them through the `subscribe` / `unsubscribe` tools, so editing this for a bot that already exists has no effect. Two supply paths, and they compose: this prompted value is written to `COMMY_SUBSCRIBE_USER_CONFIG`, and the plugin falls back to an inherited `COMMY_SUBSCRIBE` when that is empty. The plugin therefore never overwrites a value you set by other means — a systemd unit, a pane env, a nix module — which it did while it wrote the substitution to the bare name. Leaving both unset is not a boot failure; it means no subscriptions. | Paste each value into the prompt when it appears on first enable. To re-enter values later, edit `pluginConfigs[...].options` in `~/.claude/settings.json` diff --git a/clients/claude-code/hooks-manifest.test.ts b/clients/claude-code/hooks-manifest.test.ts index 92ccdbd..4cb78dc 100644 --- a/clients/claude-code/hooks-manifest.test.ts +++ b/clients/claude-code/hooks-manifest.test.ts @@ -54,6 +54,7 @@ const BOUND_HTTP_CALLERS = [ 'setChannelDescription', 'setThreadResolved', 'subscribe', + 'subscriptions', 'unreact', 'unsubscribe', ] as const @@ -64,13 +65,20 @@ const BOUND_HTTP_CALLERS = [ * a subscription row and the event queue that delivers against it — so these * bind for the same reason the publisher verbs do. * + * `subscriptions` (comms-g5zh.5) is the one READ in this set, and it belongs + * here for a different reason than the writes: it asks what the realm holds + * FOR THIS SEAT, so it can only be answered by the seat's own credential. The + * minter's answer would be a different seat's subscriptions wearing this one's + * name. It never causes a mint, though — its only caller checks that the seat + * is already bound, or has grounds to believe its bot exists, before asking. + * * 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 +const BOUND_INBOX_VERBS = ['subscribe', 'subscriptions', 'unsubscribe'] as const /** * Tools that reach `boundHttp` through an inbox verb while sitting outside the diff --git a/docs/agent-experience.md b/docs/agent-experience.md index 5dfc9ca..0904fb6 100644 --- a/docs/agent-experience.md +++ b/docs/agent-experience.md @@ -125,12 +125,14 @@ Places the current implementation fails this reference. exactly one test-side module, this one" — and production code has since eroded it. (`bootstrap.ts` naming the adapter is fine; that is the composition root choosing an implementation.) Principle 2. -- **Sticky subscriptions are keyed on `session_id`, not identity** - (`packages/mcp/subscription-store.ts`). Subscriptions belong to the - account: a human relaunching their browser keeps them. The code - deliberately rejected identity-keying to avoid treating a relaunch of a - pinned pane as a resume — but under this reference, that is the correct - behaviour, not a bug to avoid. Principle 3. +- **What survives a restart is keyed on `session_id`, not identity** + (`packages/mcp/subscription-store.ts`). Mostly closed: a seat now reads its + channel subscriptions back from the realm under its own principal, so a + relaunched pinned pane keeps them the way a human relaunching a browser + does. What is still session-keyed is the record of TOPIC-level narrows — + the part a subscription row cannot express. That record is the last piece + of client-side authority here, and the paragraph on topic narrows below + says where it belongs instead. Principle 3. - **Exploration is thinner than a human's.** Agents get `read_channel`, `read_thread` and `list_channels`; a human member also gets search and unread state. Principle 1, prospectively — this is a gap to fill, not @@ -160,6 +162,12 @@ deafen another. A filter that lives in memory is lost on resume, so it needs 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 store step is now mostly undone too. Once subscriptions are the seat's own +realm state, resume stops being "restore what I saved" and becomes "ask what I +am subscribed to" — what a human's client does. Only the topic-level narrows +still need a local record, because a subscription row names a channel and +nothing finer. + The refcounting step is the one the architecture never took, and its absence 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 @@ -181,8 +189,10 @@ 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. +subscription, and topic-level narrowing is still a client-side filter with a +local record behind it. Those are the next steps of the same unwinding, not +exemptions — channel-level narrowing has moved, and is now read back from the +realm on every boot. 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 3254ff9..4aad6e8 100644 --- a/docs/self-hosting.md +++ b/docs/self-hosting.md @@ -28,7 +28,7 @@ no-op-ish values when unset): |---|---|---| | `COMMY_BOT_NAME` | no | Persistent mode: a stable identity acquired eagerly at boot (for concierges / scheduled agents). Omit for ephemeral, per-session identities. | | `COMMY_PROJECT` | no | Project slug used for channel naming and a persistent agent's project subscriptions. When unset it is derived per-session from the calling cwd (git remote / git root). | -| `COMMY_SUBSCRIBE` | no | Comma-separated auto-subscribe tokens applied at boot: `` for a whole channel, `/` for one topic in it, `new-topics:` for the first message of each new topic. Blank means no auto-subscription — the bot still receives its own mentions, which are implicit and need no token. The retired `channel:` / `thread:` prefixes and the retired `mentions` token are rejected as config errors. Under the Claude Code plugin this may also arrive as `COMMY_SUBSCRIBE_USER_CONFIG`, which takes precedence; see [Two supply paths for the optional `COMMY_*` values](#two-supply-paths-for-the-optional-commy-values). | +| `COMMY_SUBSCRIBE` | no | Comma-separated tokens that **bootstrap a new bot's subscriptions, once**: `` for a whole channel, `/` for one topic in it, `new-topics:` for the first message of each new topic. Applied when the bot is created and never again — see [`COMMY_SUBSCRIBE` bootstraps a bot; it does not configure one](#commy_subscribe-bootstraps-a-bot-it-does-not-configure-one). Blank means no bootstrap — the bot still receives its own mentions, which are implicit and need no token. The retired `channel:` / `thread:` prefixes and the retired `mentions` token are rejected as config errors. Under the Claude Code plugin this may also arrive as `COMMY_SUBSCRIBE_USER_CONFIG`, which takes precedence; see [Two supply paths for the optional `COMMY_*` values](#two-supply-paths-for-the-optional-commy-values). | | `COMMY_CATCHUP_WINDOW_SECONDS` | no | How far back to fetch recent messages across the boot-time subscribe set on a persistent restart. Default `14400` (4 hours); `0` disables. | | `COMMY_QUEUE_IDLE_TIMEOUT_SECS` | no | How many seconds an ephemeral session's events queue survives without a poll before Zulip garbage-collects it, sent as `idle_queue_timeout` on `/register`. Default `86400` (24 hours); clamped to Zulip's 7-day `MAX_QUEUE_TIMEOUT_SECS` ceiling (`604800`). A non-positive or non-integer value fails boot with a config error. | | `COMMY_DOWNLOAD_DIR` | no | Base directory for `download_file` attachments. When set, each download's fresh temp subdirectory is created under it so files land somewhere an allowlisted agent can `Read`; when unset, downloads go to `$TMPDIR`. Must be an existing directory — a non-directory value fails boot with a config error. | @@ -69,14 +69,43 @@ booted a seat that was silently deaf to every channel it was meant to watch. Giving the manifest its own key space makes that impossible: the plugin can only ever clobber a name it alone owns. -If you configure subscriptions and want to confirm they arrived, boot logs a -line naming the applied tokens in the same vocabulary you wrote them: +If you configure subscriptions and want to confirm they arrived, the boot that +creates the bot logs a line naming the applied tokens in the same vocabulary you +wrote them: ``` commy plugin: applied 2 boot-time subscribe target(s): myproject, general/standup ``` -No such line means no boot-time subscriptions were applied. +That line appears **once in a bot's life**. On a later launch you get this +instead, which is the design working rather than a fault: + +``` +commy plugin: COMMY_SUBSCRIBE not applied — myproject-concierge already exists +and owns its subscriptions. +``` + +## `COMMY_SUBSCRIBE` bootstraps a bot; it does not configure one + +`COMMY_SUBSCRIBE` is applied at the moment a bot is created, and never read +again. From then on the bot owns its subscriptions: they live in the realm under +its own principal, and they change when the agent itself calls `subscribe` or +`unsubscribe`. + +**Editing `COMMY_SUBSCRIBE` for a bot that already exists has no effect.** To +change what a running agent listens to, ask the agent — it has the tools. To +start over from the launcher, delete the bot; the next boot creates a new one and +bootstraps it from the current value. + +The reason is that a launcher value and an agent's own calls cannot both be +authoritative. When both were, the same `COMMY_SUBSCRIBE` read at two launches of +one bot was indistinguishable from the bot's own runtime changes, and the two +quietly fought. The realm holds one answer now, and the agent is the one who +changes it. + +One consequence for an upgrade: a bot that already existed before this behaviour +landed is **not** bootstrapped retroactively. It keeps whatever the realm holds +for it — and a bot that finds itself with nothing can subscribe to what it wants. ## Realm settings that shape commy behaviour diff --git a/packages/core/ports.ts b/packages/core/ports.ts index cd72012..0e9bc0e 100644 --- a/packages/core/ports.ts +++ b/packages/core/ports.ts @@ -445,9 +445,28 @@ export interface RealmSettings { */ export type Credentials = Readonly> +/** + * Whether an acquire brought an identity into existence or bound to one that + * was already there. + * + * The substrate answers this as a by-product of acquiring — Zulip's acquire is + * a lookup that mints on miss — so a caller reads the answer rather than + * inferring one from local state. That matters because the alternative + * inferences are all wrong in a way that shows up only in production: "no + * subscriptions yet" cannot tell a never-seeded bot from one that deliberately + * unsubscribed from everything, and "no local state file" cannot tell a fresh + * seat from one whose state was pruned. + * + * `minted` is the one moment a seat first exists, and it is what bootstrap + * hangs off: `COMMY_SUBSCRIBE` seeds a seat's subscriptions there and nowhere + * else, because from that point the bot owns them. + */ +export type IdentityOrigin = 'minted' | 'existing' + export interface AcquiredIdentity { readonly credentials: Credentials readonly identity: Identity + readonly origin: IdentityOrigin } /** @@ -673,6 +692,23 @@ export interface MessageInbox { */ subscribe(target: SubscriptionTarget): Effect.Effect unsubscribe(target: SubscriptionTarget): Effect.Effect + /** + * The channels the realm currently delivers to this seat, read back from the + * subscription rows written under its own principal. + * + * This is what makes a seat's subscriptions survive a restart without the + * plugin keeping its own copy of them: on the way up, a seat asks the realm + * what it is subscribed to rather than replaying what it once asked for. + * Reading it binds — the answer is per-member state, so there is no answer + * before the seat has a principal, and an unbound seat's honest answer is not + * "nothing" but "there is no seat yet". + * + * CHANNELS ONLY, and that ceiling is the substrate's rather than ours: a + * subscription row names a channel, so a seat that only wants one topic in a + * channel and a seat that wants the whole channel are indistinguishable here. + * Narrowing below a channel stays with the caller. + */ + subscriptions(): Effect.Effect, BindError | InboxError> /** * 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.test.ts b/packages/mcp/bootstrap.test.ts index 1d775da..c9ebeb9 100644 --- a/packages/mcp/bootstrap.test.ts +++ b/packages/mcp/bootstrap.test.ts @@ -896,6 +896,7 @@ const buildFakeInbox = (options: { readonly rejectOn?: number } = {}): FakeInbox return Effect.void }), unsubscribe: () => Effect.void, + subscriptions: () => Effect.succeed([]), settingsChanges: () => Stream.empty, events: () => Stream.empty, replay: () => Effect.succeed([]), diff --git a/packages/mcp/deployed-wiring.test.ts b/packages/mcp/deployed-wiring.test.ts index c023791..b3a9f5b 100644 --- a/packages/mcp/deployed-wiring.test.ts +++ b/packages/mcp/deployed-wiring.test.ts @@ -58,7 +58,7 @@ import type { BindOnDemand } from './session-binder.ts' import { bindThrough, SessionBinder as SessionBinderTag } from './session-binder.ts' import { SessionId as SessionIdTag, type SessionIdValue } from './session-id.ts' import { SubscriptionStoreTag } from './subscription-store.ts' -import { testPlatformLayer } from './test-platform.ts' +import { testBootStoresLayer, testPlatformLayer } from './test-platform.ts' const REPO_ROOT = join(import.meta.dir, '..', '..') const PLUGIN_ROOT = join(REPO_ROOT, 'clients', 'claude-code') @@ -238,6 +238,7 @@ const bootDeployedSeat = async ( Layer.succeed(SessionIdTag, sessionIdDeferred), Layer.succeed(ResumeOutcomeTag, resumeOutcome), Layer.succeed(SessionBinderTag, binderRef), + testBootStoresLayer(), captureLogger([]), ), testPlatformLayer(CLAUDE_CODE_CHILD_ENV), diff --git a/packages/mcp/disconnect-exit.fixture.ts b/packages/mcp/disconnect-exit.fixture.ts index ff24d00..14e6ab8 100644 --- a/packages/mcp/disconnect-exit.fixture.ts +++ b/packages/mcp/disconnect-exit.fixture.ts @@ -30,6 +30,7 @@ import { clientDisconnect, makeProgram } from './server.ts' import { SessionBinderLive } from './session-binder.ts' import { SessionIdLive } from './session-id.ts' import { SubscriptionStoreTag } from './subscription-store.ts' +import { testBootStoresLayer } from './test-platform.ts' const inMemoryCursorStore = { read: () => Effect.succeed(Option.none()), @@ -80,6 +81,7 @@ NodeRuntime.runMain( substrateAdapterLayer(substrate), Layer.succeed(CursorStoreTag, inMemoryCursorStore), Layer.succeed(SubscriptionStoreTag, inMemorySubscriptionStore), + testBootStoresLayer(), SessionIdLive, SessionBinderLive, ResumeOutcomeLive, diff --git a/packages/mcp/ensure-bound.test.ts b/packages/mcp/ensure-bound.test.ts index f1f2763..f515603 100644 --- a/packages/mcp/ensure-bound.test.ts +++ b/packages/mcp/ensure-bound.test.ts @@ -23,6 +23,7 @@ const buildIdentity = (name: string): Identity => ({ const buildAcquired = (name: string): AcquiredIdentity => ({ identity: buildIdentity(name), credentials: { apiKey: `key-${name}` }, + origin: 'minted', }) const run = (self: Effect.Effect): Promise => Effect.runPromise(self) diff --git a/packages/mcp/event-pump.test.ts b/packages/mcp/event-pump.test.ts index 74f2b56..17fd50c 100644 --- a/packages/mcp/event-pump.test.ts +++ b/packages/mcp/event-pump.test.ts @@ -119,6 +119,7 @@ const queueInbox = (options: QueueInboxOptions = { events: [] }): QueueInboxHand const inbox: MessageInbox = { subscribe: () => Effect.void, unsubscribe: () => Effect.void, + subscriptions: () => Effect.succeed([]), settingsChanges: () => Stream.empty, events: () => Stream.async((emit) => { @@ -200,6 +201,7 @@ test('pump reports realm settings changes without notifying the consumer', () => const inbox: MessageInbox = { subscribe: () => Effect.void, unsubscribe: () => Effect.void, + subscriptions: () => Effect.succeed([]), settingsChanges: () => Stream.make({ editingAvailable: false }), events: () => Stream.empty, replay: () => Effect.succeed([]), diff --git a/packages/mcp/identity-cache.test.ts b/packages/mcp/identity-cache.test.ts index 20d4d84..af1739d 100644 --- a/packages/mcp/identity-cache.test.ts +++ b/packages/mcp/identity-cache.test.ts @@ -54,6 +54,7 @@ const buildIdentity = (name: string): Identity => ({ const buildAcquired = (name: string): AcquiredIdentity => ({ identity: buildIdentity(name), credentials: { apiKey: `key-${name}` }, + origin: 'minted', }) /** diff --git a/packages/mcp/server.integration.test.ts b/packages/mcp/server.integration.test.ts index 06537a5..180409a 100644 --- a/packages/mcp/server.integration.test.ts +++ b/packages/mcp/server.integration.test.ts @@ -49,6 +49,7 @@ import { CursorStoreTag } from './cursor-store.ts' // `completeAsSubstrate` is the single seam that completes it to the Zulip-shaped // `SubstrateAdapter` port; no Zulip type or brand is named directly here. import { completeAsSubstrate } from './memory-substrate.ts' +import { QueueStateStoreTag } from './queue-state-store.ts' import { ResumeOutcome as ResumeOutcomeTag } from './resume-outcome.ts' import { makeProgram } from './server.ts' import type { BindOnDemand } from './session-binder.ts' @@ -57,7 +58,7 @@ import { SessionId as SessionIdTag, type SessionIdValue } from './session-id.ts' import type { SubscribeIntent } from './subscribe-parser.ts' import type { SubscriptionStore } from './subscription-store.ts' import { SubscriptionStoreTag } from './subscription-store.ts' -import { testPlatformLayer } from './test-platform.ts' +import { createInMemoryQueueStateStore, testPlatformLayer } from './test-platform.ts' const createMemoryCursorStore = (): CursorStore => { const store = new Map() @@ -381,6 +382,7 @@ const buildHarness = async (overrides: AdapterOverrides = {}): Promise Layer.succeed(SessionIdTag, sessionIdDeferred), Layer.succeed(ResumeOutcomeTag, resumeOutcomeDeferred), Layer.succeed(SessionBinderTag, binderRef), + Layer.succeed(QueueStateStoreTag, createInMemoryQueueStateStore()), loggerLayer, ), testPlatformLayer(env), @@ -1838,7 +1840,11 @@ test('ephemeral queue-DEAD resume: mentions + channels catch-up run and backfill // ─── Type-2 default sub set for interactive CC sessions ───────── -const captureSubscribes = (): { +const captureSubscribes = ( + // What the realm reports this seat is already subscribed to — the rows a + // previous run of the same session left behind. + realmChannels: ReadonlyArray = [], +): { readonly inboxOverrides: Partial readonly tokens: ReadonlyArray } => { @@ -1855,6 +1861,7 @@ const captureSubscribes = (): { Effect.sync(() => { tokens.push(renderTarget(target)) }), + subscriptions: () => Effect.succeed(realmChannels.map((n) => decodeChannelNameSync(n))), }, } } @@ -2011,29 +2018,28 @@ test('ephemeral subscribe persists the live narrow set (defaults + new sub) unde await waitFor(() => writes.length > 0, 200) const last = writes.at(-1) expect(last?.sid).toBe(sid) - // The snapshot is the full live set. With no project there is no Type-2 - // default to seed, so it is exactly the channel just subscribed — and a - // later resume restores that. - expect(new Set((last?.intents ?? []).map((i) => JSON.stringify(i)))).toEqual( - new Set([JSON.stringify({ kind: 'channel', channelName: decodeChannelNameSync('home') })]), - ) + // Only the topic-level slice is written, and a plain channel subscribe has + // none — the realm holds that row, and a copy of it here could only + // disagree with what actually governs delivery. The write still happens, + // because its PRESENCE is what tells the next launch this session has run + // before. + expect(last?.intents).toEqual([]) } finally { await h.cleanup() } }) -test('ephemeral resume restores the persisted narrow set and does NOT re-apply Type-2 defaults', async () => { - // A prior session persisted a single channel and had dropped the other - // defaults. Resume must honour that exactly — restore the `home` channel, - // and never re-add a dropped default. - const persisted: ReadonlyArray = [ - { kind: 'channel', channelName: decodeChannelNameSync('home') }, - ] +test('ephemeral resume recovers its channels from the realm and does NOT re-apply Type-2 defaults', async () => { + // A prior run of this session subscribed `home` and had dropped the other + // defaults. `home` lives in the REALM now — that is where a channel + // subscription is held — and the session record is present but empty, which + // is what marks this a resume rather than a first launch. Resume must honour + // both: recover `home`, and never re-add a dropped default. const subscriptionStore: SubscriptionStore = { - read: () => Effect.succeed(Option.some(persisted)), + read: () => Effect.succeed(Option.some([])), write: () => Effect.void, } - const cap = captureSubscribes() + const cap = captureSubscribes(['home']) const h = await buildHarness({ ephemeral: true, env: { COMMY_PROJECT: 'myproject' }, diff --git a/packages/mcp/server.live.test.ts b/packages/mcp/server.live.test.ts index e486569..968b6fe 100644 --- a/packages/mcp/server.live.test.ts +++ b/packages/mcp/server.live.test.ts @@ -244,6 +244,7 @@ const buildHarness = ( Layer.mergeAll( substrateAdapterLayer(parseEnv.pipe(Effect.as(wrappedAdapter))), FileCursorStoreLive, + FileQueueStateStoreLive, // Feed the one shared session-id deferred into the store, which now // awaits it — mergeAll won't wire a sibling's output to a sibling's // input, so a plain merge leaves the store's SessionId unsatisfied. diff --git a/packages/mcp/server.test.ts b/packages/mcp/server.test.ts index f59d66e..ecae1f8 100644 --- a/packages/mcp/server.test.ts +++ b/packages/mcp/server.test.ts @@ -8,6 +8,7 @@ import type { HistoryReader, Identity, IdentityId as IdentityIdType, + IdentityOrigin, IdentityPort, MessageInbox, MessagePublisher, @@ -49,6 +50,7 @@ import type { IdentityCache } from './identity-cache.ts' // re-exports the `ZulipAdapter` type so this file names no `@commy/zulip` module // directly. import { completeAsSubstrate, type ZulipAdapter } from './memory-substrate.ts' +import { QueueStateStoreTag } from './queue-state-store.ts' import { ResumeOutcomeLive } from './resume-outcome.ts' import { clientDisconnect, forkIdleSweep, makeProgram, type ProgramParams } from './server.ts' import type { BindOnDemand } from './session-binder.ts' @@ -61,7 +63,7 @@ import { SessionIdLive } from './session-id.ts' import type { SubscribeIntent } from './subscribe-parser.ts' import type { SubscriptionStore } from './subscription-store.ts' import { SubscriptionStoreTag } from './subscription-store.ts' -import { testPlatformLayer } from './test-platform.ts' +import { createInMemoryQueueStateStore, testPlatformLayer } from './test-platform.ts' /** * In-memory cursor store for the boot tests — keeps the runner's homedir @@ -132,6 +134,7 @@ const runProgram = ( ? SessionBinderLive : Layer.succeed(SessionBinderTag, binderRef), ResumeOutcomeLive, + Layer.succeed(QueueStateStoreTag, createInMemoryQueueStateStore()), loggerLayer, ), testPlatformLayer(env), @@ -152,6 +155,14 @@ const validEnv = { COMMY_BOT_NAME: 'myproject-concierge', } as const +/** Ephemeral (lazy) mode: no COMMY_BOT_NAME, so the bot name comes off the session id. */ +const lazyEnv = { + ZULIP_SITE: 'https://zulip.example.com', + ZULIP_MINTER_EMAIL: 'minter-bot@zulip.example.com', + ZULIP_MINTER_API_KEY: 'kkkkkkkkkkkkkkkkkkkkkkkkkkkkkkk1', + CLAUDE_CODE_SESSION_ID: 'abcdef12-3456-4789-89ab-cdef01234567', +} as const + interface FakeAdapterCalls { readonly acquired: string[] readonly closes: { count: number } @@ -163,6 +174,10 @@ interface FakeAdapterCalls { const buildFakeAdapter = ( options: { readonly acquireError?: unknown + /** What the substrate says about the bind: a fresh mint, or a bot that was already there. */ + readonly identityOrigin?: IdentityOrigin + /** Reject every substrate-side subscribe, for the part-way-failure paths. */ + readonly subscribeError?: InboxError readonly reconcileReport?: { readonly added: ReadonlyArray readonly error: string | undefined @@ -182,6 +197,10 @@ const buildFakeAdapter = ( const acquiredIdentity: AcquiredIdentity = { credentials: { apiKey: 'fresh-key' }, identity, + // Default to a fresh mint: most boot tests exercise a bot coming into + // existence, which is when COMMY_SUBSCRIBE seeds. The already-existing + // case — a pinned bot on its second launch — is driven by the option. + origin: options.identityOrigin ?? 'minted', } const identityPort: IdentityPort = { currentIdentity: () => Effect.succeed(identity), @@ -207,11 +226,21 @@ const buildFakeAdapter = ( } const inbox: MessageInbox = { subscribe: (target: SubscriptionTarget) => - Effect.sync(() => { + Effect.suspend(() => { events.push('subscribe') + if (options.subscribeError !== undefined) return Effect.fail(options.subscribeError) subscribed.push(target) + return Effect.void }), unsubscribe: (_target: SubscriptionTarget) => Effect.void, + // What the realm would report back: a subscription row names a channel, so + // every narrow this fake was handed reads back as its channel. + subscriptions: () => + Effect.succeed( + subscribed.map((target) => + typeof target === 'string' ? target : (target.channel as ChannelName), + ), + ), settingsChanges: () => Stream.empty, events: () => Stream.empty, replay: (_since: TimestampType) => Effect.succeed([]), @@ -326,27 +355,93 @@ test('lazy mode (cc-<8> from session id) does NOT acquire at boot', async () => expect(fake.calls.closes.count).toBe(1) }) -test('lazy mode still applies COMMY_SUBSCRIBE at boot (pre-acquire subscriptions)', async () => { +// A seat asked to hold subscriptions mints at boot rather than at its first +// action, and the reason is structural: receiving needs an events queue, and a +// queue is realm state that needs a principal to hold it. So COMMY_SUBSCRIBE is +// what pulls an otherwise-lazy ephemeral seat into existence early — and the +// seeding rides that mint. +test('lazy mode mints at boot when COMMY_SUBSCRIBE gives the seat something to hold', async () => { const fake = buildFakeAdapter() const env = { - ZULIP_SITE: 'https://zulip.example.com', - ZULIP_MINTER_EMAIL: 'minter-bot@zulip.example.com', - ZULIP_MINTER_API_KEY: 'kkkkkkkkkkkkkkkkkkkkkkkkkkkkkkk1', - CLAUDE_CODE_SESSION_ID: 'abcdef12-3456-4789-89ab-cdef01234567', + ...lazyEnv, COMMY_SUBSCRIBE: 'home', } const logs: string[] = [] - await runProgram(env, fake.adapter, { loggerLayer: captureLogger(logs) }) - expect(fake.calls.acquired).toEqual([]) + await runProgram(env, fake.adapter, { + loggerLayer: captureLogger(logs), + readGitContext: () => Effect.succeed(NotInRepo()), + }) + expect(fake.calls.acquired).toEqual(['cc-abcdef12']) expect(fake.calls.subscribed).toEqual([decodeChannelNameSync('home')]) expect(fake.calls.closes.count).toBe(1) - // Boot leaves a positive trace of what it applied. An operator who configured - // subscriptions and finds no such line — or one that omits their tokens — is - // looking straight at the fault, which is the diagnostic a clobbered - // COMMY_SUBSCRIBE spent months not having. + // The one positive trace that the bootstrap ran. It appears at most once in a + // bot's life, so its absence on a later boot is the design, not a fault. expect(logs).toEqual(['commy plugin: applied 1 boot-time subscribe target(s): home']) }) +test('lazy mode does NOT mint at boot when COMMY_SUBSCRIBE is unset', async () => { + // The mirror of the test above, and the line the optimisation lives on: a + // seat the realm holds nothing for stays identity-free until it acts. + const fake = buildFakeAdapter() + await runProgram(lazyEnv, fake.adapter, { + readGitContext: () => Effect.succeed(NotInRepo()), + }) + expect(fake.calls.acquired).toEqual([]) + expect(fake.calls.subscribed).toEqual([]) +}) + +// ─── COMMY_SUBSCRIBE is a once-per-bot bootstrap (comms-g5zh.4 / comms-g5zh.8) ── +// Graeme's rulings: nix bootstraps the bot ACCOUNT and the bot owns its +// subscriptions once it exists (2026-07-20); and a bot that comes out of an +// upgrade with no subscriptions of its own is LEFT empty rather than +// bootstrapped, because a bot that wants subscriptions can subscribe +// (2026-07-31). Together those make the origin of the bind the whole test — +// there is nothing to remember between boots, so nothing is remembered. + +test('COMMY_SUBSCRIBE is inert on a bot that already exists', async () => { + // The bead's acceptance and the fleet's steady state: a pinned pane + // relaunching. The tokens are not re-applied, and an operator who edits them + // is told so rather than left reading silence. + const fake = buildFakeAdapter({ identityOrigin: 'existing' }) + const logs: string[] = [] + await runProgram({ ...validEnv, COMMY_SUBSCRIBE: 'home' }, fake.adapter, { + loggerLayer: captureLogger(logs), + readGitContext: () => Effect.succeed(NotInRepo()), + }) + expect(fake.calls.acquired).toEqual(['myproject-concierge']) + expect(fake.calls.subscribed).toEqual([]) + expect(logs).toEqual([ + 'commy plugin: COMMY_SUBSCRIBE not applied — myproject-concierge already exists and owns its ' + + 'subscriptions. Editing COMMY_SUBSCRIBE for a bot that already exists has no effect; change ' + + 'its subscriptions through the bot.', + ]) +}) + +test('an already-minted bot with no subscriptions of its own is left alone, not bootstrapped', async () => { + // comms-g5zh.8 as ruled. Every pinned bot crossing this upgrade has an empty + // own-subscription set — before this architecture the MINTER held the + // subscriptions, so no bot ever had any of its own. Bootstrapping them would + // resurrect subscriptions their agents may have dropped on purpose; a bot + // that wants subscriptions subscribes. + const fake = buildFakeAdapter({ identityOrigin: 'existing' }) + await runProgram({ ...validEnv, COMMY_SUBSCRIBE: 'home' }, fake.adapter, { + readGitContext: () => Effect.succeed(NotInRepo()), + }) + expect(fake.calls.subscribed).toEqual([]) +}) + +test('a freshly minted bot is bootstrapped', async () => { + // The other half of the same rule, and the only path that ever applies the + // tokens. A bot an administrator deleted and that we have since re-minted + // under the same name reaches here too — it is a new bot with no + // subscriptions, and the realm says so. + const fake = buildFakeAdapter({ identityOrigin: 'minted' }) + await runProgram({ ...validEnv, COMMY_SUBSCRIBE: 'home' }, fake.adapter, { + readGitContext: () => Effect.succeed(NotInRepo()), + }) + expect(fake.calls.subscribed).toEqual([decodeChannelNameSync('home')]) +}) + test('main acquire failure stringifies non-Error rejections', async () => { const fake = buildFakeAdapter({ acquireError: 'plain string rejection' }) const stderr: string[] = [] @@ -427,6 +522,87 @@ test('main drives a real memory adapter through acquire + env subscribe + close' expect(closes).toBe(1) }) +// ─── boot does not deadlock on the queue-resume verdict (comms-deg1) ────────── +// The ephemeral post-acquire hook BLOCKS on the resume verdict, and since +// subscriptions moved to the seat's own principal the boot-time subscribe binds +// — so that hook runs on the BOOT fiber. The verdict used to be reported only by +// the events producer, which `startEventPump` materialises as the LAST step of +// boot, so the boot fiber parked on a deferred only a step it would never reach +// could complete. Every ephemeral seat with COMMY_SUBSCRIBE set hung before +// announcing its tools. +// +// This test drives the real seam: a memory adapter whose `subscribe` binds +// through the same `bindOnDemand` holder the Zulip adapter uses, an events +// stream that never produces (standing in for a pump that has not started), and +// the production `ResumeOutcomeLive` — a genuinely uncompleted deferred. Before +// the fix this times out; a harness that pre-completes the verdict cannot see it. +test('boot completes for an ephemeral seat with COMMY_SUBSCRIBE and no resume verdict yet', async () => { + const binderRef = await Effect.runPromise(Ref.make>(Option.none())) + const adapter = await Effect.runPromise(memoryAdapter({ bindOnDemand: bindThrough(binderRef) })) + const neverProduces: MessageInbox['events'] = () => Stream.empty + const substrate = completeAsSubstrate( + { ...adapter, inbox: { ...adapter.inbox, events: neverProduces } }, + { close: async () => {} }, + ) + const exit = await Effect.runPromise( + Effect.exit( + Effect.promise(() => + runProgram( + { ...lazyEnv, COMMY_SUBSCRIBE: 'home' }, + substrate, + { + loggerLayer: captureLogger([]), + readGitContext: () => Effect.succeed(NotInRepo()), + }, + binderRef, + ), + ).pipe(Effect.timeoutFail({ duration: '5 seconds', onTimeout: () => 'boot hung' as const })), + ), + ) + expect(Exit.isSuccess(exit)).toBe(true) +}) + +// ─── a listen-only seat catches up at boot (comms-9iro) ────────────────────── +// The defect: a seat that only listens never posts, so under deferred identity +// it never acquired, so the REST catch-up gated behind acquire never ran and its +// downtime backlog was lost. Both halves dissolve once subscribing binds — the +// seat acquires at boot, and the hook that carries catch-up fires there. Pinned +// here because this landing rewrites that boot-bind ordering: the property is +// held by the order of steps, and nothing else would notice it changing. +test('a listen-only seat runs its catch-up at boot, with zero tool calls', async () => { + const binderRef = await Effect.runPromise(Ref.make>(Option.none())) + const adapter = await Effect.runPromise(memoryAdapter({ bindOnDemand: bindThrough(binderRef) })) + const caughtUpChannels: string[] = [] + const substrate = completeAsSubstrate( + { + ...adapter, + inbox: { ...adapter.inbox, events: () => Stream.empty }, + history: { + ...adapter.history, + readChannel: (channel: ChannelName, _range: Range) => + Effect.sync(() => { + caughtUpChannels.push(channel as string) + return [] + }), + }, + }, + { close: async () => {} }, + ) + await runProgram( + { ...lazyEnv, COMMY_SUBSCRIBE: 'home' }, + substrate, + { + loggerLayer: captureLogger([]), + readGitContext: () => Effect.succeed(NotInRepo()), + }, + binderRef, + ) + // The channel catch-up skimmed the seat's boot-time narrow without a single + // tool call: COMMY_SUBSCRIBE bound the seat at boot, and the post-acquire hook + // that carries catch-up ran there. + expect(caughtUpChannels).toEqual(['home']) +}) + test('main aborts non-zero when COMMY_SUBSCRIBE contains a malformed token', async () => { const fake = buildFakeAdapter() const env = { diff --git a/packages/mcp/server.ts b/packages/mcp/server.ts index 2dadb76..78f8fc0 100644 --- a/packages/mcp/server.ts +++ b/packages/mcp/server.ts @@ -26,7 +26,7 @@ import { Predicate, Schedule, } from 'effect' -import type { BotName, GitContext, ProjectSlug, SessionId } from './bootstrap.ts' +import type { BotName, GitContext, ParsedEnv, ProjectSlug, SessionId } from './bootstrap.ts' import { readGitContext as defaultReadGitContext, deriveProject, @@ -49,7 +49,7 @@ import { buildMcpServer } from './mcp-server.ts' import { type CatchUpError, catchUpMentions } from './mentions-catch-up.ts' import type { NarrowSet } from './narrow-set.ts' import { createNarrowSet } from './narrow-set.ts' -import { FileQueueStateStoreLive, type QueueStateStoreTag } from './queue-state-store.ts' +import { FileQueueStateStoreLive, QueueStateStoreTag } from './queue-state-store.ts' import { raceReleaseAgainstTimeout } from './release-shutdown.ts' import { ResumeOutcomeLive, ResumeOutcome as ResumeOutcomeTag } from './resume-outcome.ts' import { @@ -62,6 +62,7 @@ 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' +import type { PersistedTopicIntents } from './subscription-restore.ts' import { persistSubscriptions, restoreSubscriptions, @@ -263,6 +264,83 @@ const registerType1DefaultsOnBoot = ( ), ) +interface SeedDeps { + readonly inbox: MessageInbox + readonly narrowSet: NarrowSet + readonly parsed: ParsedEnv + readonly loggerLayer: Layer.Layer +} + +/** + * Apply `COMMY_SUBSCRIBE` to a bot exactly once in its life, at the mint. + * + * Graeme's ruling (2026-07-20): "nix is only bootstrapping the bot account — + * the bot owns it once it exists." So the tokens seed a bot's subscriptions and + * are then never consulted again — not on resume, not on reconnect. From the + * seeding onwards the subscriptions live in the realm under the seat's own + * principal and change only when the agent itself subscribes or unsubscribes. + * Editing `COMMY_SUBSCRIBE` for a bot that already exists does nothing, BY + * DESIGN: that cross-session ambiguity — the same value read at two launches of + * one bot, indistinguishable from the bot's own runtime changes — is precisely + * what is being removed. + * + * Called wherever the mint actually happens, which differs by mode and is why + * this takes an already-acquired identity rather than binding for itself. A + * persistent bot mints at the boot acquire. An ephemeral seat mints on its first + * bind, which is at boot when `COMMY_SUBSCRIBE` gives it something to hold and + * otherwise on its first attribution-producing call — so a seat whose session id + * had not arrived by boot still gets seeded, rather than half-seeded into a + * client-side filter over a stream it is not subscribed to. + * + * The ORIGIN of the bind is the whole test, and the realm answers it as a + * by-product of acquiring — no local bookkeeping is consulted, and none is + * kept. An already-existing bot is never seeded, whatever it is or is not + * subscribed to. Graeme's ruling (2026-07-31) settled the one case that looked + * like it needed more than this: a bot that comes out of an upgrade with no + * subscriptions of its own is LEFT empty rather than bootstrapped, because a + * bot that wants subscriptions can subscribe — which is the same charter read + * from the other end. That removes the only reason to record what has been + * seeded before, so nothing is recorded. + * + * Both diagnostics live here rather than at the call sites, so the two mint + * paths cannot drift into saying different things about the same decision. + */ +const seedSubscriptionsOnMint = ( + acquired: AcquiredIdentity, + deps: SeedDeps, +): Effect.Effect, SubscribeTokenError | BindError | InboxError> => + deps.parsed.subscribe === undefined + ? Effect.succeed([]) + : acquired.origin === 'existing' + ? // The one positive trace that the value was read and deliberately not + // applied. It is what the old "no applied line means you have found a + // clobbered COMMY_SUBSCRIBE" diagnostic becomes: absence no longer + // carries that meaning, because a bot past its bootstrap never emits + // the applied line again. + Effect.logInfo( + `commy plugin: COMMY_SUBSCRIBE not applied — ${acquired.identity.name} already ` + + `exists and owns its subscriptions. Editing COMMY_SUBSCRIBE for a bot that ` + + `already exists has no effect; change its subscriptions through the bot.`, + ).pipe(Effect.provide(deps.loggerLayer), Effect.as>([])) + : subscribeFromEnv(deps.inbox, deps.narrowSet, deps.parsed).pipe( + // Deliberately silent on an empty set, rather than warning. The + // process cannot distinguish a seat that wanted no subscriptions from + // one whose value was destroyed upstream — the operator's intent is + // gone by the time the value arrives here — so an empty-set warning + // would fire on the majority of perfectly healthy interactive boots + // while telling the one broken seat nothing it could act on. A line + // that is almost always noise trains the reader to skip it, which is + // how the next silent fault gets to hide. The invariant that stops the + // clobber recurring is pinned in the launcher manifest test, not here. + Effect.tap((intents) => + intents.length === 0 + ? Effect.void + : Effect.logInfo( + `commy plugin: applied ${intents.length} boot-time subscribe target(s): ${intents.map(intentToToken).join(', ')}`, + ).pipe(Effect.provide(deps.loggerLayer)), + ), + ) + /** * Default boot-time channel/thread catch-up window for persistent bots. * 4 hours covers overnight downtime without flooding the @@ -351,6 +429,7 @@ export const makeProgram = ( | SubstrateAdapter | CursorStoreTag | SubscriptionStoreTag + | QueueStateStoreTag | SessionIdTag | ResumeOutcomeTag | SessionBinderTag @@ -380,6 +459,7 @@ export const makeProgram = ( // surviving queue replayed the backlog); awaited once per session in the // ephemeral onAcquire hook below. const resumeOutcome = yield* ResumeOutcomeTag + const queueStateStore = yield* QueueStateStoreTag yield* Effect.flatMap( readBootSessionId, Option.match({ @@ -387,6 +467,57 @@ export const makeProgram = ( onSome: (sessionId) => Deferred.succeed(sessionIdDeferred, sessionId).pipe(Effect.asVoid), }), ) + // Report "there is nothing to resume" HERE, at the earliest point the + // answer is knowable, rather than leaving it to the producer alone. + // + // This closes a boot deadlock, not a diagnostic gap. The ephemeral + // `onAcquire` below BLOCKS on this verdict, and since subscriptions moved + // to the seat's own principal the boot-time subscribe binds — so + // `onAcquire` runs on the BOOT fiber. The producer that would otherwise + // complete the verdict is materialised by `startEventPump`, the last step + // of boot, which the parked boot fiber never reaches. Every ephemeral seat + // with `COMMY_SUBSCRIBE` set hangs before announcing its tools. + // + // An absent queue-state is a complete answer on its own: with nothing + // persisted there is no surviving queue, so no replay can happen and the + // catch-up fallback is the right path. A PRESENT queue-state is NOT an + // answer — whether that queue is still alive is only knowable by polling + // it — so this stays silent there and the producer reports as before. The + // deferred is first-write-wins, so the producer's own report is a harmless + // no-op whenever this one fired first. + // + // Reading the store also cannot clobber a resume candidate: + // `onQueueRegister`'s guard treats a completed verdict as licence to + // replace the persisted state wholesale, but this only completes it in the + // branch where there was nothing persisted to protect. + const reportNothingToResume: Effect.Effect = + parsed.botName !== undefined + ? Effect.void + : Deferred.poll(sessionIdDeferred).pipe( + Effect.flatMap( + Option.match({ + // No session id at boot means no bind can happen at boot + // either (the cache hands back its unbound stub), so nothing + // can park on the verdict and the producer keeps it. + onNone: () => Effect.void, + onSome: (awaitId) => + Effect.flatMap(awaitId, (sessionId) => + queueStateStore.read(sessionId).pipe( + Effect.map(Option.isNone), + // An unreadable store is treated as fresh, matching + // `resumeQueue`'s own best-effort degrade. + Effect.catchAll(() => Effect.succeed(true)), + Effect.flatMap((nothingPersisted) => + nothingPersisted + ? Deferred.succeed(resumeOutcome, false).pipe(Effect.asVoid) + : Effect.void, + ), + ), + ), + }), + ), + ) + yield* reportNothingToResume // Per-tool-call feeder: every PreToolUse-stamped call // (post/edit_message/react/unreact/current_identity) hands its session_id // here. Idempotent — after the first writer the rest return false. @@ -500,44 +631,61 @@ export const makeProgram = ( ? createType2DefaultsOnAcquire(narrowSet, adapter.inbox) : undefined - // Reactive subscription restore. Restore is a reaction to the - // session_id becoming known, not a thing a specific action triggers: a host - // that does not inject the session id into the MCP child's env boots - // session-blind, and there the id cannot arrive until the seat itself acts. - // `restoreSubscriptions` reads the session-bound store, whose `read` awaits - // the shared session-id `Deferred` internally — so this is forked ONCE into - // the connected runtime below and parks on that read until any source (the - // boot-env feeder, which covers a Claude Code seat whether fresh or - // resumed, or the first tool call of an acting seat) fills the id, then - // rehydrates with zero agent action. Must be forked, not awaited inline: an - // inline await would block boot/serving until the id lands. The store's - // presence stays a true resume signal; a corrupt or unreadable store logs - // 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. + // Rebuild this seat's narrow set from the realm, narrowed by whatever + // topic-level intents were recorded for it. See `restoreSubscriptions` for + // why those two sources and no others. // - // 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 - : 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)}`, - ), - ), - Effect.provide(loggerLayer), + // Runs for BOTH modes, which is new. A pinned pane has no session record — + // its session id changes every launch — so before this it came up with + // nothing but its boot-time defaults while the realm went on delivering + // every channel it had ever joined at runtime. That seat is the one this + // fixes; the ephemeral seat merely stops being a special case. + // + // Forked, never awaited inline. The ephemeral record's read parks on the + // shared session id, and a host that injects none leaves that id + // unarrived until the seat itself acts — an inline await would hold boot + // there. A pinned pane reads no record at all and so would not park, but + // it still forks: it re-subscribes, which binds, and nothing on the boot + // fiber should wait on a substrate round-trip it does not need. + // + // Re-subscribing is realm state under the seat's own principal, so the + // seat's naming inputs have to be in context for the bind to resolve. + const persistedTopicIntents: PersistedTopicIntents = + parsed.botName === undefined + ? subscriptionStore.read() + : // A pinned bot's record could never be its own: the store is keyed by + // session id and its id is new every launch. Reading one would park + // on an id no host supplies, for an answer that would not be about + // this bot anyway. + Effect.succeedNone + const rebuildNarrowSet: Effect.Effect = ( + parsed.botName === undefined + ? Deferred.await(sessionIdDeferred).pipe( + Effect.map((sessionId): SessionId | undefined => sessionId), ) + : Effect.succeed(undefined) + ).pipe( + Effect.flatMap((sessionId) => + withSessionContext( + restoreSubscriptions({ + persisted: persistedTopicIntents, + isBound: () => identityCache.boundIdentityIds().size > 0, + narrowSet, + inbox: adapter.inbox, + }), + { sessionId, project: parsed.project }, + ), + ), + Effect.catchAll((err) => + // Load anyway on failure, with no base: the narrow set is buffering + // and something has to end that window, or a seat that could not + // reach the realm journals deltas nothing ever replays. + Effect.logError( + `commy plugin: could not rebuild the subscription set: ${Predicate.isError(err) ? err.message : String(err)}`, + ).pipe(Effect.zipRight(Effect.sync(() => narrowSet.load(Option.none())))), + ), + Effect.provide(loggerLayer), + ) const seedDefaults = (project: ProjectSlug | undefined): Effect.Effect => registerType2Defaults !== undefined ? registerType2Defaults(project) : Effect.void // Seed the acquire-gated Type-2 defaults for a fresh session (store absent). @@ -595,7 +743,17 @@ export const makeProgram = ( ), ) - // Ephemeral-mode post-acquire hook: restore (or + // The COMMY_SUBSCRIBE bootstrap, ready for whichever path mints this + // seat's bot. See `seedSubscriptionsOnMint`. + const seedDeps: SeedDeps = { + inbox: adapter.inbox, + narrowSet, + parsed, + loggerLayer, + } + + // Ephemeral-mode post-acquire hook: seed this bot's `COMMY_SUBSCRIBE` + // bootstrap if this is its mint, restore (or // seed) the narrow set on the first action of this session_id, then — // ONLY when the queue-resume failed — replay missed history. The cache // runs this via ensure-bound's own runtime edge; the composed Effect @@ -618,7 +776,19 @@ export const makeProgram = ( project: ProjectSlug | undefined, sessionId: SessionId, ): Effect.Effect => - ensureSessionSubscriptions(sessionId, project).pipe( + // The bootstrap runs FIRST and its failures are logged rather + // than raised: this hook rides the caller's acquire, and a + // substrate hiccup applying a boot-time token must not refuse the + // tool call that happened to trigger the mint. It has to precede + // the catch-up below, which skims the narrow set this seeds. + seedSubscriptionsOnMint(acquired, seedDeps).pipe( + Effect.catchAll((err) => + Effect.logError( + `commy plugin: COMMY_SUBSCRIBE seeding failed at mint: ${Predicate.isError(err) ? err.message : String(err)}`, + ), + ), + Effect.provide(loggerLayer), + Effect.zipRight(ensureSessionSubscriptions(sessionId, project)), Effect.zipRight(Deferred.await(resumeOutcome)), Effect.flatMap((queueReplayed) => { if (queueReplayed) return Effect.void @@ -669,6 +839,11 @@ export const makeProgram = ( // The single-identity cache ignores the session_id. // Ephemeral mode: skip — the first tool call mints lazily. let type1Intents: ReadonlyArray = [] + // The persistent bot's boot acquire, held so the COMMY_SUBSCRIBE + // bootstrap below can read the origin of the bind that just happened. + // Persistent mode has no post-acquire hook to hang the seeding off, so + // this is where its mint is observable. + let persistentIdentity: AcquiredIdentity | undefined if (parsed.botName !== undefined) { const botName = parsed.botName const ensureBound = yield* identityCache.ensureBoundFor(undefined) @@ -689,7 +864,14 @@ export const makeProgram = ( ).pipe(Effect.zipRight(Effect.fail(bootErr))), // Type-1 defaults: post-acquire register the // universal `mentions` narrow plus project-specific subs. - onSuccess: () => registerType1DefaultsOnBoot(adapter.inbox, narrowSet, parsed.project), + onSuccess: (acquired) => + Effect.sync(() => { + persistentIdentity = acquired + }).pipe( + Effect.zipRight( + registerType1DefaultsOnBoot(adapter.inbox, narrowSet, parsed.project), + ), + ), }), ) } @@ -737,54 +919,51 @@ export const makeProgram = ( }), ), ) - // 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. + // Two mint paths, one bootstrap, and boot's job differs between them. + // + // PERSISTENT: the eager acquire above IS the mint, and there is no + // post-acquire hook, so the seeding happens here — after the Type-1 + // defaults, which are a separate, per-boot concern that this bead does not + // touch (they are computed from the project slug, not read from + // COMMY_SUBSCRIBE, and a pane depends on them being re-registered every + // boot). + // + // EPHEMERAL: the mint runs `ephemeralOnAcquire`, which seeds. All boot has + // to do is TRIGGER that mint, and only when COMMY_SUBSCRIBE gives the seat + // something to hold — receiving needs a queue and a queue needs a + // principal, so a seat asked to hold subscriptions needs an identity at + // boot. A seat asked to hold nothing must NOT mint here; it stays + // identity-free until it acts. const subscribedIntents = yield* withSessionContext( - subscribeFromEnv(adapter.inbox, narrowSet, parsed), + persistentIdentity !== undefined + ? seedSubscriptionsOnMint(persistentIdentity, seedDeps) + : parsed.subscribe === undefined + ? Effect.succeed>([]) + : binderFor(identityCache).pipe(Effect.as>([])), { sessionId: Option.getOrUndefined(bootSessionId), project: parsed.project }, ).pipe( + // 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. The seeding is not lost with it: an + // ephemeral seat that could not bind at boot still seeds when its first + // action mints it. 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)}`, + `subscriptions were applied yet — this seat will not receive channel traffic ` + + `until it acts. ${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 - // COMMY_SUBSCRIBE cost months: an operator who configures subscriptions - // and then finds no line here — or a line that does not name the tokens - // they set — has the fault in front of them, which is more than the empty - // case could ever have given them. - // - // Deliberately silent when the set is empty, rather than warning. The - // process cannot distinguish a seat that wanted no subscriptions from one - // whose value was destroyed upstream — the operator's intent is gone by - // the time the value arrives here — so an empty-set warning would fire on - // the majority of perfectly healthy interactive boots while telling the - // one broken seat nothing it could act on. A line that is almost always - // noise trains the reader to skip it, which is how the next silent fault - // gets to hide. The invariant that stops the clobber recurring is pinned - // in the launcher manifest test, not here. - if (subscribedIntents.length > 0) { - yield* Effect.logInfo( - `commy plugin: applied ${subscribedIntents.length} boot-time subscribe target(s): ${subscribedIntents.map(intentToToken).join(', ')}`, - ).pipe(Effect.provide(loggerLayer)) - } - - // Ephemeral resume: start journaling runtime subscribe/unsubscribe deltas - // now — after the COMMY_SUBSCRIBE base is seeded but before any tool call - // can mutate — so a delta racing the boot-forked restore below is replayed - // onto the restored base rather than clobbered by it. Persistent mode never - // restores, so it never buffers: add/remove apply directly. - if (parsed.botName === undefined) { - narrowSet.beginBuffering() - } + // Start journaling runtime subscribe/unsubscribe deltas now — after the + // boot-time defaults and any COMMY_SUBSCRIBE bootstrap have landed, but + // before a tool call can mutate — so a delta racing the boot-forked + // rebuild below is replayed onto its result rather than clobbered by it. + // Both modes buffer, because both now rebuild. + narrowSet.beginBuffering() const toolsCache = registerTools(mcp, { adapter, @@ -913,14 +1092,14 @@ export const makeProgram = ( yield* forkIdleSweep(identityCache, EPHEMERAL_IDLE_SWEEP_INTERVAL_MS) } - // Boot-forked subscription restore, scope-tied like the sweep. Parks on the - // store's `Deferred.await` until any source fills the shared session-id, - // then loads the persisted base (resume) or drains the buffer over the env - // seed (fresh); either way it replays the deltas journaled since - // `beginBuffering`, so a subscribe racing the load is never lost. A no-op - // Effect in persistent mode. Never awaited inline: an unfilled id on a - // listen-only seat would otherwise block the scope forever. - yield* Effect.forkScoped(restoreOnResume) + // Boot-forked narrow-set rebuild, scope-tied like the sweep. Asks the + // realm what this seat is subscribed to and narrows the answer with its + // recorded topic intents, then replays the deltas journaled since + // `beginBuffering` so a subscribe racing the load is never lost. Never + // awaited inline: an ephemeral seat's record parks on the shared session + // id, and an unfilled id on a listen-only seat would otherwise block the + // scope forever. + yield* Effect.forkScoped(rebuildNarrowSet) // Block until either the event stream ends / the pump fatally parks // and is interrupted (the SIGINT/SIGTERM path), OR the MCP client @@ -1014,6 +1193,7 @@ export const MainLive: Layer.Layer< | SubstrateAdapter | CursorStoreTag | SubscriptionStoreTag + | QueueStateStoreTag | SessionIdTag | ResumeOutcomeTag | SessionBinderTag diff --git a/packages/mcp/subscription-restore.test.ts b/packages/mcp/subscription-restore.test.ts index 3b902ff..48ecfd8 100644 --- a/packages/mcp/subscription-restore.test.ts +++ b/packages/mcp/subscription-restore.test.ts @@ -84,8 +84,12 @@ describe('seedDefaultsIfFresh', () => { }) }) +// Only the topic-level slice is written. Channel-wide intents are the realm's +// to hold — it has a subscription row for each — and a second copy here could +// only ever disagree with the rows that actually govern delivery. What the realm +// CANNOT hold is narrowing below a channel, and that is precisely what stays. describe('persistSubscriptions', () => { - test('writes the current narrow-set snapshot to the session-bound store', async () => { + test('writes the topic-level slice of the narrow set, not the channel-wide intents', async () => { const narrowSet = createNarrowSet() narrowSet.add(newTopics('general')) narrowSet.add(channel('commy')) @@ -95,8 +99,20 @@ describe('persistSubscriptions', () => { } await Effect.runPromise(persistSubscriptions(store, narrowSet)) expect(written.length).toBe(1) - expect(sortIntents(written[0] ?? [])).toEqual( - sortIntents([newTopics('general'), channel('commy')]), - ) + expect(sortIntents(written[0] ?? [])).toEqual(sortIntents([newTopics('general')])) + }) + + test('an empty topic slice is still written, so the record stays a true has-run-before signal', async () => { + // `seedDefaultsIfFresh` gates on the record's PRESENCE. A seat whose narrows + // are all channel-wide must still leave one behind, or its next launch would + // read as a first launch and re-seed defaults it had dropped. + const narrowSet = createNarrowSet() + narrowSet.add(channel('commy')) + const written: ReadonlyArray[] = [] + const store: Pick = { + write: (intents) => Effect.sync(() => void written.push(intents)), + } + await Effect.runPromise(persistSubscriptions(store, narrowSet)) + expect(written).toEqual([[]]) }) }) diff --git a/packages/mcp/subscription-restore.ts b/packages/mcp/subscription-restore.ts index e007254..40c8c3d 100644 --- a/packages/mcp/subscription-restore.ts +++ b/packages/mcp/subscription-restore.ts @@ -1,64 +1,128 @@ -import type { BindError, InboxError, MessageInbox } from '@commy/core/ports' +import type { BindError, ChannelName, InboxError, MessageInbox } from '@commy/core/ports' import type { PlatformError } from '@effect/platform/Error' -import { Effect, Option, type ParseResult } from 'effect' +import { Array as Arr, Effect, HashSet, Option, type ParseResult } from 'effect' import type { ProjectSlug } from './bootstrap.ts' import type { NarrowSet } from './narrow-set.ts' import { intentToTarget, type SubscribeIntent } from './subscribe-parser.ts' import type { SubscriptionStore } from './subscription-store.ts' /** - * Restore and seed are two independent reactions to a session's narrow set, - * split apart for the reactive core. Resume vs fresh is decided by - * the per-session_id subscription store, exactly as the mentions cursor decides - * resume vs fresh for catch-up: - * - store present → resume: {@link restoreSubscriptions} loads the persisted - * intents as the narrow set's base (including prior unsubscribes — a dropped - * default stays dropped) and wires each on the substrate. Runtime - * subscribe/unsubscribe deltas that raced the still-loading restore were - * journaled by the narrow set and replay onto this base in arrival order, so - * none is lost. An empty persisted set is honoured verbatim: the session - * resumes hearing only what it re-subscribed since boot. The store's `read` - * awaits the shared session-id `Deferred` internally, so restore is forked - * once at boot and blocks there until any source fills the id — no - * session_id threaded through its signature, no per-session memo, no gate on - * the mutation path. - * - store absent → fresh: {@link seedDefaultsIfFresh} registers the - * acquire-gated Type-2 defaults, leaving whatever `COMMY_SUBSCRIBE` seeded at - * boot in place. Store-gated and restore-free, so it never rehydrates a - * resumed set and needs no once-per-session memo. + * Rebuilding a seat's narrow set on the way up, from the realm plus the one + * thing the realm cannot hold. + * + * The realm is the authority on what a seat is subscribed to: those rows are + * written under the seat's own principal and they are what actually governs + * delivery. So on the way up a seat ASKS — `inbox.subscriptions()` — rather than + * replaying a local copy of what it once asked for. A local copy can disagree + * with delivery; the realm's answer cannot. + * + * What a subscription row cannot express is narrowing BELOW a channel, so + * `#chan/topic` and `new-topics:#chan` both read back as plain `#chan`. + * Reconstructing from the realm alone would silently widen every topic + * subscription into its whole channel. That is why a small local record of + * TOPIC-LEVEL intents survives — not as authority over what the seat receives, + * but as the record of how narrowly it wanted to listen. + * + * That record is INTERIM, and `docs/agent-experience.md` is careful about why: + * Zulip does have somewhere to put topic intent (`POST /user_topics` with + * `visibility_policy: FOLLOWED` persists a per-`(user, stream, topic)` row), it + * just does not DELIVER against it, so a client still filters its own queue. + * Recording intent there would make this record a cache of realm truth instead + * of the last piece of client-side authority. Deliberately not done here: this + * change removes state, and adding a new realm write is a different kind of + * change with its own risks. + * + * The two compose without overlapping: every channel the realm reports becomes + * a channel-wide narrow UNLESS a persisted topic intent names it, in which case + * the topic intents stand in its place. */ export interface SubscriptionRestoreDeps { readonly subscriptionStore: Pick readonly narrowSet: NarrowSet - readonly inbox: Pick + readonly inbox: Pick readonly registerDefaults: (project: ProjectSlug | undefined) => Effect.Effect } +/** + * The topic-level record a rebuild narrows the realm's answer with. Supplied by + * the caller rather than read here, because where it comes from differs by + * seat: an ephemeral seat has a per-session record on disk, while a pinned bot + * has none at all — its session id changes every launch, so a session-keyed + * record could never be its own. Passing it in keeps the pinned seat off a read + * that would park forever waiting for an id no host supplies. + */ +export type PersistedTopicIntents = Effect.Effect< + Option.Option>, + PlatformError | ParseResult.ParseError +> + +/** The intents a topic-level record holds: everything narrower than a channel. */ +export const isTopicLevel = (intent: SubscribeIntent): boolean => intent.kind !== 'channel' + +/** + * The subset of a narrow set worth persisting. Channel-wide intents are dropped + * because the realm already holds them — persisting them would re-create the + * divergence this design removes, where a local copy and the delivery rules + * disagree and the local copy wins. + */ +export const topicLevelIntents = ( + intents: ReadonlyArray, +): ReadonlyArray => Arr.filter(intents, isTopicLevel) + +const sameIntent = (a: SubscribeIntent, b: SubscribeIntent): boolean => + a.kind === b.kind && + a.channelName === b.channelName && + (a.kind !== 'thread' || b.kind !== 'thread' || a.threadName === b.threadName) + +const combineWithRealm = ( + realmChannels: ReadonlyArray, + topicIntents: ReadonlyArray, +): ReadonlyArray => { + const deduped = Arr.dedupeWith(topicIntents, sameIntent) + const narrowed = HashSet.fromIterable(deduped.map((intent) => intent.channelName)) + const channelIntents = Arr.filterMap(Arr.dedupe(realmChannels), (channelName) => + HashSet.has(narrowed, channelName) + ? Option.none() + : Option.some({ kind: 'channel', channelName }), + ) + return [...channelIntents, ...deduped] +} + const applyRestored = ( deps: Pick, intents: ReadonlyArray, ): Effect.Effect => - Effect.sync(() => deps.narrowSet.load(Option.some(intents))).pipe( - Effect.zipRight( - Effect.forEach(intents, (intent) => deps.inbox.subscribe(intentToTarget(intent)), { - discard: true, - }), - ), - ) + Effect.suspend(() => { + // Declared on the substrate as well as loaded locally, because the adapter + // holds per-process state of its own — which channels it is listening to, + // which ones carry a new-topics narrow — and a fresh process starts with + // none. Only what this process has NOT already declared: the boot-time + // defaults and any bootstrap went through `subscribe` on the way in, and + // re-declaring them would be a second round-trip for an answer the adapter + // already has. + const alreadyDeclared = deps.narrowSet.intents() + const undeclared = Arr.filter( + intents, + (intent) => !alreadyDeclared.some((declared) => sameIntent(declared, intent)), + ) + return Effect.sync(() => deps.narrowSet.load(Option.some(intents))).pipe( + Effect.zipRight( + Effect.forEach(undeclared, (intent) => deps.inbox.subscribe(intentToTarget(intent)), { + discard: true, + }), + ), + ) + }) /** - * Seed the acquire-gated Type-2 defaults, but only for a fresh session (store - * absent). The seed half of the old restore-or-seed, split out for the reactive - * core: restore is now a boot-forked reaction to the session_id via - * {@link restoreSubscriptions}, leaving seeding as its own store-gated step. The - * deps carry no `narrowSet`/`inbox`, so this structurally CANNOT restore — a - * resumed session (store present, even an empty persisted set) gets nothing from - * here; its set is rehydrated by the restore reaction instead. + * Seed the acquire-gated Type-2 defaults, but only for a session this + * installation has not run before (store absent). The store's presence is the + * guard, and once the session subscribes anything the persist write makes it + * present — so a later call is a no-op and no once-per-session memo is needed. * - * Idempotent by construction: the store's presence is the guard, and once the - * session subscribes anything the persist write makes the store present, so a - * later call is a no-op. This is why the reactive core needs no `restoredSessions` - * memo for the seed — the store gates it. + * Store PRESENCE is the only thing read here; its contents are not consulted. + * That distinction is what lets it stay a true "has this session run before?" + * signal even though the record itself shrank to topic-level intents. */ export const seedDefaultsIfFresh = ( deps: Pick, @@ -74,44 +138,69 @@ export const seedDefaultsIfFresh = ( ) /** - * Restore a resuming session's narrow set — the resume half of the old - * restore-or-seed, with the fresh-session seed deliberately omitted (that is - * {@link seedDefaultsIfFresh}'s job). On a store hit it loads the persisted - * intents as the narrow set's base and wires each on the substrate; an empty - * persisted set is honoured verbatim. On a store miss it still loads — with no - * base — to end the narrow set's buffering window, leaving the boot-time - * `COMMY_SUBSCRIBE` seed standing as the fresh-session set. Either way the - * narrow set replays whatever runtime deltas it journaled while restore was - * loading, so a subscribe/unsubscribe that raced the load is never lost. + * Rebuild this seat's narrow set: ask the realm what it is subscribed to, and + * narrow the answer with whatever topic-level intents were recorded for this + * session. + * + * Runs for every seat, not just a resuming one. A pinned pane has no session + * record at all — its session id changes every launch — so before this it came + * up with nothing but its boot-time defaults, and any channel it had joined at + * runtime was silently dropped even though the realm was still delivering it. + * Asking the realm covers that seat for the first time. * - * Restore is a reaction to the session_id becoming known, not a thing a specific - * action triggers: a host that does not inject the session id into the MCP - * child's env boots session-blind, and there the id cannot arrive until the seat - * itself acts. The store's `read` awaits the shared session-id `Deferred` - * internally, so this is forked once at boot (server.ts) and parks on that read - * until any source — the boot-env feeder, which covers a Claude Code seat whether - * fresh or resumed, or the first tool call of an acting seat — fills the id, then - * rehydrates. The `Deferred`'s single completion is the once-guard; no session_id - * threads through this signature and no per-session memo is needed. + * Forked at boot rather than awaited: the topic record is session-keyed and its + * read parks until the session id arrives, which on a seat whose host injects + * no id means until the seat itself acts. Whatever the outcome, the narrow set + * is loaded exactly once so the buffering window closes and the deltas + * journaled since boot replay onto the result — a subscribe that raced the + * rebuild is never lost. */ export const restoreSubscriptions = ( - deps: Pick, + deps: Pick & { + readonly persisted: PersistedTopicIntents + /** + * Whether this seat already holds an identity. Read passively — asking must + * never be the thing that mints one. + */ + readonly isBound: () => boolean + }, ): Effect.Effect => - deps.subscriptionStore.read().pipe( - Effect.flatMap( - Option.match({ - onNone: () => Effect.sync(() => deps.narrowSet.load(Option.none())), - onSome: (intents) => applyRestored(deps, intents), - }), + deps.persisted.pipe( + Effect.flatMap((persisted) => + // Reading the realm needs this seat's credential, so it can only be asked + // by a seat that has one. A seat with no identity has no subscription + // rows either — the honest answer is "nothing", not "mint a bot and go + // and look". A topic record IS grounds to bind: it can only exist because + // this session ran before, so the bot exists and binding re-acquires it + // rather than bringing a new one into being. + Option.isNone(persisted) && !deps.isBound() + ? Effect.sync(() => deps.narrowSet.load(Option.none())) + : deps.inbox.subscriptions().pipe( + Effect.flatMap((realmChannels) => + applyRestored( + deps, + combineWithRealm(realmChannels, [ + ...topicLevelIntents(Option.getOrElse(persisted, () => [])), + // Topic intents this boot has already registered — the + // boot-time defaults, and a fresh bot's COMMY_SUBSCRIBE + // bootstrap. Without these the realm's channel-wide answer + // would REPLACE them and quietly widen a `new-topics:` or + // single-topic narrow into its whole channel. + ...topicLevelIntents(deps.narrowSet.intents()), + ]), + ), + ), + ), ), ) /** - * Persist the current narrow-set snapshot under the session_id. Called after - * every runtime `subscribe`/`unsubscribe` mutation so a later resume restores - * the exact set — defaults included, unsubscribes included. + * Persist the topic-level slice of the current narrow set under the session id, + * after every runtime `subscribe`/`unsubscribe`. Channel-wide intents are + * deliberately absent: the realm holds those, and a second copy could only + * disagree with it. */ export const persistSubscriptions = ( store: Pick, narrowSet: NarrowSet, -): Effect.Effect => store.write(narrowSet.intents()) +): Effect.Effect => store.write(topicLevelIntents(narrowSet.intents())) diff --git a/packages/mcp/subscription-resume.test.ts b/packages/mcp/subscription-resume.test.ts index 860495a..d3d6845 100644 --- a/packages/mcp/subscription-resume.test.ts +++ b/packages/mcp/subscription-resume.test.ts @@ -1,5 +1,5 @@ import { expect, test } from 'bun:test' -import type { InboundEvent, MessageRef, SubscriptionTarget } from '@commy/core/ports' +import type { InboundEvent, MessageInbox, MessageRef, SubscriptionTarget } from '@commy/core/ports' import { ChannelPermalinkSchema, decodeChannelIdSync, @@ -63,16 +63,24 @@ const channelIntent = (name: string): SubscribeIntent => ({ const sortIntents = (intents: ReadonlyArray): ReadonlyArray => [...intents].sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b))) -// A substrate-subscribe spy: restore wires each restored intent on the substrate -// via `inbox.subscribe`, mirroring the real adapter's side effect. -const spyInbox = (): { - readonly inbox: { subscribe: (target: SubscriptionTarget) => Effect.Effect } +// A stand-in realm: it answers `subscriptions()` with the channels it holds +// rows for, and records the re-declarations the rebuild makes through +// `subscribe`. `realmChannels` is what this seat was subscribed to before the +// reboot — the thing a rebuild now asks for instead of replaying a local copy. +const spyInbox = ( + realmChannels: ReadonlyArray = [], +): { + readonly inbox: Pick readonly subscribes: ReadonlyArray } => { const subscribes: SubscriptionTarget[] = [] return { subscribes, - inbox: { subscribe: (target) => Effect.sync(() => void subscribes.push(target)) }, + inbox: { + subscribe: (target) => Effect.sync(() => void subscribes.push(target)), + unsubscribe: () => Effect.void, + subscriptions: () => Effect.succeed(realmChannels.map((name) => decodeChannelNameSync(name))), + } as Pick, } } @@ -138,7 +146,9 @@ test('restore rehydrates the persisted narrow set when the shared deferred is fi Effect.gen(function* () { const narrowSet = createNarrowSet() narrowSet.beginBuffering() - const { inbox, subscribes } = spyInbox() + // The realm delivers both channels; the topic record is what keeps the + // rebuild from widening them into channel-wide narrows. + const { inbox, subscribes } = spyInbox([DECISIONS_CHANNEL, 'general']) const session = yield* Deferred.make() const subscriptionStore = inMemorySubscriptionStore( session, @@ -147,7 +157,12 @@ test('restore rehydrates the persisted narrow set when the shared deferred is fi // Boot-fork restore: it parks on the store read, which awaits the deferred. const fiber = yield* Effect.fork( - restoreSubscriptions({ subscriptionStore, narrowSet, inbox }), + restoreSubscriptions({ + persisted: subscriptionStore.read(), + isBound: () => true, + narrowSet, + inbox, + }), ) // Rebooted, deaf: nothing restored while the id is unknown. @@ -176,10 +191,11 @@ test('deltas racing the load are journaled and replayed onto the restored base', Effect.runPromise( Effect.gen(function* () { const narrowSet = createNarrowSet() - // A COMMY_SUBSCRIBE default the persisted set turns out to have dropped. + // A COMMY_SUBSCRIBE default this seat has since unsubscribed from: the + // realm holds no row for it, so the rebuild does not bring it back. narrowSet.add(channelIntent('env-default')) narrowSet.beginBuffering() - const { inbox } = spyInbox() + const { inbox } = spyInbox([DECISIONS_CHANNEL, 'general']) const session = yield* Deferred.make() const subscriptionStore = inMemorySubscriptionStore( session, @@ -187,7 +203,12 @@ test('deltas racing the load are journaled and replayed onto the restored base', ) const fiber = yield* Effect.fork( - restoreSubscriptions({ subscriptionStore, narrowSet, inbox }), + restoreSubscriptions({ + persisted: subscriptionStore.read(), + isBound: () => true, + narrowSet, + inbox, + }), ) // Before the id lands: subscribe something new, unsubscribe a persisted sub. @@ -205,16 +226,18 @@ test('deltas racing the load are journaled and replayed onto the restored base', }), )) -// A fresh session (store miss) loads no base: the COMMY_SUBSCRIBE seed stands and -// buffered deltas replay onto it. Restoring nothing is the whole point — a -// never-seen session must not inherit another session's persisted set. -test('a fresh session (store miss) keeps the env seed and replays buffered deltas', () => +// A fresh session has no topic record of its own, so the realm's answer is the +// whole base. The COMMY_SUBSCRIBE seed survives because the seeding put those +// channels in the realm — not because a local copy of it was kept. And another +// session's record must not leak in. +test('a fresh session (no record) rebuilds from the realm and replays buffered deltas', () => Effect.runPromise( Effect.gen(function* () { const narrowSet = createNarrowSet() narrowSet.add(channelIntent('env-default')) narrowSet.beginBuffering() - const { inbox, subscribes } = spyInbox() + // The realm already holds the seeded channel: the bootstrap subscribed it. + const { inbox, subscribes } = spyInbox(['env-default']) const session = yield* Deferred.make() // The store holds a DIFFERENT session's data; SID_FRESH is a miss. const subscriptionStore = inMemorySubscriptionStore( @@ -223,26 +246,34 @@ test('a fresh session (store miss) keeps the env seed and replays buffered delta ) const fiber = yield* Effect.fork( - restoreSubscriptions({ subscriptionStore, narrowSet, inbox }), + restoreSubscriptions({ + persisted: subscriptionStore.read(), + isBound: () => true, + narrowSet, + inbox, + }), ) narrowSet.add(channelIntent('fresh-sub')) yield* Deferred.succeed(session, asSessionId(SID_FRESH)) yield* Fiber.join(fiber) - // Env seed kept, buffered delta replayed, nothing re-subscribed, and the - // other session's decisions thread was NOT restored. + // Realm channel recovered, buffered delta replayed, and the other + // session's decisions thread was NOT restored. expect(sortIntents(narrowSet.intents())).toEqual( sortIntents([channelIntent('env-default'), channelIntent('fresh-sub')]), ) + // Nothing re-declared on the substrate: this process already subscribed + // `env-default` on the way in, so the rebuild has no round-trip to make. expect(subscribes).toEqual([]) expect(narrowSet.matches(reactionOnDecisionsThread(), undefined)).toBe(false) }), )) -// An empty persisted set is honoured verbatim: the resumed seat hears only what -// it re-subscribed since boot, and a dropped default stays dropped. -test('an empty persisted set is honoured — the env default is not resurrected', () => +// A seat that unsubscribed from everything comes back with nothing: the realm +// holds no rows for it and its topic record is empty, so a dropped default +// stays dropped rather than being resurrected from the env. +test('a seat with no realm rows and no topic record comes back empty', () => Effect.runPromise( Effect.gen(function* () { const narrowSet = createNarrowSet() @@ -253,7 +284,12 @@ test('an empty persisted set is honoured — the env default is not resurrected' const subscriptionStore = inMemorySubscriptionStore(session, new Map([[SID_RESUME, []]])) const fiber = yield* Effect.fork( - restoreSubscriptions({ subscriptionStore, narrowSet, inbox }), + restoreSubscriptions({ + persisted: subscriptionStore.read(), + isBound: () => true, + narrowSet, + inbox, + }), ) yield* Deferred.succeed(session, asSessionId(SID_RESUME)) yield* Fiber.join(fiber) diff --git a/packages/mcp/subscription-store.ts b/packages/mcp/subscription-store.ts index 80e5e49..76b1b5f 100644 --- a/packages/mcp/subscription-store.ts +++ b/packages/mcp/subscription-store.ts @@ -18,31 +18,33 @@ import { SessionId, type SessionIdValue } from './session-id.ts' import type { SubscribeIntent } from './subscribe-parser.ts' /** - * Persistent per-session_id narrow-set snapshot. + * Per-session_id record of the narrows a subscription row cannot express. * - * The plugin-layer `narrowSet` (narrow-set.ts) decides which inbound events - * reach this agent; it is in-memory only and rebuilt from `COMMY_SUBSCRIBE` - * on every boot, so a stopped-then-resumed Claude Code session silently - * loses every runtime `subscribe`/`unsubscribe`. This store persists the - * current set to disk on each mutation and restores it on resume, so the - * session comes back with exactly the subscriptions it had — including - * runtime unsubscribes (a dropped default channel stays dropped). + * This file used to hold the whole narrow set, and its own reasoning said why: + * the set was in-memory, rebuilt from `COMMY_SUBSCRIBE` every boot, so a + * resumed session lost every runtime `subscribe`/`unsubscribe`. That premise is + * gone. Subscriptions are the seat's own realm state now, and a seat rebuilds + * by ASKING the realm what it is subscribed to (`subscription-restore.ts`). * - * The store is keyed on **session_id**, never identity id: the rule is - * about the session, and Claude Code keeps the same session_id across a - * resume. Keying on identity id would wrongly treat each relaunch of a - * pinned `COMMY_BOT_NAME` pane as a resume; a new session_id every launch - * keeps those panes on the fresh `COMMY_SUBSCRIBE`-only path for free. + * What the realm cannot answer is narrowing below a channel — a subscription + * row names a channel, so `#chan/topic` and `new-topics:#chan` read back as + * plain `#chan`. That is all this record still holds. Channel-wide intents are + * deliberately absent: a second copy of something the realm already holds could + * only ever disagree with the rows that actually govern delivery. * - * `read` yields `Option.none` only when no file exists for the session — - * the "fresh session, seed from `COMMY_SUBSCRIBE`" signal. An empty - * persisted set returns `Option.some([])`, NOT `none`: a session that - * unsubscribed from everything must resume with nothing, not fall back to - * the env defaults. A file that exists but cannot be parsed surfaces a - * `ParseError` rather than masquerading as absent. + * Still keyed on **session_id**, never identity id — Graeme's 2026-06-29 + * ruling, and it survives the shrink intact. Keying on identity would treat + * each relaunch of a pinned `COMMY_BOT_NAME` pane as a resume of the last one. * - * Writes overwrite unconditionally — the latest snapshot is authoritative, - * so a smaller set (after an unsubscribe) correctly replaces a larger one. + * `read` yields `Option.none` only when no file exists for the session, which + * is what marks a first launch. An empty record returns `Option.some([])`, NOT + * `none`: a session whose narrows are all channel-wide has still run before, + * and `seedDefaultsIfFresh` reads exactly that distinction. A file that exists + * but cannot be parsed surfaces a `ParseError` rather than masquerading as + * absent. + * + * Writes overwrite unconditionally — the latest snapshot is authoritative, so a + * smaller set (after an unsubscribe) correctly replaces a larger one. */ export interface SubscriptionStore { read(): Effect.Effect< diff --git a/packages/mcp/test-platform.ts b/packages/mcp/test-platform.ts index dc84145..0236596 100644 --- a/packages/mcp/test-platform.ts +++ b/packages/mcp/test-platform.ts @@ -1,6 +1,9 @@ +import type { EventQueueCursor } from '@commy/core/ports' import type { CommandExecutor, FileSystem } from '@effect/platform' import { NodeContext } from '@effect/platform-node' -import { ConfigProvider, Layer } from 'effect' +import { ConfigProvider, Effect, Layer, Option } from 'effect' +import { type QueueStateStore, QueueStateStoreTag } from './queue-state-store.ts' +import type { SessionIdValue } from './session-id.ts' /** * Fixture config source for the boot tests. `parseEnv` reads the ambient @@ -34,3 +37,35 @@ export const testPlatformLayer = ( env: Record, ): Layer.Layer => Layer.merge(testConfigProviderLayer(env), NodeContext.layer) + +/** + * In-memory queue-state store for the boot tests — keeps the runner's homedir + * untouched, the same reason the cursor store is faked there. Boot reads it to + * answer "is there anything to resume?"; a harness that never persists a queue + * always answers no, which is the fresh-session case. + */ +export const createInMemoryQueueStateStore = (): QueueStateStore => { + const states = new Map() + return { + read: (sessionId: SessionIdValue) => + Effect.sync(() => Option.fromNullable(states.get(sessionId as string))), + write: (sessionId: SessionIdValue, state: EventQueueCursor) => + Effect.sync(() => { + states.set(sessionId as string, state) + }), + advance: (sessionId: SessionIdValue, lastEventId: number) => + Effect.sync(() => { + const prior = states.get(sessionId as string) + if (prior !== undefined) states.set(sessionId as string, { ...prior, lastEventId }) + }), + } +} + +/** + * The boot-time store every substituted-adapter harness needs and none of them + * cares about: the queue-state store boot reads to answer "is there anything to + * resume?". A FUNCTION, not a shared constant, so one harness's writes cannot + * reach another's boot. + */ +export const testBootStoresLayer = (): Layer.Layer => + Layer.succeed(QueueStateStoreTag, createInMemoryQueueStateStore()) diff --git a/packages/memory/adapter.ts b/packages/memory/adapter.ts index 6011a42..66e830f 100644 --- a/packages/memory/adapter.ts +++ b/packages/memory/adapter.ts @@ -35,6 +35,7 @@ import type { HistoryReader, Identity, IdentityKind, + IdentityOrigin, IdentityPort, InboundEvent, Mention, @@ -126,6 +127,7 @@ interface Binding { readonly acquiredName: BotName readonly identity: Identity readonly credentials: Credentials + readonly origin: IdentityOrigin } type Emit = (event: InboundEvent) => void @@ -271,13 +273,20 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect): Effect.Effect => Ref.getAndUpdate(counter, (n) => n + 1) + // Reports which arm it took, so `acquire` reads the origin of the bind off + // the registration itself instead of probing the map a second time. Callers + // seeding a peer discard it — an origin only means something for the + // identity a seat binds AS. const registerIdentity = ( name: string, kind: IdentityKind, - ): Effect.Effect => + ): Effect.Effect< + { readonly identity: Identity; readonly origin: IdentityOrigin }, + ParseResult.ParseError + > => Effect.gen(function* () { const existing = identitiesByName.get(name) - if (existing !== undefined) return existing + if (existing !== undefined) return { identity: existing, origin: 'existing' as const } const id = yield* decodeIdentityId(String(yield* allocId(nextIdentityId))).pipe( Effect.orDie, ) @@ -285,7 +294,7 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect Effect.succeed({ identity: b.identity, credentials: b.credentials }), + onSome: (b) => + Effect.succeed({ + identity: b.identity, + credentials: b.credentials, + origin: b.origin, + }), }), ), ) @@ -395,7 +409,7 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect + Effect.flatMap(({ identity: ident, origin }) => allocId(nextCredentialsId).pipe( Effect.flatMap((credId) => { const credentials: Credentials = { @@ -407,6 +421,7 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect + requireBound().pipe( + Effect.zipRight(Ref.get(subscriptions)), + Effect.map((keys) => + Arr.dedupe( + Arr.filterMap(Arr.fromIterable(keys), (key) => + key.kind === 'mentions' ? Option.none() : Option.some(key.channelName), + ), + ), + ), + ), events: () => Stream.asyncPush((emit) => Effect.acquireRelease( @@ -967,9 +998,9 @@ export const memoryAdapter = (config: MemoryAdapterConfig = {}): Effect.Effect => registerChannel(name) const seedAgent = (name: string): Effect.Effect => - registerIdentity(name, 'agent') + registerIdentity(name, 'agent').pipe(Effect.map((registered) => registered.identity)) const seedHuman = (name: string): Effect.Effect => - registerIdentity(name, 'human') + registerIdentity(name, 'human').pipe(Effect.map((registered) => registered.identity)) // Authored by `peer`, not the bound self: mirrors publisher.post's store + // fan-out path but stamps `sender: peer` and skips requireBound. The diff --git a/packages/testing/contract.ts b/packages/testing/contract.ts index 4c3523a..1575d15 100644 --- a/packages/testing/contract.ts +++ b/packages/testing/contract.ts @@ -1174,6 +1174,40 @@ export const runAgentCommsContract = (label: string, factory: ContractFactory): ), )) + // What a seat reads back on the way up, and the ceiling on what it can + // read. A subscription names a CHANNEL on every substrate worth targeting, + // so a narrow below that — one topic, or first-messages-only — reads back as + // its plain channel. A substrate that answered anything finer would make the + // caller's own narrowing look redundant when it is not. + test('inbox.subscriptions() reports the channels this seat subscribed, and only channels', () => + Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const lobby = yield* env.seedChannel('lobby') + const annex = yield* env.seedChannel('annex') + yield* env.comms.inbox.subscribe(lobby.name) + yield* env.comms.inbox.subscribe({ + channel: annex.name, + thread: decodeThreadNameSync('planning'), + }) + const subscribed = yield* env.comms.inbox.subscriptions() + expect([...subscribed].sort()).toEqual([annex.name, lobby.name].sort()) + }), + ), + )) + + test('inbox.subscriptions() drops a channel the seat unsubscribed from', () => + Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const lobby = yield* env.seedChannel('lobby') + yield* env.comms.inbox.subscribe(lobby.name) + yield* env.comms.inbox.unsubscribe(lobby.name) + expect(yield* env.comms.inbox.subscriptions()).toEqual([]) + }), + ), + )) + test('inbox.replay(since) returns past message-posted events with ts >= since', () => Effect.runPromise( Effect.gen(function* () { diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index d33d6dd..5c6ec80 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -16,6 +16,7 @@ import type { HistoryReader, Identity, IdentityId, + IdentityOrigin, IdentityPort, InboundEvent, Message, @@ -520,6 +521,12 @@ interface BoundState { readonly credentials: Credentials readonly userId: ZulipUserRef readonly http: BotHttp + /** + * Recorded so the idempotent re-acquire below answers with the origin of the + * bind that actually happened, rather than re-deriving one from a realm the + * first acquire has already changed. + */ + readonly origin: IdentityOrigin } // Channels are addressed by name, so the narrow sets and the per-channel @@ -630,7 +637,11 @@ export const zulipAdapter = ( }), ), onSome: (bound: BoundState) => - Effect.succeed({ identity: bound.identity, credentials: bound.credentials }), + Effect.succeed({ + identity: bound.identity, + credentials: bound.credentials, + origin: bound.origin, + }), }), ), ) @@ -889,15 +900,22 @@ export const zulipAdapter = ( ), ) - interface MintedBot { + interface BoundBot { readonly userId: ZulipUserRef readonly apiKey: ApiKeyType readonly email: BotEmailType + /** + * Which arm of the lookup produced this bot. Carried out of the adapter + * because the realm answers it here and nowhere else: once the bind has + * resolved, a bot the minter just created and one it has served for + * months are indistinguishable. + */ + readonly origin: IdentityOrigin } const mintBot = ( name: BotName, - ): Effect.Effect => { + ): Effect.Effect => { const shortName = sanitiseShortName(name) return minterHttp .post('/bots', newBotSchema, { @@ -912,10 +930,11 @@ export const zulipAdapter = ( apiKey: ApiKey(res.api_key), }).pipe( Effect.map( - ({ email, apiKey }): MintedBot => ({ + ({ email, apiKey }): BoundBot => ({ userId: ZulipUserRef(res.user_id), apiKey, email, + origin: 'minted', }), ), ), @@ -938,7 +957,7 @@ export const zulipAdapter = ( const regenerateBotKey = ( existing: ZulipUser, - ): Effect.Effect => + ): Effect.Effect => minterHttp.post(`/bots/${existing.user_id}/api_key/regenerate`, regenerateKeySchema, {}).pipe( Effect.flatMap((res) => Effect.all({ @@ -946,10 +965,11 @@ export const zulipAdapter = ( apiKey: ApiKey(res.api_key), }).pipe( Effect.map( - ({ email, apiKey }): MintedBot => ({ + ({ email, apiKey }): BoundBot => ({ userId: ZulipUserRef(existing.user_id), apiKey, email, + origin: 'existing', }), ), ), @@ -958,7 +978,7 @@ export const zulipAdapter = ( const acquireBot = ( name: BotName, - ): Effect.Effect => + ): Effect.Effect => findAnyBotByName(name).pipe( Effect.flatMap( Option.match({ @@ -978,21 +998,25 @@ export const zulipAdapter = ( const attachBot = ( name: BotName, apiKey: ApiKeyType, - ): Effect.Effect => + ): Effect.Effect => findAnyBotByName(name).pipe( Effect.flatMap( Option.match({ - onNone: (): Effect.Effect => + onNone: (): Effect.Effect => Effect.fail(new AttachIdentityNotFound({ name })), onSome: ( existing, - ): Effect.Effect => + ): Effect.Effect => BotEmail(existing.email).pipe( Effect.map( - (email): MintedBot => ({ + // Attach binds a persona the operator provisioned out of + // band, so it is `existing` even on this seat's first ever + // bind — the account predates us and owns its own state. + (email): BoundBot => ({ userId: ZulipUserRef(existing.user_id), apiKey, email, + origin: 'existing', }), ), ), @@ -1005,7 +1029,7 @@ export const zulipAdapter = ( const provideMintedFor = ( name: BotName, ): Effect.Effect< - MintedBot, + BoundBot, ZulipApiError | ReactivateForbidden | AttachIdentityNotFound | ParseResult.ParseError > => { const attach = config.attachIdentity @@ -1029,7 +1053,11 @@ export const zulipAdapter = ( onSome: (existing) => { if (existing.acquiredName === name) { return Effect.succeed]>([ - { identity: existing.identity, credentials: existing.credentials }, + { + identity: existing.identity, + credentials: existing.credentials, + origin: existing.origin, + }, current, ]) } @@ -1073,8 +1101,12 @@ export const zulipAdapter = ( credentials, userId: minted.userId, http, + origin: minted.origin, } - return [{ identity: ident, credentials }, Option.some(next)] as const + return [ + { identity: ident, credentials, origin: minted.origin }, + Option.some(next), + ] as const }), ), Effect.mapError((cause) => new IdentityError({ operation: 'acquire', cause })), @@ -1193,6 +1225,14 @@ export const zulipAdapter = ( const subscriptionsResponseSchema = Schema.Struct({ result: Schema.Literal('success') }) + // GET /users/me/subscriptions — the seat's own subscription rows. Only the + // names matter here; the rest of each row describes delivery settings the + // plugin does not model. + const mySubscriptionsResponseSchema = Schema.Struct({ + result: Schema.Literal('success'), + subscriptions: Schema.Array(Schema.Struct({ name: Schema.NonEmptyString })), + }) + // POST /users/me/subscriptions response carries a per-user map of // names actually subscribed vs already subscribed. For minter-routed // calls we only care about the minter's row; defaults to empty so @@ -1855,6 +1895,20 @@ export const zulipAdapter = ( }).pipe(Effect.mapError((cause) => new InboxError({ operation: 'subscribe', cause }))), ), ), + // Reads the seat's own rows, so it goes out on the seat's credential like + // the writes do — "me" here has to be the seat, and the minter's answer + // would be a different seat's subscriptions wearing this one's name. + subscriptions: () => + boundHttp().pipe( + Effect.flatMap((http) => + http.get('/users/me/subscriptions', mySubscriptionsResponseSchema).pipe( + Effect.flatMap((res) => + Effect.forEach(res.subscriptions, (row) => decodeChannelName(row.name)), + ), + Effect.mapError((cause) => new InboxError({ operation: 'subscribe', cause })), + ), + ), + ), unsubscribe: (target) => boundHttp().pipe( Effect.flatMap((http) =>