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), - }), - ), - )