From 440670ec343835b57a03adaded1869f87226e256 Mon Sep 17 00:00:00 2001 From: Graeme Foster <80714+GraemeF@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:07:10 +0100 Subject: [PATCH] Nothing listens on another seat's behalf, so the minter subscribes to nothing (comms-g5zh.7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The minter subscribed to every public stream so it could receive on behalf of a seat that had not yet minted, kept current by a boot-time reconciler. With the events queue and the channel subscriptions both registered under the seat's own principal, no seat receives through another, and that layer has nothing left to do. The minter keeps the jobs that are its own: minting bots, and the directory and history reads that need no principal. Deletes `minter-reconciler.ts`, its `reconcileMinterSubscriptions` wiring in the Zulip adapter, the boot call in the plugin's `main`, and the covering tests — the capability they cover is gone with it. The prose that described the arrangement (`tools.ts`, `agent-experience.md`, the client README, the adapter's per-event filter) now describes what the code does. A seat that never mints costs the realm nothing. --- clients/claude-code/README.md | 14 +-- docs/agent-experience.md | 30 +++-- packages/mcp/bootstrap.ts | 2 +- packages/mcp/disconnect-exit.fixture.ts | 4 +- packages/mcp/memory-substrate.ts | 23 ++-- packages/mcp/server.integration.test.ts | 5 +- packages/mcp/server.live.test.ts | 4 +- packages/mcp/server.test.ts | 56 +-------- packages/mcp/server.ts | 19 +-- packages/mcp/tools.ts | 9 +- packages/zulip/adapter.test.ts | 146 ----------------------- packages/zulip/adapter.ts | 57 +-------- packages/zulip/minter-reconciler.test.ts | 120 ------------------- packages/zulip/minter-reconciler.ts | 59 --------- 14 files changed, 47 insertions(+), 501 deletions(-) delete mode 100644 packages/zulip/minter-reconciler.test.ts delete mode 100644 packages/zulip/minter-reconciler.ts diff --git a/clients/claude-code/README.md b/clients/claude-code/README.md index 9920498..9324770 100644 --- a/clients/claude-code/README.md +++ b/clients/claude-code/README.md @@ -63,7 +63,7 @@ process env inheritance: ### Eager vs lazy boot, in one diagram ``` -parseEnv → buildAdapter → reconcileMinterSubscriptions (non-fatal) → +parseEnv → buildAdapter → │ ├── COMMY_BOT_NAME set → │ persistent single-identity cache; @@ -90,16 +90,14 @@ parseEnv → buildAdapter → reconcileMinterSubscriptions (non-fatal) → │ ▼ subscribeFromEnv → registerTools → connect transport → - startEventPump (minter-side queue, filtered by narrowSet) → + startEventPump (seat-side queue, filtered by narrowSet) → wire shutdown (release only if acquire happened) ``` -`reconcileMinterSubscriptions` is the boot-time backstop that keeps the -minter subscribed to every public stream in the realm. It runs once per -plugin process; new streams created during the process's lifetime are -covered by the per-session POST inside `inbox.subscribe()`. Failure is -non-fatal — the diagnostic goes to stderr and boot continues with a -possibly-degraded lurker view. +Boot touches no channel the seat wasn't asked to hold. Every subscription +and the events queue itself belong to the seat's own principal, registered +when the seat first subscribes — so a seat that never mints leaves no trace +on the realm. ### Run shapes diff --git a/docs/agent-experience.md b/docs/agent-experience.md index 0904fb6..4e56dfa 100644 --- a/docs/agent-experience.md +++ b/docs/agent-experience.md @@ -149,10 +149,10 @@ something must listen on its behalf. That something was the minter, subscribed to every public stream, with the event queue registered against it rather than the per-session bot. -The example is told in the past tense because the first two steps of the -chain have since been taken apart: the queue and the subscriptions now -belong to the seat. The rest of the chain is still standing, and the -paragraphs below say which parts. +The example is told in the past tense because that arrangement is gone: the +queue and the subscriptions belong to the seat, and the minter listens for +nobody. One piece of the chain is still standing, and the paragraphs below +say which. Everything else follows from that one deferral. One shared subscriber means per-agent narrowing cannot be a realm subscription, so it becomes a @@ -188,11 +188,16 @@ had to move together, because Zulip builds a channel message's recipient set from the channel's subscription rows: a seat-owned queue over minter-held subscriptions would have received nothing at all. -What has not moved yet: the minter still holds its blanket public-stream -subscription, and 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. +The minter holds no stream subscriptions. It has two jobs, both genuinely +its own: minting bots, and the directory and history reads that need no +principal. Nothing listens on another seat's behalf, so a seat that never +mints costs the realm nothing — the saving the optimisation was after, +without the architecture that funded it. + +Topic-level narrowing is still a client-side filter with a local record +behind it — the last piece of the chain standing. That is the next step of +the same unwinding, not an exemption; channel-level narrowing is realm state +under the seat's own principal, 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. @@ -204,10 +209,9 @@ while the architecture funding it does not. What could not be deferred was receiving — a queue is state held on the agent's behalf. Reading was never the problem. -Two things would remain client-side afterwards, both legitimately: -topic-level narrows and the event-queue handle. The difference is that they -would filter the agent's own queue rather than a shared one — local, small, -and interfering with nobody. +Two things remain client-side, both legitimately: topic-level narrows and +the event-queue handle. Both filter the agent's own queue rather than a +shared one — local, small, and interfering with nobody. They are legitimate for different reasons, and only one of them is an exemption. The event-queue handle is principle 3's second: the substrate diff --git a/packages/mcp/bootstrap.ts b/packages/mcp/bootstrap.ts index 87706ab..77f0ef5 100644 --- a/packages/mcp/bootstrap.ts +++ b/packages/mcp/bootstrap.ts @@ -635,7 +635,7 @@ export const readGitContext = ( /** * The full driven surface `main` composes against: the * universal `AgentComms` aggregate plus the Zulip-shaped boot extras - * (reconcile / download / upload / close). Lives in the plugin — core + * (download / upload / close). Lives in the plugin — core * stays substrate-neutral, and `registerTools` keeps its narrower * `AgentComms` dependency via structural subtyping. Request-time DI * (methods carrying `R = HttpClient`) is deferred. diff --git a/packages/mcp/disconnect-exit.fixture.ts b/packages/mcp/disconnect-exit.fixture.ts index 14e6ab8..4900a32 100644 --- a/packages/mcp/disconnect-exit.fixture.ts +++ b/packages/mcp/disconnect-exit.fixture.ts @@ -44,8 +44,8 @@ const inMemorySubscriptionStore = { /** * Complete the in-memory substrate to the `ZulipAdapter` shape the program - * expects. `reconcileMinterSubscriptions` (boot) and `close` (shutdown - * finalizer) are exercised, so the helper's inert no-ops suffice; + * expects. `close` (shutdown finalizer) is exercised, so the helper's inert + * no-op suffices; * `uploadFile`/`downloadFile` are never reached without an MCP client driving * tools, so they die loudly if anything calls them. */ diff --git a/packages/mcp/memory-substrate.ts b/packages/mcp/memory-substrate.ts index 1e09b85..a05f043 100644 --- a/packages/mcp/memory-substrate.ts +++ b/packages/mcp/memory-substrate.ts @@ -13,8 +13,8 @@ import { Duration, Effect } from 'effect' * docs/architecture.md § Test architecture). But the `SubstrateAdapter` port * those programs depend on is currently *typed as* {@link ZulipAdapter}, so any * provided double must be completed from the universal {@link AgentComms} core - * to that Zulip-shaped aggregate: `reconcileMinterSubscriptions`, - * `downloadFile`, `uploadFile`, `close`. Concentrating that completion here + * to that Zulip-shaped aggregate: `downloadFile`, `uploadFile`, `close`. + * Concentrating that completion here * keeps the rule self-enforcing: among the above-port tests, `@commy/zulip` * appears in exactly one module, this one. Tests that deliberately drive the * real adapter (`queue-resume`, `bootstrap`, the live suite) name it for the @@ -23,32 +23,27 @@ import { Duration, Effect } from 'effect' * * The members themselves no longer speak Zulip. `downloadFile` / `uploadFile` * are the port's `AttachmentStore`, so their doubles are built from - * `@commy/core/ports` types alone; only `reconcileMinterSubscriptions` and - * `close` are still Zulip-shaped, and they are why the aggregate is still - * named here at all. + * `@commy/core/ports` types alone; only `close` is still Zulip-shaped, and it + * is why the aggregate is still named here at all. * * `ZulipAdapter` is re-exported so callers annotate their doubles without * naming `@commy/zulip` themselves. */ export type { ZulipAdapter } from '@commy/zulip/adapter' -/** The four members that complete `AgentComms` to a `ZulipAdapter`. */ -type SubstrateExtras = Pick< - ZulipAdapter, - 'reconcileMinterSubscriptions' | 'downloadFile' | 'uploadFile' | 'close' -> +/** The three members that complete `AgentComms` to a `ZulipAdapter`. */ +type SubstrateExtras = Pick /** - * Per-member overrides. Anything omitted falls back to an inert default: a - * no-op reconcile report, an empty download, a stub upload result, a no-op - * close. Tests override only the member whose behaviour they actually assert. + * Per-member overrides. Anything omitted falls back to an inert default: an + * empty download, a stub upload result, a no-op close. Tests override only the + * member whose behaviour they actually assert. */ type SubstrateExtrasOverrides = Partial const stubAttachmentRef = decodeAttachmentRefSync('/user_uploads/0/stub') const inertExtras: SubstrateExtras = { - reconcileMinterSubscriptions: () => Effect.succeed({ added: [], error: undefined }), // Empty bytes, but a real filename: the port's contract is that the adapter // names the file, so a double that answered a constant would let a caller // deriving its own name from the handle pass its tests. diff --git a/packages/mcp/server.integration.test.ts b/packages/mcp/server.integration.test.ts index 180409a..4067032 100644 --- a/packages/mcp/server.integration.test.ts +++ b/packages/mcp/server.integration.test.ts @@ -1245,9 +1245,8 @@ test('post by self does NOT fire a claude/channel notification (self-echo suppre // ─── Pump narrow-set filter ───────────────────────────────────────────────── test('pump filter: event for a never-subscribed channel does NOT fire claude/channel notification', async () => { - // Production wiring assertion. The Zulip minter is subscribed to every - // public stream (per `minter-reconciler.ts`), so the adapter - // inbox yields events for streams the calling session never subscribed + // Production wiring assertion. A seat's own event queue carries no narrow, + // so it can yield events for streams the calling session never subscribed // to via the MCP `subscribe` tool or `COMMY_SUBSCRIBE` env. The // plugin-layer NarrowSet (`narrow-set.ts`) is the filter that decides // which of those events the MCP host actually sees. This test exercises diff --git a/packages/mcp/server.live.test.ts b/packages/mcp/server.live.test.ts index 968b6fe..10a9d78 100644 --- a/packages/mcp/server.live.test.ts +++ b/packages/mcp/server.live.test.ts @@ -92,7 +92,7 @@ const liveEnv = (): LiveEnv => { } // Same minter-call spacing as the substrate live suite — -// boot's reconcile + acquire + subscribe sequence plus per-test +// boot's acquire + subscribe sequence plus per-test // release all hit the shared minter, so the plugin live suite trips // the same per-user limit if we don't pace. const MINTER_PACE = Duration.millis(900) @@ -425,7 +425,7 @@ describeLive('commy plugin live integration — zulip.example.com', () => { const e = liveEnv() yield* Effect.scoped( Effect.gen(function* () { - // Pace before main()'s boot-time minter calls (reconcile + subscribe). + // Pace before main()'s boot-time minter calls (acquire + subscribe). yield* Effect.sleep(MINTER_PACE) const client = yield* buildHarness({ COMMY_SUBSCRIBE: `${e.channelName}`, diff --git a/packages/mcp/server.test.ts b/packages/mcp/server.test.ts index ecae1f8..1488a4e 100644 --- a/packages/mcp/server.test.ts +++ b/packages/mcp/server.test.ts @@ -167,7 +167,6 @@ interface FakeAdapterCalls { readonly acquired: string[] readonly closes: { count: number } readonly subscribed: SubscriptionTarget[] - readonly reconcileCalls: { count: number } readonly events: string[] } @@ -178,16 +177,11 @@ const buildFakeAdapter = ( 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 - } } = {}, ): { readonly adapter: ZulipAdapter; readonly calls: FakeAdapterCalls } => { const acquired: string[] = [] const closes = { count: 0 } const subscribed: SubscriptionTarget[] = [] - const reconcileCalls = { count: 0 } const events: string[] = [] const identity: Identity = { id: decodeIdentityIdSync('bot:myproject-concierge'), @@ -258,25 +252,15 @@ const buildFakeAdapter = ( channelDescription: () => Effect.succeed(Option.none()), presence: (_id: Identity): Effect.Effect => Effect.succeed('offline'), } - const defaultReconcileReport = { - added: [] as ReadonlyArray, - error: undefined as string | undefined, - } const adapter = completeAsSubstrate( { identity: identityPort, publisher, inbox, history, directory }, { - reconcileMinterSubscriptions: () => - Effect.sync(() => { - events.push('reconcile') - reconcileCalls.count += 1 - return options.reconcileReport ?? defaultReconcileReport - }), close: async () => { closes.count += 1 }, }, ) - return { adapter, calls: { acquired, closes, subscribed, reconcileCalls, events } } + return { adapter, calls: { acquired, closes, subscribed, events } } } test('main resolves cleanly when given a valid env', async () => { @@ -348,7 +332,7 @@ test('lazy mode (cc-<8> from session id) does NOT acquire at boot', async () => } const exit = await runProgram(env, fake.adapter, { loggerLayer: captureLogger(stderr) }) // No acquire call, no acquire-failure stderr, clean boot, adapter still - // closed. Reconcile is silent in the no-op case. + // closed. expect(Exit.isSuccess(exit)).toBe(true) expect(fake.calls.acquired).toEqual([]) expect(stderr).toEqual([]) @@ -641,42 +625,6 @@ test('main applies env-driven subscriptions in order after acquire and Type-1 de expect(fake.calls.closes.count).toBe(1) }) -test('main reconciles minter subscriptions during boot before env subscribes', async () => { - const fake = buildFakeAdapter({ - reconcileReport: { - added: [decodeChannelNameSync('commy'), decodeChannelNameSync('general')], - error: undefined, - }, - }) - const log: string[] = [] - const env = { ...validEnv, COMMY_SUBSCRIBE: 'home' } - await runProgram(env, fake.adapter, { loggerLayer: captureLogger(log) }) - expect(fake.calls.reconcileCalls.count).toBe(1) - expect(fake.calls.events.indexOf('reconcile')).toBeLessThan( - fake.calls.events.indexOf('subscribe'), - ) - expect(log.some((line) => line.includes('commy') && line.includes('general'))).toBe(true) -}) - -test('main calls reconcile but stays silent when there is nothing to add', async () => { - const fake = buildFakeAdapter() - const log: string[] = [] - await runProgram(validEnv, fake.adapter, { loggerLayer: captureLogger(log) }) - expect(fake.calls.reconcileCalls.count).toBe(1) - expect(log).toEqual([]) -}) - -test('main keeps booting when reconcile reports an error (log + continue)', async () => { - const fake = buildFakeAdapter({ - reconcileReport: { added: [], error: 'realm unreachable' }, - }) - const log: string[] = [] - const exit = await runProgram(validEnv, fake.adapter, { loggerLayer: captureLogger(log) }) - expect(Exit.isSuccess(exit)).toBe(true) - expect(fake.calls.acquired).toEqual(['myproject-concierge']) - expect(log.some((line) => line.includes('realm unreachable'))).toBe(true) -}) - // ─── Type-1 default sub set for project concierges ────────────── test('persistent mode + project registers Type-1 defaults (new-topics + thread/general)', async () => { diff --git a/packages/mcp/server.ts b/packages/mcp/server.ts index a4d10e6..e924ca8 100644 --- a/packages/mcp/server.ts +++ b/packages/mcp/server.ts @@ -374,7 +374,7 @@ const buildIdentityCache = ( /** * The plugin's boot program as ONE composed Effect, - * from parse → reconcile → identity → tools → pump, run at a single + * from parse → identity → tools → pump, run at a single * `runMain` edge. Services (substrate adapter, cursor store, * ConfigProvider, logger) arrive through the app Layer; * {@link ProgramParams} carries the remaining per-run knobs. @@ -580,23 +580,6 @@ export const makeProgram = ( return Effect.map(deriveProject({ cwd, readGitContext }), Option.getOrUndefined) } - // Minter subscription reconcile: boot-time backstop that - // keeps the minter subscribed to every public stream. Non-fatal — - // log + continue. Silent in the steady-state no-op case. - yield* adapter.reconcileMinterSubscriptions().pipe( - Effect.flatMap((reconcile) => { - if (reconcile.error !== undefined) { - return Effect.logError(`commy plugin: minter reconcile failed: ${reconcile.error}`) - } - if (reconcile.added.length > 0) { - return Effect.logInfo( - `commy plugin: minter reconcile — subscribed minter to ${reconcile.added.length} new public stream(s): ${reconcile.added.join(', ')}`, - ) - } - return Effect.void - }), - ) - // Sample the realm-wide editing switch once, before the tool list is // built, so a seat on a realm with editing off is never offered // `edit_message`. Sampled here rather than held as a static capability diff --git a/packages/mcp/tools.ts b/packages/mcp/tools.ts index 24fdd6a..4cc8f18 100644 --- a/packages/mcp/tools.ts +++ b/packages/mcp/tools.ts @@ -164,9 +164,8 @@ export interface ToolsCache extends ToolsMemory { * `narrowSet` is the consumer-side filter for the inbound event * pump. `subscribe` / `unsubscribe` mutate it so the pump tees only * intended events to the MCP host. The substrate-side call - * (`inbox.subscribe` / `inbox.unsubscribe`) handles streams created - * after the plugin booted; the boot-time minter reconciler - * covers the rest. + * (`inbox.subscribe` / `inbox.unsubscribe`) carries the same change to + * the realm under the seat's own principal. */ export interface RegisterToolsDeps { readonly adapter: AgentComms @@ -911,8 +910,8 @@ const buildToolDefs = (deps: RegisterToolsDeps, cache: InternalCache): ReadonlyA } // Two sinks (see bootstrap.subscribeFromEnv): the consumer-side // narrow tells the event pump to tee matching events through; - // the substrate-side call subscribes the minter to streams the - // boot-time reconciler didn't have a chance to cover. + // the substrate-side call subscribes THIS SEAT to the channel, so + // the realm delivers its messages to the seat's own queue. yield* Effect.sync(() => narrowSet.add(intent)).pipe( Effect.andThen(adapter.inbox.subscribe(intentToTarget(intent))), ) diff --git a/packages/zulip/adapter.test.ts b/packages/zulip/adapter.test.ts index 74a4aa5..aff6462 100644 --- a/packages/zulip/adapter.test.ts +++ b/packages/zulip/adapter.test.ts @@ -3032,152 +3032,6 @@ effectTest('publisher.post after acquire uses BOUND bot creds, not minter creds' }), ) -const seedStreamsList = ( - stub: StubHttpClient, - streams: ReadonlyArray<{ readonly stream_id: number; readonly name: string }>, -): Effect.Effect => - stub.respond('GET', '/api/v1/streams', { - body: { result: 'success', streams }, - }) - -effectTest('reconcileMinterSubscriptions GETs /streams filtered to public-not-subscribed', () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* seedStreamsList(stub, []) - yield* seedUsers(stub, []) - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - yield* adapter.reconcileMinterSubscriptions() - const req = yield* findRequest(stub, 'GET', '/api/v1/streams') - expect(req.url.searchParams.get('include_public')).toBe('true') - expect(req.url.searchParams.get('include_subscribed')).toBe('false') - yield* Effect.promise(() => adapter.close()) - }), -) - -effectTest('reconcileMinterSubscriptions returns empty added when the realm reports no gap', () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* seedStreamsList(stub, []) - yield* seedUsers(stub, []) - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - const report = yield* adapter.reconcileMinterSubscriptions() - expect(report).toEqual({ added: [], error: undefined }) - const reqs = yield* stub.captured - expect( - reqs.find((r) => r.method === 'POST' && r.url.pathname === '/api/v1/users/me/subscriptions'), - ).toBeUndefined() - yield* Effect.promise(() => adapter.close()) - }), -) - -effectTest( - 'reconcileMinterSubscriptions batches every unsubscribed public stream into one POST', - () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* seedStreamsList(stub, [ - { stream_id: 11, name: 'commy' }, - { stream_id: 12, name: 'myproject-a' }, - { stream_id: 13, name: 'myproject-b' }, - ]) - yield* stub.respond('POST', '/api/v1/users/me/subscriptions', { - body: { - result: 'success', - subscribed: { 'minter@example.com': ['commy', 'myproject-a', 'myproject-b'] }, - already_subscribed: {}, - unauthorized: [], - }, - }) - yield* seedUsers(stub, []) - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - const report = yield* adapter.reconcileMinterSubscriptions() - expect(report.added).toEqual([ - decodeChannelNameSync('commy'), - decodeChannelNameSync('myproject-a'), - decodeChannelNameSync('myproject-b'), - ]) - expect(report.error).toBeUndefined() - const post = yield* findRequest(stub, 'POST', '/api/v1/users/me/subscriptions') - const subs = JSON.parse( - new URLSearchParams(post.body).get('subscriptions') ?? '[]', - ) as unknown - expect(subs).toEqual([{ name: 'commy' }, { name: 'myproject-a' }, { name: 'myproject-b' }]) - yield* Effect.promise(() => adapter.close()) - }), -) - -effectTest( - 'reconcileMinterSubscriptions reports only the streams the realm confirms as newly subscribed', - () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* seedStreamsList(stub, [ - { stream_id: 11, name: 'commy' }, - { stream_id: 12, name: 'myproject-b' }, - ]) - // Race: another reconciler already subscribed `myproject-b` between - // our list and post. Zulip puts it under already_subscribed and the - // reconciler's report mirrors that. - yield* stub.respond('POST', '/api/v1/users/me/subscriptions', { - body: { - result: 'success', - subscribed: { 'minter@example.com': ['commy'] }, - already_subscribed: { 'minter@example.com': ['myproject-b'] }, - unauthorized: [], - }, - }) - yield* seedUsers(stub, []) - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - const report = yield* adapter.reconcileMinterSubscriptions() - expect(report.added).toEqual([decodeChannelNameSync('commy')]) - expect(report.error).toBeUndefined() - yield* Effect.promise(() => adapter.close()) - }), -) - -effectTest('reconcileMinterSubscriptions captures list failure without throwing', () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* stub.respond('GET', '/api/v1/streams', { - status: 500, - body: { result: 'error', msg: 'realm unreachable' }, - }) - yield* seedUsers(stub, []) - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - const report = yield* adapter.reconcileMinterSubscriptions() - expect(report.added).toEqual([]) - expect(report.error).toBe('realm unreachable') - const reqs = yield* stub.captured - expect( - reqs.find((r) => r.method === 'POST' && r.url.pathname === '/api/v1/users/me/subscriptions'), - ).toBeUndefined() - yield* Effect.promise(() => adapter.close()) - }), -) - -effectTest('reconcileMinterSubscriptions routes via minter creds', () => - Effect.gen(function* () { - const stub = yield* makeStubHttpClient - yield* seedStreamsList(stub, [{ stream_id: 11, name: 'commy' }]) - yield* stub.respond('POST', '/api/v1/users/me/subscriptions', { - body: { - result: 'success', - subscribed: { 'minter@example.com': ['commy'] }, - already_subscribed: {}, - unauthorized: [], - }, - }) - yield* seedUsers(stub, []) - const adapter = yield* zulipAdapter(stub, yield* makeConfig()) - yield* adapter.reconcileMinterSubscriptions() - const listReq = yield* findRequest(stub, 'GET', '/api/v1/streams') - const postReq = yield* findRequest(stub, 'POST', '/api/v1/users/me/subscriptions') - expect(decodeBasicAuth(listReq.headers.get('Authorization'))).toEqual(minterAuth) - expect(decodeBasicAuth(postReq.headers.get('Authorization'))).toEqual(minterAuth) - yield* Effect.promise(() => adapter.close()) - }), -) - // --- attachmentReference --- test('attachmentReference renders a Zulip markdown link, filename as text and url as target', () => { diff --git a/packages/zulip/adapter.ts b/packages/zulip/adapter.ts index 5c6ec80..a24f68c 100644 --- a/packages/zulip/adapter.ts +++ b/packages/zulip/adapter.ts @@ -106,8 +106,6 @@ import { mentionTokens, unresolvedMentions, } from './mentions.ts' -import type { ReconcileReport } from './minter-reconciler.ts' -import { reconcileMinterSubscriptions } from './minter-reconciler.ts' import { buildMessageRef, permalinkBase, withChannelPermalink } from './permalink.ts' import { type RenderedContentLookup, renderedContentForBatch } from './rendered-content.ts' import { mentionsOfMessage } from './rendered-mentions.ts' @@ -220,14 +218,6 @@ export interface ZulipAdapterConfig { export type ZulipAdapter = AgentComms & AttachmentStore & { - /** - * Subscribe the minter to every public stream it isn't yet on. - * Boot-time backstop so the plugin's event pump observes - * events on streams created after the minter's initial subscription - * set. Non-throwing: failure is captured in the returned report and - * the caller decides whether to log + continue or abort. - */ - reconcileMinterSubscriptions(): Effect.Effect close(): Promise } @@ -1233,22 +1223,6 @@ export const zulipAdapter = ( 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 - // the schema parses cleanly when the realm omits the minter (a race - // where every requested stream was already subscribed by someone - // else in the interim). - const reconcileSubscriptionsResponseSchema = Schema.Struct({ - result: Schema.Literal('success'), - subscribed: Schema.optional( - Schema.Record({ key: Schema.String, value: Schema.Array(Schema.String) }), - ), - already_subscribed: Schema.optional( - Schema.Record({ key: Schema.String, value: Schema.Array(Schema.String) }), - ), - }) - const streamsListResponseSchema = Schema.Struct({ result: Schema.Literal('success'), streams: Schema.Array(Schema.Struct({ name: Schema.NonEmptyString, stream_id: Schema.Int })), @@ -1763,7 +1737,7 @@ export const zulipAdapter = ( // Per-event filter. The new-topics-in-channel narrow is the only // narrow that requires adapter-side state (seen topics); channel:X is - // enforced server-side via the minter's /users/me/subscriptions list. + // enforced server-side by the seat's own /users/me/subscriptions rows. // We therefore only intercept messages belonging to a channel that has // the new-topics narrow active, and pass everything else through // unchanged (preserves the plumbing contract: events queue → @@ -2210,34 +2184,6 @@ export const zulipAdapter = ( ), } - const reconcileMinter = (): Effect.Effect => - reconcileMinterSubscriptions({ - listUnsubscribedPublicStreams: () => - minterHttp - .get('/streams', streamsListResponseSchema, { - include_public: true, - include_subscribed: false, - }) - .pipe( - Effect.flatMap((res) => - Effect.forEach(res.streams, (s) => - decodeChannelName(s.name).pipe(Effect.map((name) => ({ name }))), - ), - ), - ), - subscribeToStreams: (names) => - minterHttp - .post('/users/me/subscriptions', reconcileSubscriptionsResponseSchema, { - subscriptions: JSON.stringify(names.map((name) => ({ name }))), - }) - .pipe( - Effect.flatMap((res) => { - const mintedFor = res.subscribed?.[config.minterEmail] ?? [] - return Effect.forEach(mintedFor, (name) => decodeChannelName(name)) - }), - ), - }) - return { // Zulip stamps integer epoch seconds, so two posts inside the same // second collide on `ts`; a caller needing distinct timestamps must @@ -2248,7 +2194,6 @@ export const zulipAdapter = ( inbox, history, directory, - reconcileMinterSubscriptions: reconcileMinter, downloadFile: (ref: AttachmentRef) => decodeUserUploadPath(ref).pipe( Effect.flatMap((urlPath) => diff --git a/packages/zulip/minter-reconciler.test.ts b/packages/zulip/minter-reconciler.test.ts deleted file mode 100644 index f50215e..0000000 --- a/packages/zulip/minter-reconciler.test.ts +++ /dev/null @@ -1,120 +0,0 @@ -import { expect, test } from 'bun:test' -import { type ChannelName, decodeChannelNameSync } from '@commy/core/ports' -import { Data, Effect } from 'effect' -import type { ReconcilerDeps } from './minter-reconciler.ts' -import { reconcileMinterSubscriptions } from './minter-reconciler.ts' - -// Production deps fail with tagged errors (ZulipApiError / ParseError); this -// stand-in mirrors that shape so the failure-path tests exercise the -// reconciler's `instanceof Error ? .message` rendering branch with a -// representative error rather than a bare global Error. -class DepFailure extends Data.TaggedError('DepFailure')<{ readonly message: string }> {} - -const buildDeps = ( - overrides: Partial> = {}, -): ReconcilerDeps & { - readonly listCalls: { value: number } - readonly subscribeCalls: { value: ReadonlyArray[] } -} => { - const listCalls = { value: 0 } - const subscribeCalls: { value: ReadonlyArray[] } = { value: [] } - const baseDeps: ReconcilerDeps = { - listUnsubscribedPublicStreams: () => - Effect.sync(() => { - listCalls.value += 1 - return [] - }), - subscribeToStreams: (names) => - Effect.sync(() => { - subscribeCalls.value = [...subscribeCalls.value, names] - return names - }), - } - return { ...baseDeps, ...overrides, listCalls, subscribeCalls } -} - -test('returns empty report when minter is already up to date', async () => { - const deps = buildDeps({ - listUnsubscribedPublicStreams: () => Effect.succeed([]), - }) - - const report = await Effect.runPromise(reconcileMinterSubscriptions(deps)) - - expect(report).toEqual({ added: [], error: undefined }) - expect(deps.subscribeCalls.value).toEqual([]) -}) - -test('batches a single subscribe call for every unsubscribed stream', async () => { - const streams = [ - { name: decodeChannelNameSync('commy') }, - { name: decodeChannelNameSync('myproject-a') }, - { name: decodeChannelNameSync('myproject-b') }, - ] - const deps = buildDeps({ - listUnsubscribedPublicStreams: () => Effect.succeed(streams), - subscribeToStreams: (names) => Effect.succeed(names), - }) - - const report = await Effect.runPromise(reconcileMinterSubscriptions(deps)) - - expect(report).toEqual({ - added: [ - decodeChannelNameSync('commy'), - decodeChannelNameSync('myproject-a'), - decodeChannelNameSync('myproject-b'), - ], - error: undefined, - }) -}) - -test('reports only the streams the substrate confirms as newly subscribed', async () => { - // Race: another process already subscribed `myproject-b` in the window - // between list and subscribe. The substrate response excludes it - // from `subscribed`. The reconciler reports only the actual adds. - const deps = buildDeps({ - listUnsubscribedPublicStreams: () => - Effect.succeed([ - { name: decodeChannelNameSync('commy') }, - { name: decodeChannelNameSync('myproject-b') }, - ]), - subscribeToStreams: () => Effect.succeed([decodeChannelNameSync('commy')]), - }) - - const report = await Effect.runPromise(reconcileMinterSubscriptions(deps)) - - expect(report.added).toEqual([decodeChannelNameSync('commy')]) - expect(report.error).toBeUndefined() -}) - -test('captures list failure and never invokes subscribe', async () => { - const deps = buildDeps({ - listUnsubscribedPublicStreams: () => - Effect.fail(new DepFailure({ message: 'realm unreachable' })), - }) - - const report = await Effect.runPromise(reconcileMinterSubscriptions(deps)) - - expect(report).toEqual({ added: [], error: 'realm unreachable' }) - expect(deps.subscribeCalls.value).toEqual([]) -}) - -test('captures subscribe failure after a successful list', async () => { - const deps = buildDeps({ - listUnsubscribedPublicStreams: () => Effect.succeed([{ name: decodeChannelNameSync('commy') }]), - subscribeToStreams: () => Effect.fail(new DepFailure({ message: 'subscribe rejected' })), - }) - - const report = await Effect.runPromise(reconcileMinterSubscriptions(deps)) - - expect(report).toEqual({ added: [], error: 'subscribe rejected' }) -}) - -test('coerces non-Error throws to a string error message', async () => { - const deps = buildDeps({ - listUnsubscribedPublicStreams: () => Effect.fail('not-an-error-instance'), - }) - - const report = await Effect.runPromise(reconcileMinterSubscriptions(deps)) - - expect(report.error).toBe('not-an-error-instance') -}) diff --git a/packages/zulip/minter-reconciler.ts b/packages/zulip/minter-reconciler.ts deleted file mode 100644 index 2e8e8a4..0000000 --- a/packages/zulip/minter-reconciler.ts +++ /dev/null @@ -1,59 +0,0 @@ -import { messageOf } from '@commy/core/messageOf' -import type { ChannelName } from '@commy/core/ports' -import { Effect } from 'effect' - -/** - * Minter subscription reconciler. - * - * Substrates with realm-level subscription (Zulip) require the - * minter to be subscribed to every public stream so the plugin's - * event pump observes events even from lurker (un-acquired) sessions. - * This module is the boot-time pass that closes any - * gap between "every public stream" and the minter's current - * subscription set. Per-session `inbox.subscribe()` continues to - * register the minter for streams created after boot. - * - * Pure logic; substrate I/O is injected. The Zulip adapter wires - * `minterHttp` into `listUnsubscribedPublicStreams` / - * `subscribeToStreams` and exposes the composed call as a method - * on its public shape. - * - * Failure is non-fatal: any failing dep is captured in `error` and - * the boot path keeps running with a degraded lurker view (a - * "log + continue" decision). The deps' typed error - * channel is collapsed into the report, so the reconciler never - * fails — its E channel is `never`. - */ -export interface ReconcilerDeps { - readonly listUnsubscribedPublicStreams: () => Effect.Effect< - ReadonlyArray<{ readonly name: ChannelName }>, - E - > - readonly subscribeToStreams: ( - names: ReadonlyArray, - ) => Effect.Effect, E> -} - -export interface ReconcileReport { - readonly added: ReadonlyArray - readonly error: string | undefined -} - -export const reconcileMinterSubscriptions = ( - deps: ReconcilerDeps, -): Effect.Effect => - deps.listUnsubscribedPublicStreams().pipe( - Effect.flatMap((candidates) => - candidates.length === 0 - ? Effect.succeed({ added: [], error: undefined }) - : deps - .subscribeToStreams(candidates.map((c) => c.name)) - .pipe(Effect.map((added) => ({ added, error: undefined }))), - ), - Effect.catchAll((err) => - Effect.succeed({ - added: [], - error: messageOf(err), - }), - ), - )