diff --git a/packages/mcp/channels-catch-up.test.ts b/packages/mcp/channels-catch-up.test.ts index fb4ab65..6778fb7 100644 --- a/packages/mcp/channels-catch-up.test.ts +++ b/packages/mcp/channels-catch-up.test.ts @@ -54,6 +54,7 @@ const buildMessage = ( readonly channel?: string readonly thread?: string readonly mentions?: ReadonlyArray + readonly sender?: Identity } = {}, ): Message => ({ ref: { @@ -73,7 +74,7 @@ const buildMessage = ( `https://zulip.example.com/#narrow/channel/${opts.channel ?? 'general'}/near/${ts}`, ), }, - sender: human, + sender: opts.sender ?? human, body: decodeMessageBodySync(body), ts: decodeTimestampSync(ts), mentions: userMentions(opts.mentions ?? []), @@ -267,6 +268,25 @@ describe('catchUpChannels', () => { expect(notifier.payloads[0]?.meta['mentioned']).toBe('true') }) + test("the bound bot's own posts are not dispatched back to it", async () => { + const history = buildHistorySpy({ + general: [ + buildMessage(4800, 'from-carol'), + buildMessage(4850, 'from-me', { sender: botIdentity }), + buildMessage(4900, 'carol-again'), + ], + }) + const notifier = buildNotifierSpy() + await runCatchUp({ + intents: [{ kind: 'channel', channelName: decodeChannelNameSync('general') }], + history: history.history, + notifier: notifier.notifier, + botIdentityId: botIdentity.id, + windowSeconds: 400, + }) + expect(notifier.payloads.map((p) => p.content)).toEqual(['from-carol', 'carol-again']) + }) + test('botIdentityId undefined → payload omits mentioned meta but still dispatches', async () => { const history = buildHistorySpy({ general: [buildMessage(4900, 'hi', { mentions: [botIdentity] })], diff --git a/packages/mcp/channels-catch-up.ts b/packages/mcp/channels-catch-up.ts index f3958e8..df09bb8 100644 --- a/packages/mcp/channels-catch-up.ts +++ b/packages/mcp/channels-catch-up.ts @@ -123,8 +123,10 @@ export const catchUpChannels = (deps: ChannelsCatchUpDeps): Effect.Effect m.sender.id !== deps.botIdentityId), Order.mapInput(Order.number, (m: Message) => m.ts), ) yield* Effect.forEach( diff --git a/packages/mcp/identity-cache.test.ts b/packages/mcp/identity-cache.test.ts index af1739d..dfbcb31 100644 --- a/packages/mcp/identity-cache.test.ts +++ b/packages/mcp/identity-cache.test.ts @@ -8,18 +8,7 @@ import { IdentityError, UnboundEphemeralSession, } from '@commy/core/ports' -import { - Cause, - Clock, - Deferred, - Duration, - Effect, - Exit, - Fiber, - Option, - TestClock, - TestContext, -} from 'effect' +import { Cause, Deferred, Effect, Exit, Fiber, Option, TestContext } from 'effect' import type { ProjectSlug, SessionId } from './bootstrap.ts' import { parseSessionId, sanitiseProjectSlug } from './bootstrap.ts' import { createEnsureBound } from './ensure-bound.ts' @@ -107,12 +96,6 @@ const buildAdapterSpy = (): AdapterSpy => { } } -/** - * Run an Effect with the deterministic `TestClock` provided. `ensureBoundFor` - * stamps `lastUsedMs` from `Clock.currentTimeMillis`; under `TestClock` that - * starts at 0 and only advances on `TestClock.adjust`, so the idle-sweep - * assertions are reproducible without an injected `now` lambda. - */ const runTest = (self: Effect.Effect): Promise => Effect.runPromise(self.pipe(Effect.provide(TestContext.TestContext))) @@ -165,23 +148,6 @@ describe('createSingleIdentityCache (persistent mode)', () => { expect([...cache.boundIdentityIds()]).toEqual([buildIdentity('persistent-bot').id]) }), )) - - test('sweepIdle is a no-op for the persistent singleton', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const ensureBound = yield* createEnsureBound({ - acquire: spy.acquire, - name: decodeBotNameSync('persistent-bot'), - }) - const cache = createSingleIdentityCache({ ensureBound }) - const eb = yield* cache.ensureBoundFor(sid('33333333')) - yield* eb() - yield* cache.sweepIdle(Number.POSITIVE_INFINITY) - expect(spy.releaseCalls).toEqual([]) - expect([...cache.boundIdentityIds()]).toEqual([buildIdentity('persistent-bot').id]) - }), - )) }) describe('createEphemeralIdentityCache (ephemeral mode)', () => { @@ -192,7 +158,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('abcdef12')) const acquired = yield* eb() @@ -208,7 +173,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const first = yield* cache.ensureBoundFor(sid('aaaaaaaa')) const second = yield* cache.ensureBoundFor(sid('aaaaaaaa')) @@ -225,7 +189,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const ebA = yield* cache.ensureBoundFor(sid('aaaaaaaa')) yield* ebA() @@ -245,7 +208,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) yield* cache.ensureBoundFor(sid('aaaaaaaa')) // no acquire yet const eb = yield* cache.ensureBoundFor(sid('bbbbbbbb')) @@ -262,7 +224,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) expect([...cache.boundIdentityIds()]).toEqual([]) const eb1 = yield* cache.ensureBoundFor(sid('a1a1a1a1')) @@ -281,7 +242,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb1 = yield* cache.ensureBoundFor(sid('a1f7a1f7')) yield* eb1() @@ -292,97 +252,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { }), )) - test('sweepIdle releases the entry when idle > idleReleaseMs', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - }) - const eb = yield* cache.ensureBoundFor(sid('1d1ee575')) - yield* eb() - expect([...cache.boundIdentityIds()]).toEqual([decodeIdentityIdSync('bot:cc-1d1ee575')]) - yield* TestClock.adjust(Duration.millis(5001)) - const nowMs = yield* Clock.currentTimeMillis - yield* cache.sweepIdle(nowMs) - expect(spy.releaseCalls).toHaveLength(1) - expect([...cache.boundIdentityIds()]).toEqual([]) - }), - )) - - test('sweepIdle is a no-op when entry is fresh', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - }) - const eb = yield* cache.ensureBoundFor(sid('f7e57e57')) - yield* eb() - yield* TestClock.adjust(Duration.millis(4999)) - const nowMs = yield* Clock.currentTimeMillis - yield* cache.sweepIdle(nowMs) - expect(spy.releaseCalls).toEqual([]) - expect([...cache.boundIdentityIds()]).toEqual([decodeIdentityIdSync('bot:cc-f7e57e57')]) - }), - )) - - test('sweepIdle is a no-op when nothing is bound', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - }) - yield* cache.sweepIdle(Number.POSITIVE_INFINITY) - expect(spy.releaseCalls).toEqual([]) - }), - )) - - test('sweepIdle skips an entry that has not yet acquired', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - }) - yield* cache.ensureBoundFor(sid('aaaaaaaa')) // create entry but never call - yield* TestClock.adjust(Duration.millis(10_000)) - const nowMs = yield* Clock.currentTimeMillis - yield* cache.sweepIdle(nowMs) - expect(spy.releaseCalls).toEqual([]) - }), - )) - - test('after sweepIdle releases, next ensureBoundFor for same session_id remints', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - }) - const eb = yield* cache.ensureBoundFor(sid('7e7e7e7e')) - yield* eb() - yield* TestClock.adjust(Duration.millis(6000)) - const nowMs = yield* Clock.currentTimeMillis - yield* cache.sweepIdle(nowMs) - expect(spy.releaseCalls).toHaveLength(1) - const eb2 = yield* cache.ensureBoundFor(sid('7e7e7e7e')) - yield* eb2() - expect(spy.acquireCalls).toEqual(['cc-7e7e7e7e', 'cc-7e7e7e7e']) - }), - )) - test('ensureBoundFor(undefined) returns the unbound stub even when a slot is active (no leak across /clear)', () => runTest( Effect.gen(function* () { @@ -390,7 +259,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('5005ed01')) yield* eb() @@ -410,7 +278,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('5005ed01')) yield* eb() @@ -429,7 +296,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(undefined) const err = yield* captureError(eb()) @@ -444,7 +310,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(undefined) const err = yield* captureError(eb()) @@ -461,7 +326,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('e177e177')) yield* eb() @@ -478,7 +342,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) yield* cache.releaseAllBound() expect(spy.releaseCalls).toEqual([]) @@ -492,7 +355,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('abcdef12'), slug('myproject')) yield* eb() @@ -507,7 +369,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb1 = yield* cache.ensureBoundFor(sid('a3a3a3a3'), slug('myproject-a')) yield* eb1() @@ -524,7 +385,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb1 = yield* cache.ensureBoundFor(sid('aa110110'), slug('first-proj')) const eb2 = yield* cache.ensureBoundFor(sid('aa110110'), slug('second-proj')) @@ -541,7 +401,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('aabbccdd'), slug('commy')) yield* eb() @@ -558,7 +417,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, }) const eb = yield* cache.ensureBoundFor(sid('ba4e1f33')) yield* eb() @@ -574,7 +432,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: (id) => Effect.sync(() => { acquired.push(id) @@ -596,7 +453,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: () => Effect.sync(() => { fired += 1 @@ -619,7 +475,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: (id) => Effect.sync(() => { acquired.push(id.identity.name) @@ -633,32 +488,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { }), )) - test('onAcquire fires after sweepIdle release on re-acquire', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const acquired: string[] = [] - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - onAcquire: (id) => - Effect.sync(() => { - acquired.push(id.identity.name) - }), - }) - const eb = yield* cache.ensureBoundFor(sid('7e7e7e7e')) - yield* eb() - yield* TestClock.adjust(Duration.millis(6000)) - const nowMs = yield* Clock.currentTimeMillis - yield* cache.sweepIdle(nowMs) - expect(acquired).toEqual(['cc-7e7e7e7e']) - const eb2 = yield* cache.ensureBoundFor(sid('7e7e7e7e')) - yield* eb2() - expect(acquired).toEqual(['cc-7e7e7e7e', 'cc-7e7e7e7e']) - }), - )) - test('onAcquire does not fire when acquire rejects', () => runTest( Effect.gen(function* () { @@ -669,7 +498,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { new IdentityError({ operation: 'acquire', cause: new Error('acquire boom') }), ), release: () => Effect.void, - idleReleaseMs: 60_000, onAcquire: () => Effect.sync(() => { fired += 1 @@ -691,7 +519,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: () => Deferred.succeed(reached, undefined).pipe(Effect.zipRight(Deferred.await(gate))), }) @@ -715,7 +542,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: () => Effect.sync(() => { if (shouldThrow) throw new Error('catch-up boom') @@ -732,30 +558,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { }), )) - test('lastUsedMs bumps on each ensureBoundFor for the same sid', () => - runTest( - Effect.gen(function* () { - const spy = buildAdapterSpy() - const cache = yield* createEphemeralIdentityCache({ - acquire: spy.acquire, - release: spy.release, - idleReleaseMs: 5000, - }) - const eb = yield* cache.ensureBoundFor(sid('b00b00b0')) - yield* eb() - // Bump activity at +4000 — within the idle window, so the stamp moves. - yield* TestClock.adjust(Duration.millis(4000)) - yield* cache.ensureBoundFor(sid('b00b00b0')) - // After bump, even at +6000 total the entry is only 2000ms idle. - yield* TestClock.adjust(Duration.millis(2000)) - const nowMs = yield* Clock.currentTimeMillis - yield* cache.sweepIdle(nowMs) - expect(spy.releaseCalls).toEqual([]) - }), - )) - - // ─── onAcquire hook ──────────────────────────────────────────────────── - test('onAcquire fires once per fresh slot, receives identity + project', () => runTest( Effect.gen(function* () { @@ -764,7 +566,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: (id, project) => Effect.sync(() => { calls.push({ name: id.identity.name, project }) @@ -786,7 +587,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: (id, project) => Effect.sync(() => { calls.push({ name: id.identity.name, project }) @@ -811,7 +611,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: (_id, project) => Effect.sync(() => { calls.push({ project }) @@ -831,7 +630,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: () => Effect.sync(() => { fires += 1 @@ -860,7 +658,6 @@ describe('createEphemeralIdentityCache (ephemeral mode)', () => { const cache = yield* createEphemeralIdentityCache({ acquire: spy.acquire, release: spy.release, - idleReleaseMs: 60_000, onAcquire: () => Effect.sync(() => { fires += 1 diff --git a/packages/mcp/identity-cache.ts b/packages/mcp/identity-cache.ts index 2ba02a4..3f97811 100644 --- a/packages/mcp/identity-cache.ts +++ b/packages/mcp/identity-cache.ts @@ -6,7 +6,7 @@ import type { UnknownIdentity, } from '@commy/core/ports' import { UnboundEphemeralSession } from '@commy/core/ports' -import { Clock, Effect, SynchronizedRef } from 'effect' +import { Effect, SynchronizedRef } from 'effect' import type { ProjectSlug, SessionId } from './bootstrap.ts' import { composeBotName } from './bootstrap.ts' import type { EnsureBound } from './ensure-bound.ts' @@ -30,8 +30,7 @@ export type EnsureBoundError = UnknownIdentity | IdentityError | UnboundEphemera * is set; the bot lives for the child's lifetime. * - **Ephemeral**: at-most-one slot keyed by `session_id`. A new sid * releases the prior identity (Zulip deactivates the bot) before - * acquiring the new one. Idle sweep releases the slot after N - * minutes of inactivity. The slot collapses to a single entry — + * acquiring the new one. The slot collapses to a single entry — * see the 1-slot simplification note for why this isn't a * `Map`. * @@ -51,8 +50,8 @@ export interface IdentityCache { * derived from the hook-injected cwd at the boundary in * `tools.ts`. Process-level project state is gone — that was * the leak source. - * - `state.sessionId === sid` → bump `lastUsedMs`, return existing - * (project arg is ignored; the slot is named once). + * - `state.sessionId === sid` → return existing (project arg is + * ignored; the slot is named once). * - `state.sessionId !== sid` → release the prior identity (if * acquired), then mint a fresh entry for the new sid using * the new call's `project`. @@ -84,13 +83,6 @@ export interface IdentityCache { * shutdown body when the MCP child is exiting; idempotent. */ releaseAllBound(): Effect.Effect - /** - * If the slot's last activity is older than `idleReleaseMs`, release - * the bound identity and clear the slot. Called by a periodic timer - * the server boot wires up in ephemeral mode. Persistent mode is a - * no-op. - */ - sweepIdle(nowMs: number): Effect.Effect } export interface SingleIdentityCacheDeps { @@ -110,8 +102,6 @@ export const createSingleIdentityCache = (deps: SingleIdentityCacheDeps): Identi // extra teardown to do; keeping these present means server-boot wiring // stays symmetric across the two factories. releaseAllBound: () => Effect.void, - // Persistent identities live for the child's lifetime. - sweepIdle: () => Effect.void, } } @@ -126,13 +116,11 @@ export interface EphemeralIdentityCacheDeps { ) => Effect.Effect /** * Adapter-side release (`adapter.identity.release`). Called on - * session transitions (release-then-acquire) and idle sweeps. The + * session transitions (release-then-acquire). The * adapter is single-bound, so there's no per-name release argument * — release affects whichever identity is currently bound. */ readonly release: () => Effect.Effect - /** Idle threshold for `sweepIdle`. */ - readonly idleReleaseMs: number /** * Post-acquire hook. Fires after every * successful adapter-side acquire — both the fresh-slot path (first @@ -168,7 +156,6 @@ export interface EphemeralIdentityCacheDeps { interface Slot { readonly sessionId: SessionId readonly ensureBound: EnsureBound - readonly lastUsedMs: number } const unboundStub: EnsureBound = Object.assign( @@ -218,7 +205,6 @@ export const createEphemeralIdentityCache = ( const mintSlot = ( sessionId: SessionId, project: ProjectSlug | undefined, - nowMs: number, prior: Slot | undefined, ): Effect.Effect, Slot]> => { const name = deriveBotName(sessionId, project) @@ -227,11 +213,7 @@ export const createEphemeralIdentityCache = ( acquire, name, afterAcquire: afterAcquireFor(project, sessionId), - }).pipe( - Effect.map( - (ensureBound) => [ensureBound, { sessionId, ensureBound, lastUsedMs: nowMs }] as const, - ), - ) + }).pipe(Effect.map((ensureBound) => [ensureBound, { sessionId, ensureBound }] as const)) } // Atomic read-release-clear: the effectful release runs inside the @@ -248,11 +230,9 @@ export const createEphemeralIdentityCache = ( ) return { - // Reads the activity stamp from Effect's `Clock` (the same source - // `forkIdleSweep` feeds `sweepIdle`), so the idle comparison is - // consistent and `TestClock`-drivable. The slot swap is serialised - // through `SynchronizedRef.modifyEffect`, which also allocates the new - // entry's `EnsureBound` inside the atomic region. + // The slot swap is serialised through `SynchronizedRef.modifyEffect`, + // which also allocates the new entry's `EnsureBound` inside the atomic + // region. ensureBoundFor: (sessionId, project): Effect.Effect> => sessionId === undefined ? // Refuse to surface the slot when no sid was supplied. @@ -263,20 +243,15 @@ export const createEphemeralIdentityCache = ( // preserved: a follow-up call with the original sid still reaches // the same binding. Effect.succeed(unboundStub) - : Clock.currentTimeMillis.pipe( - Effect.flatMap((nowMs) => - SynchronizedRef.modifyEffect(slotRef, (slot) => { - if (slot !== undefined && slot.sessionId === sessionId) { - const bumped: Slot = { ...slot, lastUsedMs: nowMs } - return Effect.succeed([slot.ensureBound, bumped] as const) - } - // Fresh sid or a transition off the prior slot. The prior - // slot's release is captured into the new entry's first - // acquire (see `acquireForTransition`). - return mintSlot(sessionId, project, nowMs, slot) - }), - ), - ), + : SynchronizedRef.modifyEffect(slotRef, (slot) => { + if (slot !== undefined && slot.sessionId === sessionId) { + return Effect.succeed([slot.ensureBound, slot] as const) + } + // Fresh sid or a transition off the prior slot. The prior + // slot's release is captured into the new entry's first + // acquire (see `acquireForTransition`). + return mintSlot(sessionId, project, slot) + }), boundIdentityIds: () => { const slot = Effect.runSync(SynchronizedRef.get(slotRef)) if (slot === undefined) return new Set() @@ -284,17 +259,5 @@ export const createEphemeralIdentityCache = ( return current === undefined ? new Set() : new Set([current.identity.id]) }, releaseAllBound: () => releaseBoundSlot, - sweepIdle: (nowMs) => - // Idle check and release are one atomic region — a slot a request - // fiber bumped (or replaced) between fork and sweep is re-read here, - // so the sweep only releases a slot still genuinely idle past the - // threshold; otherwise it leaves the slot untouched. - SynchronizedRef.updateEffect(slotRef, (slot) => - slot !== undefined && - slot.ensureBound.current() !== undefined && - nowMs - slot.lastUsedMs > deps.idleReleaseMs - ? Effect.as(deps.release(), undefined) - : Effect.succeed(slot), - ), } }) diff --git a/packages/mcp/server.test.ts b/packages/mcp/server.test.ts index 1488a4e..04301dc 100644 --- a/packages/mcp/server.test.ts +++ b/packages/mcp/server.test.ts @@ -10,6 +10,7 @@ import type { IdentityId as IdentityIdType, IdentityOrigin, IdentityPort, + InboundEvent, MessageInbox, MessagePublisher, Presence, @@ -19,11 +20,17 @@ import type { } from '@commy/core/ports' import { type ChannelName, + ChannelPermalinkSchema, + decodeChannelIdSync, decodeChannelNameSync, decodeDisplayNameSync, decodeIdentityIdSync, + decodeMessageBodySync, + decodeMessageIdSync, decodeThreadNameSync, + decodeTimestampSync, InboxError, + MessagePermalinkSchema, } from '@commy/core/ports' import { memoryAdapter } from '@commy/memory/adapter' import { @@ -34,7 +41,6 @@ import { Layer, Option, Ref, - Scope, Stream, TestClock, TestContext, @@ -42,7 +48,6 @@ import { import { EnvConfigError, NotInRepo, parseEnv, substrateAdapterLayer } from './bootstrap.ts' import type { CursorStore } from './cursor-store.ts' import { CursorStoreTag } from './cursor-store.ts' -import type { IdentityCache } from './identity-cache.ts' // Above-port unit tests drive the substrate through hand-rolled port fakes and // the in-memory adapter only — never the real Zulip adapter (see // docs/architecture.md § Test architecture). `completeAsSubstrate` is the single @@ -52,7 +57,7 @@ import type { IdentityCache } from './identity-cache.ts' 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 { clientDisconnect, makeProgram, type ProgramParams } from './server.ts' import type { BindOnDemand } from './session-binder.ts' import { bindThrough, @@ -101,13 +106,13 @@ const inMemorySubscriptionStore = (): SubscriptionStore => { } /** - * Run the boot program with a substituted adapter. The + * Build the boot program with a substituted adapter. The * adapter layer is parse-gated like the production `ZulipAdapterLive`, so * an invalid env fails the layer build — the adapter is never acquired * and `close()` never fires — mirroring production exactly. The logger is * provided both at the program edge and to `onAcquire` (cache edge). */ -const runProgram = ( +const provideProgram = ( env: Record, adapter: ZulipAdapter, params: ProgramParams = {}, @@ -121,29 +126,28 @@ const runProgram = ( // has to stay pristine, so a test that cares about a log line passes its own // `captureLogger` and asserts on it. const loggerLayer = params.loggerLayer ?? captureLogger([]) - return Effect.runPromiseExit( - makeProgram({ ...params, loggerLayer }).pipe( - Effect.provide( - Layer.provideMerge( - Layer.mergeAll( - substrateAdapterLayer(parseEnv.pipe(Effect.as(adapter))), - Layer.succeed(CursorStoreTag, inMemoryCursorStore()), - Layer.succeed(SubscriptionStoreTag, inMemorySubscriptionStore()), - SessionIdLive, - binderRef === undefined - ? SessionBinderLive - : Layer.succeed(SessionBinderTag, binderRef), - ResumeOutcomeLive, - Layer.succeed(QueueStateStoreTag, createInMemoryQueueStateStore()), - loggerLayer, - ), - testPlatformLayer(env), + return makeProgram({ ...params, loggerLayer }).pipe( + Effect.provide( + Layer.provideMerge( + Layer.mergeAll( + substrateAdapterLayer(parseEnv.pipe(Effect.as(adapter))), + Layer.succeed(CursorStoreTag, inMemoryCursorStore()), + Layer.succeed(SubscriptionStoreTag, inMemorySubscriptionStore()), + SessionIdLive, + binderRef === undefined ? SessionBinderLive : Layer.succeed(SessionBinderTag, binderRef), + ResumeOutcomeLive, + Layer.succeed(QueueStateStoreTag, createInMemoryQueueStateStore()), + loggerLayer, ), + testPlatformLayer(env), ), ), ) } +const runProgram = (...args: Parameters) => + Effect.runPromiseExit(provideProgram(...args)) + /** Squash a failure Exit to its underlying error for instanceof / message assertions. */ const failureValue = (exit: Exit.Exit): unknown => Exit.isFailure(exit) ? Cause.squash(exit.cause) : undefined @@ -587,6 +591,97 @@ test('a listen-only seat runs its catch-up at boot, with zero tool calls', async expect(caughtUpChannels).toEqual(['home']) }) +// ─── a quiet seat keeps its identity until it exits ─────────────────────────── +// The defect: a seat that made no posting or reacting call for an hour had its +// bot released while its child was still alive. The events queue belongs to +// the seat's own bot, so the seat went deaf; a seat blocked waiting for an +// answer is exactly the one that makes no calls. Its next post then re-acquired +// and replayed hours of history at it. The stream below stands in for that +// queue: it delivers only while the seat still holds its identity. +test('a seat that makes no tool call for hours keeps its identity and goes on receiving', async () => { + const binderRef = await Effect.runPromise(Ref.make>(Option.none())) + const adapter = await Effect.runPromise(memoryAdapter({ bindOnDemand: bindThrough(binderRef) })) + const lifecycle: string[] = [] + let holdsIdentity = false + const peer: Identity = { + id: decodeIdentityIdSync('peer-7'), + name: decodeDisplayNameSync('peer'), + kind: 'agent', + } + const answer: InboundEvent = { + kind: 'message-posted', + message: { + ref: { + id: decodeMessageIdSync('answer-1'), + channel: { + id: decodeChannelIdSync('home'), + name: decodeChannelNameSync('home'), + permalink: ChannelPermalinkSchema.make('https://zulip.example.com/#narrow/channel/home'), + }, + thread: Option.none(), + permalink: MessagePermalinkSchema.make( + 'https://zulip.example.com/#narrow/channel/home/near/answer-1', + ), + }, + sender: peer, + body: decodeMessageBodySync('the answer'), + ts: decodeTimestampSync(1715450000), + mentions: [], + reactions: [], + }, + } + const substrate = completeAsSubstrate( + { + ...adapter, + identity: { + ...adapter.identity, + acquire: (name) => + adapter.identity.acquire(name).pipe( + Effect.tap(() => + Effect.sync(() => { + holdsIdentity = true + lifecycle.push('acquire') + }), + ), + ), + release: (opts) => + adapter.identity.release(opts).pipe( + Effect.tap(() => + Effect.sync(() => { + holdsIdentity = false + lifecycle.push('release') + }), + ), + ), + }, + inbox: { + ...adapter.inbox, + events: () => + Stream.fromEffect(TestClock.adjust(Duration.hours(3))).pipe( + Stream.flatMap(() => (holdsIdentity ? Stream.make(answer) : Stream.empty)), + ), + }, + history: { ...adapter.history, readChannel: () => Effect.succeed([]) }, + }, + { close: async () => {} }, + ) + const exit = await Effect.runPromiseExit( + provideProgram( + { ...lazyEnv, COMMY_SUBSCRIBE: 'home' }, + substrate, + { + notifier: async (payload) => { + lifecycle.push(`delivered ${payload.content}`) + }, + readGitContext: () => Effect.succeed(NotInRepo()), + }, + binderRef, + ).pipe(Effect.provide(TestContext.TestContext)), + ) + expect(Exit.isSuccess(exit)).toBe(true) + expect(lifecycle).toEqual(['acquire', 'delivered the answer', 'release']) +}) + test('main aborts non-zero when COMMY_SUBSCRIBE contains a malformed token', async () => { const fake = buildFakeAdapter() const env = { @@ -773,64 +868,3 @@ test('clientDisconnect resolves once and detaches its listeners after EOF', asyn expect(stdin.listenerCount('end')).toBe(0) expect(stdin.listenerCount('close')).toBe(0) }) - -// ─── ephemeral idle sweep on Clock + Schedule ────────────── - -interface SweepSpyCache { - readonly cache: Pick - readonly nowMsCalls: number[] -} - -const buildSweepSpyCache = (): SweepSpyCache => { - const nowMsCalls: number[] = [] - return { - nowMsCalls, - cache: { - sweepIdle: (nowMs) => - Effect.sync(() => { - nowMsCalls.push(nowMs) - }), - }, - } -} - -test('forkIdleSweep runs sweepIdle on the spaced schedule, stamping the Clock time', () => - Effect.runPromise( - Effect.scoped( - Effect.gen(function* () { - const spy = buildSweepSpyCache() - const intervalMs = 5 * 60 * 1000 - // Forking happens inside the scope; Schedule.spaced runs the body - // once immediately at fork, then once per interval thereafter. - yield* forkIdleSweep(spy.cache, intervalMs) - // Let the immediate first sweep settle. - yield* TestClock.adjust(Duration.zero) - expect(spy.nowMsCalls).toEqual([0]) - // Each interval advance fires exactly one more sweep, stamped with - // the advancing Clock time (proves the body reads Clock, not Date.now). - yield* TestClock.adjust(Duration.millis(intervalMs)) - expect(spy.nowMsCalls).toEqual([0, intervalMs]) - yield* TestClock.adjust(Duration.millis(intervalMs)) - expect(spy.nowMsCalls).toEqual([0, intervalMs, intervalMs * 2]) - }), - ).pipe(Effect.provide(TestContext.TestContext)), - )) - -test('forkIdleSweep fiber is interrupted when its scope closes (does not block exit)', () => - Effect.runPromise( - Effect.gen(function* () { - const spy = buildSweepSpyCache() - const intervalMs = 5 * 60 * 1000 - const scope = yield* Scope.make() - yield* forkIdleSweep(spy.cache, intervalMs).pipe(Scope.extend(scope)) - yield* TestClock.adjust(Duration.zero) - expect(spy.nowMsCalls).toEqual([0]) - yield* TestClock.adjust(Duration.millis(intervalMs)) - expect(spy.nowMsCalls).toEqual([0, intervalMs]) - // Closing the scope interrupts the forked sweep fiber: no further - // sweeps fire even as time advances past more intervals. - yield* Scope.close(scope, Exit.void) - yield* TestClock.adjust(Duration.millis(intervalMs * 3)) - expect(spy.nowMsCalls).toEqual([0, intervalMs]) - }).pipe(Effect.provide(TestContext.TestContext)), - )) diff --git a/packages/mcp/server.ts b/packages/mcp/server.ts index e924ca8..ce3c08b 100644 --- a/packages/mcp/server.ts +++ b/packages/mcp/server.ts @@ -12,20 +12,7 @@ import { CommandExecutor, FetchHttpClient, FileSystem, type HttpClient } from '@ import { NodeContext, NodeRuntime } from '@effect/platform-node' import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js' import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js' -import type { Scope } from 'effect' -import { - Cause, - Clock, - ConfigProvider, - Data, - Deferred, - Duration, - Effect, - Layer, - Option, - Predicate, - Schedule, -} from 'effect' +import { Cause, ConfigProvider, Data, Deferred, Effect, Layer, Option, Predicate } from 'effect' import type { BotName, GitContext, ParsedEnv, ProjectSlug, SessionId } from './bootstrap.ts' import { readGitContext as defaultReadGitContext, @@ -131,42 +118,6 @@ export interface ProgramParams { /** Race budget for `identity.release()` in the shutdown finalizer. */ const RELEASE_TIMEOUT_MS = 5000 -/** - * Idle timeout for the ephemeral identity cache. Sessions that go an hour - * without an attribution-producing tool call get their bot deactivated and - * the slot cleared. A returning session re-acquires via Zulip's reactivate - * + regenerate-api-key path. - */ -const EPHEMERAL_IDLE_RELEASE_MS = 60 * 60 * 1000 -const EPHEMERAL_IDLE_SWEEP_INTERVAL_MS = 5 * 60 * 1000 - -/** - * Fork the ephemeral idle sweep onto a periodic schedule scoped to the - * enclosing fiber Scope: - * - * - `Schedule.spaced` keeps `intervalMs` between sweeps. (`Effect.repeat` - * runs the body once at fork too; the boot-time sweep is a no-op since - * no slot is idle yet.) - * - `Effect.forkScoped` ties the sweep fiber to the caller's Scope, so - * when that Scope closes (the pump's scoped region unwinds on natural - * end or signal-driven interrupt) the fiber is interrupted. - * - The forked fiber never blocks process exit: it's a child of the - * Scope, not the main fiber's join set. - * - * The sweep reads the current wall-clock from Effect's `Clock` rather - * than `Date.now()`, so tests can drive it deterministically with - * `TestClock`. - */ -export const forkIdleSweep = ( - cache: Pick, - intervalMs: number, -): Effect.Effect => { - const sweep = Clock.currentTimeMillis.pipe(Effect.flatMap((nowMs) => cache.sweepIdle(nowMs))) - return Effect.asVoid( - Effect.forkScoped(Effect.repeat(sweep, Schedule.spaced(Duration.millis(intervalMs)))), - ) -} - /** * Type-2 default sub set for interactive CC sessions. Fires * once per ephemeral slot, right after the substrate-side acquire @@ -367,7 +318,6 @@ const buildIdentityCache = ( return createEphemeralIdentityCache({ acquire: adapter.identity.acquire, release: adapter.identity.release, - idleReleaseMs: EPHEMERAL_IDLE_RELEASE_MS, ...(ephemeralOnAcquire !== undefined ? { onAcquire: ephemeralOnAcquire } : {}), }) } @@ -880,9 +830,6 @@ export const makeProgram = ( : Effect.void, ) - // Ephemeral mode runs a periodic idle sweep (forked into this scope). - const runsIdleSweep = parsed.botName === undefined - // Boot-time subscribe now binds: a subscription and its events queue are // realm state under the seat's own principal, so the seam needs this // seat's naming inputs in the fiber-local context the bind reads. @@ -1077,13 +1024,7 @@ export const makeProgram = ( // finalizer so LIFO teardown runs pump-cancel first. yield* Effect.addFinalizer(() => pump.cancel) - // Ephemeral idle sweep forked into this scope — interrupted when the - // scope closes (pump end or signal-driven interrupt). - if (runsIdleSweep) { - yield* forkIdleSweep(identityCache, EPHEMERAL_IDLE_SWEEP_INTERVAL_MS) - } - - // Boot-forked narrow-set rebuild, scope-tied like the sweep. Asks the + // Boot-forked narrow-set rebuild, scope-tied. 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 diff --git a/packages/mcp/subscription-resume.test.ts b/packages/mcp/subscription-resume.test.ts index b97e27d..b3a9be0 100644 --- a/packages/mcp/subscription-resume.test.ts +++ b/packages/mcp/subscription-resume.test.ts @@ -334,7 +334,6 @@ const buildPersistRig = (): Effect.Effect => const identityCache = yield* createEphemeralIdentityCache({ acquire: adapter.identity.acquire, release: adapter.identity.release, - idleReleaseMs: 60 * 60 * 1000, }) yield* installBinder(binderRef, binderFor(identityCache)) const narrowSet = createNarrowSet() diff --git a/packages/mcp/tools-session.test.ts b/packages/mcp/tools-session.test.ts index 259f78f..c03e649 100644 --- a/packages/mcp/tools-session.test.ts +++ b/packages/mcp/tools-session.test.ts @@ -26,13 +26,11 @@ interface SessionRig { const buildSessionRig = ( options: { - readonly idleReleaseMs?: number readonly projectForCwd?: (cwd: string | undefined) => Effect.Effect readonly feedSessionId?: (sessionId: SessionId) => Effect.Effect } = {}, ): Effect.Effect => Effect.gen(function* () { - const idleReleaseMs = options.idleReleaseMs ?? 60 * 60 * 1000 // The mint seam, wired as `server.ts` wires it. The adapter reads through a // holder; the binder is installed once the cache exists, because the cache // is built FROM the adapter's acquire. Nothing in the tool layer decides @@ -42,7 +40,6 @@ const buildSessionRig = ( const identityCache = yield* createEphemeralIdentityCache({ acquire: adapter.identity.acquire, release: adapter.identity.release, - idleReleaseMs, }) yield* installBinder(binderRef, binderFor(identityCache)) const narrowSet = createNarrowSet()