fix(client-runtime): close journals with session scopes

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<void> {
+    const session = this.sessions.get(sessionId)
+    if (session === undefined) return Promise.resolve()
     this.sessions.delete(sessionId)
+    return session.dispose()
   }
 
   /**

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<SessionId>()
+  /** Scope and journal teardowns started by synchronous list projection. */
+  private readonly scopeDisposals = new Set<Promise<void>>()
 
   /**
    * @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>): 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). */

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<void> {
     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. */

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<void> }
   start: ReturnType<typeof vi.fn<ConnectionHandle['start']>>
   dispatchRemote(event: string, args: readonly unknown[]): void
 }
@@ -31,7 +32,6 @@ async function mount(configure?: (api: FakeApiClient) => void): Promise<Bench> {
     for (const listener of listeners.get(event) ?? []) listener(...args as never[])
   }
   const start = vi.fn<ConnectionHandle['start']>(() => ({ 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<Bench> {
   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<void> {
@@ -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)
   })
 })

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<RpcResponse<{ items: never[] }>> = () => 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<T>(method: string, payload: unknown, response: Promise<T>): Promise<T> {
     this.calls.push({ method, payload })
     return response
@@ -455,6 +462,7 @@ export class FakeApiClient implements IApiClient {
     signal: AbortSignal = new AbortController().signal,
   ): AsyncGenerator<SessionFollowFrame> {
     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)

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.