diff --git a/packages/client/runtime/src/client/sessions/manager.ts b/packages/client/runtime/src/client/sessions/manager.ts index 4b1fecacc3..b8a3f91e2d 100644 --- a/packages/client/runtime/src/client/sessions/manager.ts +++ b/packages/client/runtime/src/client/sessions/manager.ts @@ -232,8 +232,11 @@ export class SessionManager { * truth — a later get() lazily rebuilds and open() backfills history. * @param sessionId - the session to drop. */ - drop(sessionId: SessionId): void { + drop(sessionId: SessionId): Promise { + const session = this.sessions.get(sessionId) + if (session === undefined) return Promise.resolve() this.sessions.delete(sessionId) + return session.dispose() } /** diff --git a/packages/client/runtime/src/client/sessions/service.ts b/packages/client/runtime/src/client/sessions/service.ts index d654d65085..dd2036449b 100644 --- a/packages/client/runtime/src/client/sessions/service.ts +++ b/packages/client/runtime/src/client/sessions/service.ts @@ -266,6 +266,8 @@ export class SessionRuntime implements ISessions { private watched: SessionId | undefined /** Removed-while-staged sessions whose teardown waits for the stage to move away. */ private readonly deferredRemovals = new Set() + /** Scope and journal teardowns started by synchronous list projection. */ + private readonly scopeDisposals = new Set>() /** * @param ctx - client root context (scope fibers mount under it). @@ -343,6 +345,14 @@ export class SessionRuntime implements ISessions { } }, 'sessions: conversation registry rebuild') } + rootCtx.effect(() => async () => { + for (const [id, record] of this.scopes) { + this.scopes.delete(id) + this.deferredRemovals.delete(id) + this.dropScope(id, record) + } + await Promise.all(this.scopeDisposals) + }, 'sessions: scoped resources') rootCtx.reflect.provide('sessions', this, undefined) } @@ -491,7 +501,7 @@ export class SessionRuntime implements ISessions { this.manager.handleSessionError(...args) } - /** Rebuild the Session baseline and every opened window after connection. */ + /** Refresh Session and subagent catalogs after connection; opened journals resume independently. */ handleConnected(): void { this.manager.handleConnected() } @@ -780,14 +790,25 @@ export class SessionRuntime implements ISessions { * durable truth, a reopen lazily rebuilds and backfills via open(). */ private dropScope(id: SessionId, record: ScopeRecord): void { - void record.fiber.dispose() // Release the Session's dispatch point with the scope it belongs to (a // surviving instance — the live Intent — rebinds when resolve re-mints). record.session.unbindScope() // Optional lookup: slots and sessions are sibling services with no // declared dependency; a slots-less boot (object-layer tests) skips. this.rootCtx.get('slots')?.pruneStoreScope(id) - this.manager.drop(id) + this.trackScopeDisposal(id, 'scope fiber', record.fiber.dispose()) + this.trackScopeDisposal(id, 'journal', this.manager.drop(id)) + } + + /** Retain one asynchronous scope cleanup through runtime disposal and contain its failure. */ + private trackScopeDisposal(id: SessionId, part: string, task: void | Promise): void { + const tracked = Promise.resolve(task).catch((error: unknown) => { + this.rootCtx.logger.warn( + `client-runtime: Session ${JSON.stringify(id)} ${part} cleanup failed: ${error instanceof Error ? error.message : String(error)}`, + ) + }) + this.scopeDisposals.add(tracked) + void tracked.then(() => { this.scopeDisposals.delete(tracked) }) } /** Run deferred teardowns whose session is no longer staged (called when the stage moves). */ diff --git a/packages/client/runtime/src/client/sessions/session.ts b/packages/client/runtime/src/client/sessions/session.ts index 6e269068a4..bff1f91f5d 100644 --- a/packages/client/runtime/src/client/sessions/session.ts +++ b/packages/client/runtime/src/client/sessions/session.ts @@ -538,12 +538,15 @@ export class Session implements SessionFace { this.notifier.markDirty() } - /** Stop the Session's live Remote source. */ - dispose(): void { + /** + * Stop the Session's live Remote source. + * @returns when the active journal generation and consumer are quiescent. + */ + dispose(): Promise { this.openGeneration++ const events = this.events this.events = undefined - void events?.dispose() + return events?.dispose() ?? Promise.resolve() } /** Rebuild the current window after a low-frequency Definition or view registration change. */ diff --git a/packages/client/runtime/tests/client-apply.client.spec.ts b/packages/client/runtime/tests/client-apply.client.spec.ts index 329536541c..cfc6dff0a7 100644 --- a/packages/client/runtime/tests/client-apply.client.spec.ts +++ b/packages/client/runtime/tests/client-apply.client.spec.ts @@ -17,6 +17,7 @@ import { FakeApiClient, fakeRemote, ok } from './fake-api.client.ts' interface Bench { ctx: Context api: FakeApiClient + runtime: { dispose(): Promise } start: ReturnType> dispatchRemote(event: string, args: readonly unknown[]): void } @@ -31,7 +32,6 @@ async function mount(configure?: (api: FakeApiClient) => void): Promise { for (const listener of listeners.get(event) ?? []) listener(...args as never[]) } const start = vi.fn(() => ({ stop: () => {} })) - const bench: Bench = { ctx, api, start, dispatchRemote } const handle: ConnectionHandle = { api, isLoopback: true, @@ -59,8 +59,8 @@ async function mount(configure?: (api: FakeApiClient) => void): Promise { ctx.reflect.provide('remote.commands', remote.commands) ctx.reflect.provide('remote.session', remote.session) ctx.reflect.provide('remote.workspace', remote.workspace) - await ctx.plugin(RuntimeClient).await() - return bench + const runtime = await ctx.plugin(RuntimeClient).await() + return { ctx, api, runtime, start, dispatchRemote } } async function flushMicrotasks(): Promise { @@ -171,7 +171,17 @@ describe('runtime client apply', () => { it('does not own the Connection loop and closes its Remote streams on unload', async () => { const bench = await mount() - await bench.ctx.fiber.dispose() + const sessions = bench.ctx.get('sessions') as SessionRuntime + bench.dispatchRemote('api-session/added', [{ + sessionId: 's-open', updatedAt: 1, running: false, blank: false, + }]) + await flushMicrotasks() + sessions.open('s-open' as never) + await vi.waitFor(() => { expect(bench.api.activeFollows('s-open' as never)).toBe(1) }) + + await bench.runtime.dispose() + expect(bench.start).not.toHaveBeenCalled() + expect(bench.api.activeFollows('s-open' as never)).toBe(0) }) }) diff --git a/packages/client/runtime/tests/fake-api.client.ts b/packages/client/runtime/tests/fake-api.client.ts index fc57e76fb8..45cc2e30cd 100644 --- a/packages/client/runtime/tests/fake-api.client.ts +++ b/packages/client/runtime/tests/fake-api.client.ts @@ -122,6 +122,8 @@ export function fakeRemote(api = new FakeApiClient()): RuntimeRemotes { export class FakeApiClient implements IApiClient { /** Chronological call record: [method, payload]. */ readonly calls: { method: string; payload: unknown }[] = [] + /** Session ids in physical follow-generation opening order. */ + readonly followStarts: SessionId[] = [] // Programmable slots (defaults answer OK-empty); reassign per case. onList: (payload: unknown) => Promise> = () => Promise.resolve(ok({ items: [] })) @@ -388,6 +390,11 @@ export class FakeApiClient implements IApiClient { return this.calls.filter(c => c.method === method).map(c => c.payload) } + /** Number of currently attached journal generations for one Session. */ + activeFollows(sessionId: SessionId): number { + return this.followConns.get(sessionId)?.length ?? 0 + } + private record(method: string, payload: unknown, response: Promise): Promise { this.calls.push({ method, payload }) return response @@ -455,6 +462,7 @@ export class FakeApiClient implements IApiClient { signal: AbortSignal = new AbortController().signal, ): AsyncGenerator { const sessionId = addressSessionId(request.address) + this.followStarts.push(sessionId) const key = addressKey(request.address) const initialPage = this.onHistory({ sessionId, maxMessages: 50 }) this.openingPages.set(key, initialPage) diff --git a/packages/client/runtime/tests/sessions-service.client.spec.ts b/packages/client/runtime/tests/sessions-service.client.spec.ts index ab2638847f..75d1690fce 100644 --- a/packages/client/runtime/tests/sessions-service.client.spec.ts +++ b/packages/client/runtime/tests/sessions-service.client.spec.ts @@ -172,6 +172,30 @@ describe('scope tree', () => { b.svc.open(sid('s2')) // stage moves; sweep must NOT tear down the re-listed s1 expect(b.svc.scope(sid('s1'))).toBe(scoped) }) + + it('closes an opened journal when its removed scope drops', async () => { + const b = bench() + await feedList(b, [{ id: 's1' }]) + b.svc.open(sid('s1')) + const session = b.svc.binding(sid('s1'))?.session + if (session === undefined) throw new Error('expected the selected Session binding') + await vi.waitFor(() => { expect(b.api.activeFollows(sid('s1'))).toBe(1) }) + const notified = vi.fn() + session.subscribe(notified) + + await feedList(b, []) + await feedList(b, [{ id: 's2' }]) + b.svc.open(sid('s2')) + + await vi.waitFor(() => { expect(b.api.activeFollows(sid('s1'))).toBe(0) }) + await b.api.pushFollow(sid('s1'), { + type: 'event', + event: { seq: 0, timestamp: 0, type: 'turn/start', data: { turn: 0 } } as never, + }) + await Promise.resolve() + expect(b.api.followStarts.filter(id => id === sid('s1'))).toHaveLength(1) + expect(notified).not.toHaveBeenCalled() + }) }) describe('current selection (migrated from ui-layout, arbitrated into the list snapshot)', () => { @@ -324,13 +348,17 @@ describe('cell (render-layer session kit)', () => { b.svc.binding(sid('s1')) expect(historyCalls()).toHaveLength(0) b.svc.open(sid('s1')) - expect(historyCalls().map(c => (c.payload as { sessionId: string }).sessionId)).toEqual(['s1']) + await vi.waitFor(() => { + expect(historyCalls().map(c => (c.payload as { sessionId: string }).sessionId)).toEqual(['s1']) + }) // Same current again: no second pull. b.svc.open(sid('s1')) expect(historyCalls()).toHaveLength(1) // Stage moves: the new occupant opens. b.svc.open(sid('s2')) - expect(historyCalls().map(c => (c.payload as { sessionId: string }).sessionId)).toEqual(['s1', 's2']) + await vi.waitFor(() => { + expect(historyCalls().map(c => (c.payload as { sessionId: string }).sessionId)).toEqual(['s1', 's2']) + }) }) it('startup restore: a persisted selection validated by the first projection opens its window unprompted', async () => { @@ -345,8 +373,10 @@ describe('cell (render-layer session kit)', () => { const b = bench() expect(b.api.calls.filter(c => c.method === 'session.history')).toHaveLength(0) await feedList(b, [{ id: 's1' }]) // projection validates the persisted id → current lands → stage follows - const historyCalls = b.api.calls.filter(c => c.method === 'session.history') - expect(historyCalls.map(c => (c.payload as { sessionId: string }).sessionId)).toEqual(['s1']) + await vi.waitFor(() => { + const historyCalls = b.api.calls.filter(c => c.method === 'session.history') + expect(historyCalls.map(c => (c.payload as { sessionId: string }).sessionId)).toEqual(['s1']) + }) } finally { vi.unstubAllGlobals() } @@ -709,7 +739,7 @@ describe('coverage tails (branch duals)', () => { await feedList(b, [{ id: 's1' }]) b.svc.open(sid('s1')) const historyCalls = () => b.api.calls.filter(c => c.method === 'session.history') - expect(historyCalls()).toHaveLength(1) + await vi.waitFor(() => { expect(historyCalls()).toHaveLength(1) }) await feedList(b, []) // removed while staged: current masks to undefined, stage holds → deferred expect(b.svc.scope(sid('s1'))).toBeDefined() // Resurfacing re-projects current = s1: same stage occupant, no second pull.