test(session): cover the split controllers and passive follow

Move the facade, listener, state, wait and history tests onto the new
controllers, and cover passive activation, omitted cursors and the
event subscription frames.

apps/web/tests/preset-migration.snapshot.ts

@@ -85,11 +85,11 @@ describe.skipIf(webSnapshotMode() === 'record')('historical preset restoration t
       await expect(readFile(childSuccessor)).rejects.toMatchObject({ code: 'ENOENT' })
       const signal = new AbortController().signal
       const address = { kind: 'subagent', parentSessionId: id, childSessionId: childId, mode: 'one-shot' } as const
-      const childPage = await scaffold.ctx.sessionController.page({ address, throughSeq: childRows.length - 1 }, signal)
+      const childPage = await scaffold.ctx.sessionController.events({ address, throughSeq: childRows.length - 1 }, signal)
       expect(childPage.records.map(record => record.event.type)).toEqual(childRows.map(row => row['type']))
-      await expect(scaffold.ctx.sessionController.page({ address: { ...address, childSessionId: brokenId, mode: 'unknown' }, throughSeq: 1 }, signal))
+      await expect(scaffold.ctx.sessionController.events({ address: { ...address, childSessionId: brokenId, mode: 'unknown' }, throughSeq: 1 }, signal))
         .rejects.toThrow()
-      await expect(scaffold.ctx.sessionController.page({ address: { kind: 'session', sessionId: id }, throughSeq: rows.length - 1 }, signal))
+      await expect(scaffold.ctx.sessionController.events({ address: { kind: 'session', sessionId: id }, throughSeq: rows.length - 1 }, signal))
         .resolves.toMatchObject({ hasMore: false })
       const childWriter = await scaffold.ctx.sessionPersistence.open(childId, 'write')
       await childWriter.close()

benchmarks/session-open/session-open.worker.ts

@@ -29,7 +29,8 @@ import SessionTitleService from '@deepseek-ai/dsh-session-title'
 import * as SessionTurnOutlinePlugin from '@deepseek-ai/dsh-session-turn-outline'
 import TokenMeter from '@deepseek-ai/dsh-token-meter'
 // These Host-only adapters have no public Node export and are compiled into the benchmark worker.
-import { SessionHistoryController } from '../../packages/api/session-controller/src/history.ts'
+import { HistoryController } from '../../packages/api/session-controller/src/history/controller.ts'
+import { WebController } from '../../packages/api/session-controller/src/web/controller.ts'
 import { assertBuiltBenchmarkRuntime } from '../support/built-worker.ts'
 import { SYNTHETIC_SESSION_ID } from './session-open.constants.ts'
 
@@ -151,7 +152,7 @@ class SessionBenchmarkHost {
   private constructor(
     private readonly ctx: Context,
     private readonly scenario: SessionOpenBenchmarkScenario,
-    private readonly history: SessionHistoryController | undefined,
+    private readonly history: WebController | undefined,
   ) {}
 
   static async create(root: string, scenario: SessionOpenBenchmarkScenario): Promise<SessionBenchmarkHost> {
@@ -165,10 +166,10 @@ class SessionBenchmarkHost {
     }
     await installProjectionSet(ctx)
     await ctx.plugin(JsonlSessionPersistence, { root, compression: 'zstd' })
-    let history: SessionHistoryController | undefined
+    let history: WebController | undefined
     if (scenario === 'first-history') {
       new BenchmarkSessionQuery(ctx)
-      history = new SessionHistoryController(ctx, (observation) => {
+      history = new WebController(ctx, new HistoryController(ctx), (observation) => {
         // First-history ends at snapshot delivery; Agent-resume owns live activation and retention.
         observation[Symbol.dispose]()
       })

packages/api/session-controller/tests/client-contract.client.spec.ts

@@ -1,6 +1,6 @@
 import { describe, expect, expectTypeOf, it, vi } from 'vitest'
 import type { PromptContentPart as AttachmentPromptContentPart } from '@deepseek-ai/dsh-attachment/types'
-import { SessionSeq, type SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
+import { SessionSeq, type SessionId, type SessionSeqCursor } from '@deepseek-ai/dsh-session/types'
 import {
   MutableSessionEventSource, type SessionAssistantSettlementEntry,
   type SessionLiveEventEntry, type SessionTransientEventEntry,
@@ -11,6 +11,7 @@ import type { ProjectionsBaseline } from '../src/client/sessions/projection-stor
 import { ProjectionValueStore } from '../src/client/sessions/projection-store.ts'
 import type { PromptContentPart as SessionPromptContentPart, SessionPageRequest } from '../src/types.ts'
 import { ev, plainTurn } from './event-script.client.ts'
+import { history, pageRule } from './remote/session.client.ts'
 
 type RenameSuccess = Extract<Awaited<ReturnType<ISession['rename']>>, { readonly ok: true }>
 
@@ -70,7 +71,21 @@ describe('Client Session contracts', () => {
     expectTypeOf<RenameSuccess['value']['seq']>().toEqualTypeOf<SessionSeq>()
     expectTypeOf<ProjectionValueStore['apply']>().parameter(2).toEqualTypeOf<SessionSeqCursor>()
     expectTypeOf<ProjectionsBaseline['asOfSeq']>().toEqualTypeOf<SessionSeqCursor>()
-    expectTypeOf<SessionPageRequest['throughSeq']>().toEqualTypeOf<number>()
+    expectTypeOf<SessionPageRequest['throughSeq']>().toEqualTypeOf<number | undefined>()
+  })
+
+  it('returns the current page cursor when throughSeq is omitted and preserves explicit cuts', async () => {
+    const address = { kind: 'session', sessionId: 'page-fixture' as SessionId } as const
+    const rule = pageRule(history(plainTurn(SessionSeq(0), 1, 'question', 'answer')))
+    const current = await rule({ address })
+    expect(current).toMatchObject({ ok: true, value: { cursor: 5, hasMore: false } })
+    const cut = await rule({ address, throughSeq: 2 })
+    expect(cut).toMatchObject({ ok: true, value: { cursor: 2, hasMore: false } })
+    if (!cut.ok) throw new Error('expected a successful history page')
+    expect(cut.value.records.map(record => record.event.seq)).toEqual([0, 1, 2])
+    await expect(rule({ address, throughSeq: -1 })).resolves.toEqual({
+      ok: true, value: { cursor: -1, records: [], hasMore: false },
+    })
   })
 
   it('keeps its text and image prompt parts identical to attachment intake', () => {

packages/api/session-controller/tests/controller.host.spec.ts

@@ -3,7 +3,8 @@ import { Context } from '@deepseek-ai/cordis'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { mountAgentLoopTestDependencies, mountAgentLoopTestHarness } from '@deepseek-ai/dsh-agent-loop-testkit'
-import { createUserMessage } from '@deepseek-ai/dsh-llm'
+import { LlmAttemptId, createUserMessage } from '@deepseek-ai/dsh-llm'
+import type TypertGatewayService from '@deepseek-ai/dsh-api-gateway'
 import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
@@ -22,6 +23,75 @@ describe('SessionController facade', () => {
     expect(SessionController.inject).not.toContain('tools')
   })
 
+  it('keeps assistant stream observation live while list initialization awaits', async () => {
+    const ctx = new Context()
+    onTestFinished(() => ctx.fiber.dispose())
+    await mountAgentLoopTestDependencies(ctx)
+    await mountAgentLoopTestHarness(ctx)
+    const gateway: Partial<TypertGatewayService> = { getPendingRemoteEvents: () => [] }
+    ctx.provide('typertGateway', gateway as TypertGatewayService)
+    const session = ctx.sessions.create(SessionId('assistant-interleave'), { meta: { cwd: '/workspace' } })
+    const partial: Partial<Agent> = { id: session.id, session, status: 'running', ctx }
+    const agent = partial as Agent
+    const entered = Promise.withResolvers<undefined>()
+    const gate = Promise.withResolvers<undefined>()
+    const original = ctx.sessionProjections.register.bind(ctx.sessionProjections)
+    let first = true
+    vi.spyOn(ctx.sessionProjections, 'register').mockImplementation(async (...args: Parameters<typeof original>) => {
+      if (first) {
+        first = false
+        entered.resolve(undefined)
+        await gate.promise
+      }
+      return original(...args)
+    })
+    const mounting = createSessionTestController(ctx, defaults)
+    await entered.promise
+    // An Agent start racing the list read must not be lost before the controller publishes.
+    ctx.emit('agent/assistant-stream', {
+      agent,
+      frame: { type: 'start', attemptId: LlmAttemptId('boot-attempt'), revision: 1, turn: 1, step: 1 },
+    })
+    gate.resolve(undefined)
+    const controller = await mounting
+    const frames = controller.follow({
+      address: { kind: 'session', sessionId: session.id }, assistantStream: true,
+    }, new AbortController().signal)[Symbol.asyncIterator]()
+    await expect(frames.next()).resolves.toMatchObject({
+      value: { type: 'snapshot', assistantStream: { revision: 1 } },
+    })
+    await frames.return?.(undefined)
+  })
+
+  it('serves passive reads, subscriptions and waits through its plain controllers', async () => {
+    const ctx = new Context()
+    onTestFinished(() => ctx.fiber.dispose())
+    await mountAgentLoopTestDependencies(ctx)
+    await mountAgentLoopTestHarness(ctx)
+    const gateway: Partial<TypertGatewayService> = { getPendingRemoteEvents: () => [] }
+    ctx.provide('typertGateway', gateway as TypertGatewayService)
+    const controller = await createSessionTestController(ctx, defaults)
+    const sessionId = SessionId('passive-facade')
+    ctx.sessions.create(sessionId, { meta: { cwd: '/workspace' } })
+    const address = { kind: 'session' as const, sessionId }
+    const signal = new AbortController().signal
+
+    await expect(controller.events({ address }, signal)).resolves.toMatchObject({ records: [], hasMore: false })
+    const historyFrames = controller.observeEvent({ address }, signal)[Symbol.asyncIterator]()
+    await expect(historyFrames.next()).resolves.toMatchObject({ value: { type: 'ready' } })
+    await historyFrames.return?.(undefined)
+    const projectionFrames = controller.observeProjections({ sessionId }, signal)[Symbol.asyncIterator]()
+    await expect(projectionFrames.next()).resolves.toMatchObject({ value: { type: 'ready' } })
+    await projectionFrames.return?.(undefined)
+    await expect(controller.state({ sessionId }, signal))
+      .resolves.toMatchObject({ active: false, running: false, queue: null })
+    const stateFrames = controller.observeState({ sessionId }, signal)[Symbol.asyncIterator]()
+    await expect(stateFrames.next()).resolves.toMatchObject({ value: { active: false } })
+    await stateFrames.return?.(undefined)
+    await expect(controller.wait({ sessionId, afterSeq: -1, maxMessages: 10, timeoutMs: 50 }, signal))
+      .resolves.toMatchObject({ reason: 'idle' })
+  })
+
   it('owns Host service methods and publishes Agent lifecycle projections', async () => {
     const ctx = new Context()
     onTestFinished(() => ctx.fiber.dispose())
@@ -164,6 +234,7 @@ describe('SessionController facade', () => {
       const abort = new AbortController()
       const iterator = controller.follow({
         address: { kind: 'session', sessionId },
+        activate: true,
       }, abort.signal)[Symbol.asyncIterator]()
 
       await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'snapshot' } })
@@ -216,6 +287,7 @@ describe('SessionController facade', () => {
     })
     const iterator = controller.follow({
       address: { kind: 'session', sessionId },
+      activate: true,
     }, new AbortController().signal)[Symbol.asyncIterator]()
 
     await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'snapshot' } })

packages/api/session-controller/tests/event-script.client.ts

@@ -8,7 +8,7 @@ import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
 import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import type {
   SessionEventEntry,
-  SessionPage,
+  SessionEventsPage,
   SessionWireEvent,
 } from '../src/types.ts'
 
@@ -185,8 +185,9 @@ export function entries(events: readonly SessionEvent[]): SessionEventEntry[] {
 }
 
 /** Build one view-less history response value. */
-export function historyValue(events: readonly SessionEvent[], hasMore = false): SessionPage {
+export function historyValue(events: readonly SessionEvent[], hasMore = false): SessionEventsPage {
   return {
+    cursor: events.at(-1)?.seq ?? -1,
     records: entries(events),
     hasMore,
   }

packages/api/session-controller/tests/history-controller.host.spec.ts

@@ -0,0 +1,171 @@
+import { Context } from '@deepseek-ai/cordis'
+import { createMessage } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SessionId, SessionSeq } from '@deepseek-ai/dsh-session'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { HistoryController } from '../src/history/controller.ts'
+import { WebController } from '../src/web/controller.ts'
+import { installSessionReadTestServices } from './test-remote.ts'
+
+const roots: Context[] = []
+const aborters: AbortController[] = []
+afterEach(async () => {
+  for (const abort of aborters.splice(0)) abort.abort()
+  await Promise.all(roots.splice(0).map(ctx => ctx.fiber.dispose()))
+  vi.restoreAllMocks()
+})
+async function fixture() {
+  const ctx = new Context()
+  roots.push(ctx)
+  await ctx.plugin(SessionStore)
+  installSessionReadTestServices(ctx)
+  const history = new HistoryController(ctx)
+  const web = new WebController(ctx, history, (observation) => { observation[Symbol.dispose]() })
+  const session = ctx.sessions.create(SessionId('history-target'), { meta: { cwd: '/history-test' } })
+  const abort = new AbortController()
+  aborters.push(abort)
+  const address = { kind: 'session' as const, sessionId: session.id }
+  return { ctx, history, web, session, abort, address }
+}
+
+describe('HistoryController passive reads', () => {
+  it('subscribes without reading an Observation and emits later durable events', async () => {
+    const f = await fixture()
+    const observe = vi.spyOn(f.ctx.sessionQuery, 'observeSession')
+    const stream = f.history.observeEvent({ address: f.address }, f.abort.signal)[Symbol.asyncIterator]()
+    expect(await stream.next()).toEqual({ done: false, value: { type: 'ready' } })
+    expect(observe).not.toHaveBeenCalled()
+    f.session.append('turn/start', { turn: 1 })
+    expect(await stream.next()).toMatchObject({ value: { type: 'event', event: { seq: 0 } } })
+    f.abort.abort()
+    expect(await stream.next()).toMatchObject({ done: true })
+  })
+
+  it('subscribes to sparse projection changes without an Observation', async () => {
+    const f = await fixture()
+    const observe = vi.spyOn(f.ctx.sessionQuery, 'observeSession')
+    let receive: Parameters<typeof f.ctx.sessionProjections.onChanged>[0] | undefined
+    const dispose = vi.fn()
+    vi.spyOn(f.ctx.sessionProjections, 'onChanged').mockImplementation((callback) => {
+      receive = callback
+      return dispose
+    })
+    const stream = f.history.observeProjections({ sessionId: f.session.id }, f.abort.signal)[Symbol.asyncIterator]()
+    expect(await stream.next()).toMatchObject({ value: { type: 'ready' } })
+    receive?.(f.session, 'title', 'new title', SessionSeq(0))
+    expect(await stream.next()).toMatchObject({ value: { type: 'projection', key: 'title', value: 'new title', seq: 0 } })
+    expect(observe).not.toHaveBeenCalled()
+    f.abort.abort()
+    expect(await stream.next()).toMatchObject({ done: true })
+    expect(dispose).toHaveBeenCalledOnce()
+  })
+
+  it('reads newest assistant messages within the exclusive/inclusive interval and detects actual truncation', async () => {
+    const f = await fixture()
+    f.session.append('turn/start', { turn: 1 })
+    const first = f.session.append('assistant/message', {
+      turn: 1, step: 1,
+      message: createMessage({ role: 'assistant', content: [{ type: 'text', text: 'first' }], source: { kind: 'model', provider: 'p', model: 'm' } }),
+      stream: [],
+    }, { surfaceOp: 'append' })
+    const second = f.session.append('assistant/message', {
+      turn: 1, step: 2,
+      message: createMessage({ role: 'assistant', content: [{ type: 'text', text: 'second' }], source: { kind: 'model', provider: 'p', model: 'm' } }),
+      stream: [],
+    }, { surfaceOp: 'append' })
+    f.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    const observe = vi.spyOn(f.ctx.sessionQuery, 'observeSession')
+    const all = await f.history.assistantOutput(f.session.id, -1, 3, 2, f.abort.signal)
+    expect(all.records.map(record => record.event.seq)).toEqual([second.seq, first.seq])
+    expect(all.truncated).toBe(false)
+    const limited = await f.history.assistantOutput(f.session.id, -1, 3, 1, f.abort.signal)
+    expect(limited.records.map(record => record.event.seq)).toEqual([second.seq])
+    expect(limited.truncated).toBe(true)
+    const interval = await f.history.assistantOutput(f.session.id, first.seq, second.seq, 2, f.abort.signal)
+    expect(interval.records.map(record => record.event.seq)).toEqual([second.seq])
+    expect(observe).toHaveBeenCalledWith(f.session.id, { signal: f.abort.signal, projectionMode: 'none' })
+    await expect(f.history.assistantOutput(f.session.id, -1, 9, 2, f.abort.signal)).rejects.toMatchObject({ code: 'gateway/internal' })
+  })
+
+  it('reads at the observed log end when throughSeq is omitted', async () => {
+    const f = await fixture()
+    f.session.append('turn/start', { turn: 1 })
+    f.session.append('assistant/message', {
+      turn: 1, step: 1,
+      message: createMessage({ role: 'assistant', content: [{ type: 'text', text: 'latest' }], source: { kind: 'model', provider: 'p', model: 'm' } }),
+      stream: [],
+    }, { surfaceOp: 'append' })
+    const page = await f.history.history({ address: f.address }, f.abort.signal)
+    expect(page.cursor).toBe(f.session.seq - 1)
+    expect(page.records.at(-1)?.event.seq).toBe(f.session.seq - 1)
+  })
+
+  it('keeps the append copy a model-facing replacement shadowed', async () => {
+    const f = await fixture()
+    f.session.append('turn/start', { turn: 1 })
+    const replaced = f.session.append('assistant/message', {
+      turn: 1, step: 1,
+      message: createMessage({ role: 'assistant', content: [{ type: 'text', text: 'visible' }], source: { kind: 'model', provider: 'p', model: 'm' } }),
+      stream: [],
+    }, { surfaceOp: 'append' })
+    // Compaction rewrites the model-facing surface with a replacement that shadows the original;
+    // the human-readable transcript keeps the shadowed append copy.
+    const replacement = f.session.append('user/message', createMessage({
+      role: 'user', content: [{ type: 'text', text: 'compacted' }], source: { kind: 'user' },
+    }), { surfaceOp: { op: 'replace', startSeq: replaced.seq, endSeq: replaced.seq }, sourceEventSeqs: [replaced.seq] })
+    const output = await f.history.assistantOutput(f.session.id, -1, replacement.seq, 5, f.abort.signal)
+    expect(output.records.map(record => record.event.seq)).toEqual([replaced.seq])
+  })
+
+  it('releases the opening Observation when activation has nothing to promote', async () => {
+    const f = await fixture()
+    const source = await f.ctx.sessionQuery.observeSession(f.session.id)
+    const dispose = vi.fn(() => { source[Symbol.dispose]() })
+    vi.spyOn(f.ctx.sessionQuery, 'observeSession').mockResolvedValue({ ...source, [Symbol.dispose]: dispose })
+    const stream = f.web.follow({ address: f.address, activate: true }, f.abort.signal)[Symbol.asyncIterator]()
+    await stream.next()
+    f.abort.abort()
+    await stream.next()
+    expect(dispose).toHaveBeenCalled()
+  })
+
+  it('releases an acquired Observation when caller cancellation wins its return', async () => {
+    const f = await fixture()
+    const source = await f.ctx.sessionQuery.observeSession(f.session.id)
+    const dispose = vi.fn(() => { source[Symbol.dispose]() })
+    vi.spyOn(f.ctx.sessionQuery, 'observeSession').mockImplementation(async () => {
+      f.abort.abort()
+      return { ...source, [Symbol.dispose]: dispose }
+    })
+    await expect(f.history.usingObservation({ sessionId: f.session.id, projection: false }, f.abort.signal)).rejects.toBeDefined()
+    expect(dispose).toHaveBeenCalledOnce()
+  })
+
+  it('maps a non-Error query cancellation to a cancelled Remote error', async () => {
+    const f = await fixture()
+    // A cross-realm or opaque source may reject with the code alone; the message falls back.
+    vi.spyOn(f.ctx.sessionQuery, 'observeSession').mockRejectedValue({ code: 'SESSION_QUERY_ABORTED' })
+    await expect(f.history.history({ address: f.address }, f.abort.signal)).rejects.toMatchObject({
+      code: 'gateway/cancelled', message: 'Session read was cancelled',
+    })
+  })
+
+  it('maps missing subagent Follow reads through the Address error', async () => {
+    const f = await fixture()
+    const stream = f.web.follow({ address: {
+      kind: 'subagent', parentSessionId: f.session.id, childSessionId: SessionId('missing-child'), mode: 'continuable',
+    } }, f.abort.signal)[Symbol.asyncIterator]()
+    await expect(stream.next()).rejects.toMatchObject({ code: 'subagent/not-found' })
+  })
+
+  it('does not promote a live Session when activation was requested', async () => {
+    const f = await fixture()
+    const promote = vi.fn()
+    const web = new WebController(f.ctx, f.history, promote)
+    const stream = web.follow({ address: f.address, activate: true }, f.abort.signal)[Symbol.asyncIterator]()
+    expect(await stream.next()).toMatchObject({ value: { type: 'snapshot' } })
+    const pending = stream.next()
+    f.abort.abort()
+    expect(await pending).toMatchObject({ done: true })
+    expect(promote).not.toHaveBeenCalled()
+  })
+})

packages/api/session-controller/tests/listeners.host.spec.ts

@@ -0,0 +1,263 @@
+import { Context } from '@deepseek-ai/cordis'
+import type { Agent, AssistantStreamFrame } from '@deepseek-ai/dsh-agent'
+import { LlmAttemptId } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SessionId, SessionSeq } from '@deepseek-ai/dsh-session'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { EventListener, ListenerBuffer, ProjectionListener, eventItem } from '../src/history/listener.ts'
+import { SessionAssistantStreamsListener } from '../src/history/assistant-stream.ts'
+import type { SessionFollowFrame, SessionEventEntry } from '../src/types.ts'
+import { installSessionReadTestServices } from './test-remote.ts'
+
+const roots: Context[] = []
+const aborts: AbortController[] = []
+function cancellation() {
+  const abort = new AbortController()
+  aborts.push(abort)
+  return abort
+}
+afterEach(async () => {
+  for (const abort of aborts.splice(0)) abort.abort()
+  await Promise.all(roots.splice(0).map(ctx => ctx.fiber.dispose()))
+  vi.restoreAllMocks()
+})
+async function fixture() {
+  const ctx = new Context()
+  roots.push(ctx)
+  await ctx.plugin(SessionStore)
+  installSessionReadTestServices(ctx)
+  const session = ctx.sessions.create(SessionId('listener-target'), { meta: { cwd: '/listeners' } })
+  const abort = cancellation()
+  const partial: Partial<Agent> = { id: session.id, session }
+  const agent = partial as Agent
+  const emit = (frame: AssistantStreamFrame) => { ctx.emit('agent/assistant-stream', { agent, frame }) }
+  return { ctx, session, abort, emit, agent }
+}
+const start = (revision = 1): AssistantStreamFrame => ({ type: 'start', attemptId: LlmAttemptId('attempt'), revision, turn: 1, step: 1 })
+
+describe('ListenerBuffer', () => {
+  it('orders heterogeneous inputs lazily, supports prepend, and admits only one consumer', async () => {
+    const { ctx, abort } = await fixture()
+    using buffer = new ListenerBuffer<string>(ctx, abort.signal)
+    const convert = vi.fn((value: number) => [`n:${value}`])
+    const numbers = buffer.createInput(convert)
+    const words = buffer.createInput<string>(value => [value])
+    numbers.append(2)
+    words.prepend('first')
+    expect(convert).not.toHaveBeenCalled()
+    const values = buffer.items()[Symbol.asyncIterator]()
+    expect(await values.next()).toEqual({ done: false, value: 'first' })
+    expect(await values.next()).toEqual({ done: false, value: 'n:2' })
+    await expect(buffer.items()[Symbol.asyncIterator]().next()).rejects.toMatchObject({ code: 'gateway/internal' })
+    const waiting = values.next()
+    numbers.append(3)
+    expect(await waiting).toMatchObject({ value: 'n:3' })
+    await values.return?.()
+    const action = vi.fn()
+    buffer.capture(action)
+    numbers.append(4)
+    words.prepend('closed')
+    expect(action).not.toHaveBeenCalled()
+    buffer.own(action)
+    expect(action).toHaveBeenCalledOnce()
+    expect(() => buffer.createInput<string>(value => [value])).toThrow('closed')
+  })
+
+  it('propagates captured failure both before and during transformation', async () => {
+    const { ctx, abort } = await fixture()
+    using first = new ListenerBuffer<number>(ctx, abort.signal)
+    first.capture(() => { throw new Error('captured') })
+    await expect(first.items()[Symbol.asyncIterator]().next()).rejects.toThrow('captured')
+    using second = new ListenerBuffer<number>(ctx, abort.signal)
+    const input = second.createInput<number>(function* (value) {
+      second.capture(() => { throw new Error('transform failure') })
+      yield value
+    })
+    input.append(1)
+    await expect(second.items()[Symbol.asyncIterator]().next()).rejects.toThrow('transform failure')
+  })
+
+  it('does not emit after transformation closes its own buffer', async () => {
+    const { ctx, abort } = await fixture()
+    using buffer = new ListenerBuffer<number>(ctx, abort.signal)
+    buffer.createInput<number>(function* (value) { buffer[Symbol.dispose](); yield value }).append(1)
+    expect(await buffer.items()[Symbol.asyncIterator]().next()).toEqual({ done: true, value: undefined })
+  })
+
+  it('releases abort registration when effect installation fails', async () => {
+    const { ctx, abort } = await fixture()
+    const remove = vi.spyOn(abort.signal, 'removeEventListener')
+    vi.spyOn(ctx, 'effect').mockImplementation(() => { throw new Error('effect failure') })
+    expect(() => new ListenerBuffer(ctx, abort.signal)).toThrow('effect failure')
+    expect(remove).toHaveBeenCalledWith('abort', expect.any(Function))
+  })
+})
+
+describe('EventListener', () => {
+  it('starts at the first notification without acquiring a snapshot', async () => {
+    const { ctx, session, abort } = await fixture()
+    using listener = new EventListener(ctx, session.id, abort.signal)
+    const event = session.append('turn/start', { turn: 1 })
+    const values = listener.events()[Symbol.asyncIterator]()
+    expect(await values.next()).toMatchObject({ value: { type: 'event', event } })
+    abort.abort()
+    expect(await values.next()).toMatchObject({ done: true })
+  })
+
+  it('subscribes before an empty Session is created and follows its first later event', async () => {
+    const { ctx, abort } = await fixture()
+    const id = SessionId('created-after-bind')
+    using listener = new EventListener(ctx, id, abort.signal)
+    const session = ctx.sessions.create(id)
+    const values = listener.events()[Symbol.asyncIterator]()
+    const waiting = values.next()
+    const event = session.append('turn/start', { turn: 1 })
+    expect(await waiting).toMatchObject({ value: { event } })
+    await values.return?.()
+  })
+
+  it('filters buffered events through a supplied cursor', async () => {
+    const { ctx, session, abort } = await fixture()
+    using listener = new EventListener(ctx, session.id, abort.signal)
+    session.append('turn/start', { turn: 1 })
+    const event = session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    listener.setCursor(0)
+    const values = listener.events()[Symbol.asyncIterator]()
+    expect(await values.next()).toMatchObject({ value: { event } })
+    await values.return?.()
+  })
+
+  it('leaves an external buffer open and drops notifications after listener disposal', async () => {
+    const { ctx, session, abort } = await fixture()
+    using queue = new ListenerBuffer<SessionEventEntry>(ctx, abort.signal)
+    const listener = new EventListener(ctx, session.id, abort.signal, queue)
+    session.append('turn/start', { turn: 1 })
+    await expect(listener.events()[Symbol.asyncIterator]().next()).rejects.toThrow('externally owned')
+    listener[Symbol.dispose]()
+    const values = queue.items()[Symbol.asyncIterator]()
+    const next = values.next()
+    abort.abort()
+    expect(await next).toMatchObject({ done: true })
+  })
+
+  it('stops within a creation batch when its binding is disposed', async () => {
+    const { ctx, session, abort } = await fixture()
+    session.append('turn/start', { turn: 1 })
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    using queue = new ListenerBuffer<SessionEventEntry>(ctx, abort.signal)
+    const listener = new EventListener(ctx, session.id, abort.signal, queue)
+    listener.setCursor(-1)
+    ctx.emit('session/created', session)
+    const values = queue.items()[Symbol.asyncIterator]()
+    expect(await values.next()).toMatchObject({ value: { event: { seq: 0 } } })
+    listener[Symbol.dispose]()
+    const waiting = values.next()
+    abort.abort()
+    expect(await waiting).toMatchObject({ done: true })
+  })
+
+  it('cleans a partially installed registration when a later one fails', async () => {
+    const { ctx, session, abort } = await fixture()
+    const on = ctx.on.bind(ctx)
+    let count = 0
+    vi.spyOn(ctx, 'on').mockImplementation((...args) => {
+      if (++count === 2) throw new Error('registration failure')
+      return on(...args)
+    })
+    expect(() => new EventListener(ctx, session.id, abort.signal)).toThrow('registration failure')
+  })
+
+  it('handles raw events and rejects a noncontiguous supplied creation prefix', async () => {
+    const { session } = await fixture()
+    const event = session.append('turn/start', { turn: 1 })
+    expect(eventItem({ type: 'event', session, event })).toEqual({ events: [event], next: 1 })
+    expect(eventItem({ type: 'created', session, end: 0 })).toEqual({ events: [], next: undefined })
+    vi.spyOn(session, 'snapshotEvents').mockReturnValue([{ ...event, seq: SessionSeq(1) }])
+    expect(() => eventItem({ type: 'created', session, end: 2 }, 0)).toThrow('skipped seq')
+    expect(eventItem({ type: 'created', session, end: 2 })).toMatchObject({ next: 2 })
+  })
+})
+
+describe('ProjectionListener', () => {
+  it('emits sparse changes only for the selected Session', async () => {
+    const { ctx, session, abort } = await fixture()
+    let receive!: Parameters<typeof ctx.sessionProjections.onChanged>[0]
+    const dispose = vi.fn()
+    vi.spyOn(ctx.sessionProjections, 'onChanged').mockImplementation((listener) => { receive = listener; return dispose })
+    using listener = new ProjectionListener(ctx, session.id, abort.signal)
+    const other = ctx.sessions.create(SessionId('other'))
+    receive(other, 'subagent', null, SessionSeq(0))
+    receive(session, 'subagent', null, SessionSeq(1))
+    const changes = listener.changes()[Symbol.asyncIterator]()
+    expect(await changes.next()).toMatchObject({ value: { type: 'projection', key: 'subagent', value: null, seq: 1 } })
+    abort.abort()
+    expect(await changes.next()).toMatchObject({ done: true })
+    expect(dispose).toHaveBeenCalledOnce()
+  })
+
+  it('releases its buffer when projection subscription fails', async () => {
+    const { ctx, session, abort } = await fixture()
+    const remove = vi.spyOn(abort.signal, 'removeEventListener')
+    vi.spyOn(ctx.sessionProjections, 'onChanged').mockImplementation(() => { throw new Error('projection subscribe failed') })
+    expect(() => new ProjectionListener(ctx, session.id, abort.signal)).toThrow('projection subscribe failed')
+    expect(remove).toHaveBeenCalled()
+  })
+})
+
+describe('AssistantStreamListener', () => {
+  it('cuts baseline-covered arrivals and clears reconnect state on Agent disposal', async () => {
+    const { ctx, session, abort, emit, agent } = await fixture()
+    const streams = new SessionAssistantStreamsListener(ctx)
+    using queue = new ListenerBuffer<SessionFollowFrame>(ctx, abort.signal)
+    using listener = streams.bind(session.id, abort.signal, queue)
+    expect(listener.snapshot()).toEqual({ baseline: { revision: 0 }, ordinal: 0 })
+    emit(start())
+    expect(listener.snapshot()).toMatchObject({ baseline: { revision: 1 }, ordinal: 1 })
+    listener.setOrdinalCut(1)
+    const frames = queue.items()[Symbol.asyncIterator]()
+    const end: AssistantStreamFrame = { type: 'end', attemptId: LlmAttemptId('attempt'), revision: 2, index: 0, outcome: { kind: 'abandoned' } }
+    emit(end)
+    expect(await frames.next()).toMatchObject({ value: { type: 'assistant-stream', frame: end } })
+    ctx.emit('agent/disposed', { agent })
+    expect(listener.snapshot()).toEqual({ baseline: { revision: 0 }, ordinal: 2 })
+    abort.abort()
+    expect(await frames.next()).toMatchObject({ done: true })
+  })
+
+  it('records the durable cursor for an attempt started after a committed event', async () => {
+    const { ctx, session, abort, emit } = await fixture()
+    session.append('turn/start', { turn: 1 })
+    const streams = new SessionAssistantStreamsListener(ctx)
+    using queue = new ListenerBuffer<SessionFollowFrame>(ctx, abort.signal)
+    using listener = streams.bind(session.id, abort.signal, queue)
+    emit(start())
+    listener.setOrdinalCut(0)
+    const frames = queue.items()[Symbol.asyncIterator]()
+    expect(await frames.next()).toMatchObject({ value: { frame: { type: 'start', startedAfterSeq: 0 } } })
+    await frames.return?.()
+  })
+
+  it('leaves queue ownership intact and drops notifications after unbinding', async () => {
+    const { ctx, session, abort, emit } = await fixture()
+    const streams = new SessionAssistantStreamsListener(ctx)
+    using queue = new ListenerBuffer<SessionFollowFrame>(ctx, abort.signal)
+    const listener = streams.bind(session.id, abort.signal, queue)
+    emit(start())
+    listener[Symbol.dispose]()
+    const next = queue.items()[Symbol.asyncIterator]().next()
+    abort.abort()
+    expect(await next).toMatchObject({ done: true })
+  })
+
+  it('propagates registration failure without closing the shared queue', async () => {
+    const { ctx, session, abort } = await fixture()
+    const streams = new SessionAssistantStreamsListener(ctx)
+    using queue = new ListenerBuffer<SessionFollowFrame>(ctx, abort.signal)
+    vi.spyOn(ctx, 'on').mockImplementation(() => { throw new Error('assistant subscribe failed') })
+    expect(() => streams.bind(session.id, abort.signal, queue)).toThrow('assistant subscribe failed')
+    // The failed binding leaves the shared queue usable for the rest of the connection.
+    const input = queue.createInput<SessionFollowFrame>(value => [value])
+    const event = session.append('turn/start', { turn: 1 })
+    input.append({ type: 'event', event })
+    await expect(queue.items()[Symbol.asyncIterator]().next()).resolves.toMatchObject({ value: { type: 'event', event } })
+  })
+})

packages/api/session-controller/tests/manager.client.spec.ts

@@ -862,7 +862,7 @@ describe('subagent catalogs', () => {
           childSessionId: S2,
           mode: 'continuable',
         },
-        assistantStream: true,
+        activate: true, assistantStream: true,
         maxMessages: 500,
         turnWindow: { minMessages: 50, minTurns: 2 },
       },
@@ -958,7 +958,7 @@ describe('subagent catalogs', () => {
     if (source === 'control') {
       manager.handleControlFrame({ type: 'baseline', value: { projections: { [S1]: baseline } } })
     } else {
-      mock.stream(FOLLOW, followScript(ok({ records: [], hasMore: false, projections: baseline })))
+      mock.stream(FOLLOW, followScript(ok({ cursor: baseline.asOfSeq, records: [], hasMore: false, projections: baseline })))
       await session.open()
     }
 
@@ -983,7 +983,7 @@ describe('subagent catalogs', () => {
       if (source === 'control') {
         manager.handleControlFrame({ type: 'baseline', value: { projections: { [S1]: baseline } } })
       } else {
-        mock.stream(FOLLOW, followScript(ok({ records: [], hasMore: false, projections: baseline })))
+        mock.stream(FOLLOW, followScript(ok({ cursor: baseline.asOfSeq, records: [], hasMore: false, projections: baseline })))
         await manager.get(S1).open()
       }
       response.resolve(ok({ kind: 'migration-required', values: {
@@ -1545,6 +1545,7 @@ describe('connected generation', () => {
 
   it('refreshes query baselines without rebuilding independently resumed Session sources', async ({ mock, remote, start }) => {
     mock.stream(FOLLOW, followScript(ok({
+      cursor: 5,
       records: entries(plainTurn(SessionSeq(0), 0, 'a', 'b')) as never[],
       hasMore: false,
       modelSelection: { provider: 'deepseek-official', model: 'deepseek-chat' },

packages/api/session-controller/tests/projection-discovery.host.spec.ts

@@ -239,7 +239,7 @@ describe('Session projection discovery', () => {
       kind: 'sequenced', asOfSeq: 0,
       values: { subagentCatalog: [{ id: 'current-child', mode: 'continuable', label: 'worker' }] },
     })
-    expect(observe).toHaveBeenCalledWith(parent.header.id, { signal })
+    expect(observe).toHaveBeenCalledWith(parent.header.id, { signal, projectionMode: 'all' })
     expect(open).toHaveBeenCalledWith(parent.header.id, 'read', { signal })
     expect(fixture.ctx.sessions.get(parent.header.id)).toBeUndefined()
     expect(fixture.ctx.agents.get(parent.header.id)).toBeUndefined()
@@ -310,7 +310,7 @@ describe('Session projection discovery', () => {
       .rejects.toMatchObject({ code: 'session/projections-unavailable' })
   })
 
-  it.each([false, true])('rechecks live history when the stored source disappears after preflight (attached: %s)', async (attached) => {
+  it.each([false, true])('returns the observation not-found result without a second live lookup (attached: %s)', async (attached) => {
     const fixture = await harness()
     const parent = await stored(fixture, SessionId('missing-after-preflight'))
     const snapshot = await fixture.ctx.sessionPersistence.stat(parent.header.id)
@@ -322,9 +322,8 @@ describe('Session projection discovery', () => {
       })
 
     const result = await fixture.controller.projections({ sessionId: parent.header.id }, new AbortController().signal)
-    expect(result).toEqual(attached
-      ? { kind: 'sequenced', ...fixture.ctx.sessionProjections.snapshot(fixture.ctx.sessions.get(parent.header.id)!) }
-      : null)
+    expect(result).toBeNull()
+    expect(fixture.ctx.sessions.get(parent.header.id) !== undefined).toBe(attached)
   })
 
   it('serves the live registry cut instead of a conflicting cached hint', async () => {
@@ -342,7 +341,7 @@ describe('Session projection discovery', () => {
     expect(cached).not.toHaveBeenCalled()
   })
 
-  it.each([true, false])('prefers a Session attached during stat (stored header present: %s)', async (present) => {
+  it.each([true, false])('uses migration preflight before observing a Session attached during stat (stored header present: %s)', async (present) => {
     const fixture = await harness()
     const parent = await stored(fixture, SessionId('stat-race'), {}, 3)
     const snapshot = await fixture.ctx.sessionPersistence.stat(parent.header.id)
@@ -358,7 +357,9 @@ describe('Session projection discovery', () => {
       const live = fixture.ctx.sessions.create(parent.header.id, { meta: parent.header })
       live.append('turn/start', { turn: 1 })
       finish.resolve(present ? snapshot : undefined)
-      await expect(operation).resolves.toEqual({ kind: 'sequenced', ...fixture.ctx.sessionProjections.snapshot(live) })
+      await expect(operation).resolves.toEqual(present
+        ? { kind: 'migration-required', values: {} }
+        : { kind: 'sequenced', ...fixture.ctx.sessionProjections.snapshot(live) })
     } finally {
       finish.resolve(undefined)
       await Promise.allSettled([operation])

packages/api/session-controller/tests/projection-store.client.spec.ts

@@ -214,7 +214,7 @@ describe('Session tail-page seeding', () => {
 
   it('treats a blockless response as no reset: pushed values survive', async ({ mock, start }) => {
     const session = await sessionBench(mock, start, SID)
-    mock.stream(FOLLOW, followScript(ok({ records: entries(plainTurn(SessionSeq(0), 0, 'a', 'b')) as never[], hasMore: false })))
+    mock.stream(FOLLOW, followScript(ok({ cursor: 5, records: entries(plainTurn(SessionSeq(0), 0, 'a', 'b')) as never[], hasMore: false })))
     await session.open()
     session.projections.apply('test/marks', { marks: ['pushed'] }, SessionSeq(9))
     await session.resync()

packages/api/session-controller/tests/reference-ownership.client.spec.ts

@@ -21,7 +21,7 @@ declare module '@deepseek-ai/dsh-api-session-controller/client' {
 const viewSource: SessionReferenceSource = 'referenceTestView'
 const workSource: SessionReferenceSource = 'referenceTestWork'
 const ID = SessionId('reference-session')
-const EMPTY_HISTORY = ok({ records: [], hasMore: false })
+const EMPTY_HISTORY = ok({ cursor: -1, records: [], hasMore: false })
 const it = createClientTest({ roster: webApp.closure(['@deepseek-ai/dsh-api-gateway']) })
 
 async function bench(mock: RemoteMock, start: () => Promise<TestClient>, listed = true) {

packages/api/session-controller/tests/remote/history.client.ts

@@ -1,7 +1,7 @@
 /** Pure history response builders for assembled Session tests. */
 import { SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session/types'
 import type {
-  SessionAssistantStreamBaseline, SessionFollowFrame, SessionFollowRequest, SessionPage, SessionProjectionBaseline,
+  SessionAssistantStreamBaseline, SessionFollowFrame, SessionFollowRequest, SessionPage, SessionEventsPage, SessionProjectionBaseline,
 } from '../../src/types.ts'
 import { historyRecordLastSeq } from '../../src/client/sessions/history-records.ts'
 
@@ -11,8 +11,8 @@ import { historyRecordLastSeq } from '../../src/client/sessions/history-records.
  * @param throughSeq - inclusive final sequence.
  * @returns the page without records beyond the cursor.
  */
-export function pageThrough(page: SessionPage, throughSeq: number): SessionPage {
-  return { ...page, records: page.records.filter(record => historyRecordLastSeq(record) <= throughSeq) }
+export function pageThrough(page: SessionPage, throughSeq: number): SessionEventsPage {
+  return { ...page, cursor: throughSeq, records: page.records.filter(record => historyRecordLastSeq(record) <= throughSeq) }
 }
 
 /**

packages/api/session-controller/tests/remote/session.client.ts

@@ -12,7 +12,7 @@ import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import type { RemoteFailure, RemoteResult } from '@deepseek-ai/dsh-typert-protocol'
 import type {
   SessionAssistantStreamBaseline, SessionFollowFrame, SessionFollowRequest,
-  SessionPage, SessionPageRequest,
+  SessionEventsPage, SessionPageRequest,
 } from '../../src/types.ts'
 import { entries, historyValue } from '../event-script.client.ts'
 import { followSnapshot, pageThrough } from './history.client.ts'
@@ -24,7 +24,7 @@ export const FOLLOW = 'session/follow'
 export const PAGE = 'session/page'
 
 /** A history answer as the Host returns it, possibly still pending. */
-export type HistoryAnswer = RemoteResult<SessionPage> | Promise<RemoteResult<SessionPage>>
+export type HistoryAnswer = RemoteResult<SessionEventsPage> | Promise<RemoteResult<SessionEventsPage>>
 /** A history answer, or a function of the request that produces one. */
 export type HistorySource<Request> = HistoryAnswer | ((request: Request) => HistoryAnswer)
 
@@ -43,7 +43,7 @@ export function err<T>(error: RemoteFailure): RemoteResult<T> {
  * @param hasMore - whether older history exists.
  * @returns the success result.
  */
-export function history(events: readonly SessionEvent[], hasMore = false): RemoteResult<SessionPage> {
+export function history(events: readonly SessionEvent[], hasMore = false): RemoteResult<SessionEventsPage> {
   return ok(historyValue(events, hasMore))
 }
 
@@ -102,11 +102,11 @@ export function followScript(
  */
 export function pageRule(
   history: HistorySource<SessionPageRequest>,
-): UnaryRuleFn<readonly [SessionPageRequest], Promise<RemoteResult<SessionPage>>> {
+): UnaryRuleFn<readonly [SessionPageRequest], Promise<RemoteResult<SessionEventsPage>>> {
   return async (page) => {
     const result = await answer(history, page)
     if (!result.ok) return result
-    return ok(pageThrough(result.value, page.throughSeq))
+    return ok(pageThrough(result.value, page.throughSeq ?? result.value.cursor))
   }
 }
 
@@ -144,12 +144,12 @@ export const sessionWorld: RemoteTable = {
       attachment: { attachmentId: 'a', mediaType: 'image/png', bytes: 1, width: 1, height: 1 },
       data: 'AA==',
     }),
-    'session/page': pageRule(ok({ records: [], hasMore: false })),
+    'session/page': pageRule(ok({ cursor: -1, records: [], hasMore: false })),
     'session/projections': ok({ kind: 'sequenced', asOfSeq: -1, values: { subagentCatalog: [] } }),
     'subagents/prompt': ok({ messageId: 'fake-message' }),
     'subagents/interruptByParent': ok({ accepted: true }),
   },
   stream: {
-    'session/follow': followScript(ok({ records: [], hasMore: false })),
+    'session/follow': followScript(ok({ cursor: -1, records: [], hasMore: false })),
   },
 }

packages/api/session-controller/tests/session-cold.host.spec.ts

@@ -11,7 +11,6 @@ import { describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import SessionStore from '@deepseek-ai/dsh-session'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
-import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
 import { subagentIdentityProjectionDefinition } from '@deepseek-ai/dsh-subagent/src/projection.ts'
 import TypertRegistry from '@deepseek-ai/dsh-typert-registry'
 import { createUserMessage, MessageId } from '@deepseek-ai/dsh-llm'
@@ -574,10 +573,7 @@ describe('subagent ownership fence', () => {
     const remote = await createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
     await ctx.sessionProjections.register(subagentIdentityProjectionDefinition)
 
-    const history = await new SessionHistoryController(
-      ctx,
-      (observation) => { observation[Symbol.dispose]() },
-    ).page({
+    const history = await ctx.sessionController.events({
       address: {
         kind: 'subagent',
         parentSessionId: meta.parentSession as SessionId,

packages/api/session-controller/tests/session-history-journal.host.spec.ts

@@ -8,7 +8,8 @@ import SessionStore from '@deepseek-ai/dsh-session'
 import { LlmAttemptId, ToolCallId, createMessage, createToolResultMessage, createUserMessage } from '@deepseek-ai/dsh-llm'
 import type { MessageSource } from '@deepseek-ai/dsh-llm'
 import type { Session, SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
-import { SessionHistoryController } from '@deepseek-ai/dsh-api-session-controller/src/history.ts'
+import { HistoryController } from '../src/history/controller.ts'
+import { WebController } from '../src/web/controller.ts'
 import type { SessionFollowFrame, SessionPage, SessionWireEvent } from '@deepseek-ai/dsh-api-session-controller/types'
 import { createSessionTestRemote, installSessionReadTestServices } from './test-remote.ts'
 
@@ -57,6 +58,12 @@ async function harness(): Promise<{ ctx: Context }> {
   return { ctx }
 }
 
+/** Initialize the independent snapshot owner and Web aggregation for direct journal tests. */
+async function controllers(ctx: Context): Promise<{ history: HistoryController; web: WebController }> {
+  const history = new HistoryController(ctx)
+  return { history, web: new WebController(ctx, history, (observation) => { observation[Symbol.dispose]() }) }
+}
+
 /** Drain one Session follow until `count` event frames arrive. */
 async function collect(
   iterable: AsyncIterable<SessionFollowFrame>,
@@ -73,11 +80,11 @@ async function collect(
 
 /** Open follow and wait until its cursor is fixed before appending fixtures. */
 async function openFollow(
-  history: SessionHistoryController,
+  web: WebController,
   sessionId: SessionId,
   signal: AbortSignal,
 ): Promise<AsyncIterable<SessionFollowFrame>> {
-  const iterator = history.follow({
+  const iterator = web.follow({
     address: { kind: 'session', sessionId },
   }, signal)[Symbol.asyncIterator]()
   await expect(iterator.next()).resolves.toMatchObject({
@@ -107,9 +114,9 @@ describe('Session history raw journal', () => {
   it('opens an empty opted-in Assistant baseline before any live attempt exists', async () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -127,7 +134,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const originalObserve = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery)
     const entered = Promise.withResolvers<undefined>()
     const release = Promise.withResolvers<undefined>()
@@ -137,7 +144,7 @@ describe('Session history raw journal', () => {
       return originalObserve(...args)
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -182,7 +189,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const attemptId = LlmAttemptId('live-follow-attempt')
     const emit = (frame: AssistantStreamFrame): void => {
       ctx.emit('agent/assistant-stream', { agent, frame })
@@ -197,7 +204,7 @@ describe('Session history raw journal', () => {
       time: 1, chunk: firstChunk,
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -249,7 +256,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const attemptId = LlmAttemptId(`${session.id}:1`)
     ctx.emit('agent/assistant-stream', {
       agent,
@@ -267,7 +274,7 @@ describe('Session history raw journal', () => {
       },
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -311,7 +318,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const attemptId = LlmAttemptId('revision-gap-attempt')
     ctx.emit('agent/assistant-stream', {
       agent,
@@ -329,7 +336,7 @@ describe('Session history raw journal', () => {
       },
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -351,7 +358,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const attemptId = LlmAttemptId('dense-index-attempt')
     ctx.emit('agent/assistant-stream', {
       agent,
@@ -369,7 +376,7 @@ describe('Session history raw journal', () => {
       },
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -391,7 +398,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const attemptId = LlmAttemptId('cached-baseline-attempt')
     ctx.emit('agent/assistant-stream', {
       agent,
@@ -402,12 +409,12 @@ describe('Session history raw journal', () => {
     })
 
     const firstAbort = new AbortController()
-    const firstIterator = history.follow({
+    const firstIterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, firstAbort.signal)[Symbol.asyncIterator]()
     const secondAbort = new AbortController()
-    const secondIterator = history.follow({
+    const secondIterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, secondAbort.signal)[Symbol.asyncIterator]()
@@ -431,9 +438,9 @@ describe('Session history raw journal', () => {
   it('opens an empty Assistant baseline before the target Agent emits frames', async () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -456,9 +463,9 @@ describe('Session history raw journal', () => {
     const target = ctx.sessions.create(undefined, { meta: { cwd: '/target' } })
     const other = ctx.sessions.create(undefined, { meta: { cwd: '/other' } })
     const otherAgent = { id: other.id, session: other, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: target.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -489,7 +496,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const observationStarted = Promise.withResolvers<undefined>()
     const releaseObservation = Promise.withResolvers<undefined>()
     const originalObserve = ctx.sessionQuery.observeSession.bind(ctx.sessionQuery)
@@ -499,7 +506,7 @@ describe('Session history raw journal', () => {
       return await originalObserve(sessionId, options)
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -539,7 +546,7 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const attemptId = LlmAttemptId(`${session.id}:1`)
     ctx.emit('agent/assistant-stream', {
       agent,
@@ -557,7 +564,7 @@ describe('Session history raw journal', () => {
       return await originalObserve(sessionId, options)
     })
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
       assistantStream: true,
     }, abort.signal)[Symbol.asyncIterator]()
@@ -613,9 +620,9 @@ describe('Session history raw journal', () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
     const agent = { id: session.id, session, status: 'running', ctx } as Agent
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const iterator = history.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
     }, abort.signal)[Symbol.asyncIterator]()
     const opening = await iterator.next()
@@ -655,9 +662,9 @@ describe('Session history raw journal', () => {
   it('follows raw tool events and preserves result metadata without a Tools service', async () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const stream = await openFollow(history, session.id, abort.signal)
+    const stream = await openFollow(web, session.id, abort.signal)
     const collected = collect(stream, 2, abort)
     const call = session.append('tool/call', {
       turn: 1, step: 1, callId: ToolCallId('raw-call'), name: 'custom', arguments: '{malformed',
@@ -684,9 +691,9 @@ describe('Session history raw journal', () => {
   it('follows live results without rescanning Session history', async () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const stream = await openFollow(history, session.id, abort.signal)
+    const stream = await openFollow(web, session.id, abort.signal)
     const iterator = stream[Symbol.asyncIterator]()
 
     session.append('tool/call', {
@@ -957,9 +964,9 @@ describe('Session history raw journal', () => {
   it('follows a result after turn/end without reading the addressed Session log', async () => {
     const { ctx } = await harness()
     const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
-    const history = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
+    const { web } = await controllers(ctx)
     const abort = new AbortController()
-    const stream = await openFollow(history, session.id, abort.signal)
+    const stream = await openFollow(web, session.id, abort.signal)
     const iterator = stream[Symbol.asyncIterator]()
 
     session.append('turn/start', { turn: 1 })

packages/api/session-controller/tests/session.client.spec.ts

@@ -15,7 +15,7 @@ import { ok, type RemoteMock } from '@deepseek-ai/dsh-remote-mock'
 import { createClientTest, type TestClient, webApp } from '@deepseek-ai/dsh-client-test-runtime/src/assembly/index.ts'
 import { Session } from '../src/client/sessions/session.ts'
 import { SessionEventStream } from '../src/client/transport.ts'
-import type { SessionFollowRequest, SessionPage, SessionPageRequest } from '../src/types.ts'
+import type { SessionFollowRequest, SessionEventsPage, SessionPageRequest } from '../src/types.ts'
 import { entries, ev, historyValue, plainTurn } from './event-script.client.ts'
 import { sessionBench } from './remote/bench.client.ts'
 import {
@@ -122,7 +122,7 @@ describe('Session open', () => {
 
   it('stitches live frames landing right behind the opening snapshot, dropping the page overlap', async ({ mock, start }) => {
     const session = await sessionBench(mock, start, SID)
-    const gate = Promise.withResolvers<SessionPage>()
+    const gate = Promise.withResolvers<SessionEventsPage>()
     const page = plainTurn(SessionSeq(10), 0, '早', '安')
     mock.stream(FOLLOW, async ([request], stream) => {
       stream.push(followSnapshot(await gate.promise, request as SessionFollowRequest))
@@ -243,7 +243,7 @@ describe('paging', () => {
     mock.stream(FOLLOW, followScript(history(newest, true)))
     await session.open()
 
-    const gate = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const gate = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule(request => (request.beforeSeq === 12 ? history(middle, true) : history(oldest))))
     mock.remote.session.page.mockReturnValueOnce(gate.promise)
     const jump = session.loadThrough(SessionSeq(0))
@@ -279,7 +279,7 @@ describe('paging', () => {
     mock.stream(FOLLOW, followScript(history(newest, true)))
     await session.open()
     const secondRequested = Promise.withResolvers<undefined>()
-    const lastPage = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const lastPage = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule((request) => {
       if (request.beforeSeq === 12) return history(middle, true)
       secondRequested.resolve(undefined)
@@ -334,7 +334,7 @@ describe('paging', () => {
     mock.stream(FOLLOW, followScript(history(newest, true)))
     await session.open()
     const secondRequested = Promise.withResolvers<undefined>()
-    const lastPage = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const lastPage = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule((request) => {
       if (request.beforeSeq === 12) return history(plainTurn(SessionSeq(6), 1, 'middle', 'answer'), true)
       secondRequested.resolve(undefined)
@@ -360,7 +360,7 @@ describe('paging', () => {
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(12), 2, 'e', 'f'), true)))
     await session.open()
 
-    const gate = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const gate = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule(history(oldest)))
     mock.remote.session.page.mockReturnValueOnce(gate.promise)
     const first = session.loadThrough(SessionSeq(6))
@@ -381,7 +381,7 @@ describe('paging', () => {
     await session.open()
 
     // A plain single-page pull holds the busy flag while the jump is refused.
-    const gate = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const gate = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockReturnValueOnce(gate.promise)
     const older = session.loadOlder()
     await session.loadThrough(SessionSeq(0)) // refused: must not park seq 0 anywhere
@@ -409,7 +409,7 @@ describe('paging', () => {
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(12), 2, 'x', 'y'), true)))
     await session.open()
 
-    const gate = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const gate = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockReturnValueOnce(gate.promise)
     const jump = session.loadThrough(SessionSeq(0))
     // The address is rebuilt while the first page is in flight; the same follow script serves the new generation.
@@ -459,7 +459,7 @@ describe('paging', () => {
     const session = await sessionBench(mock, start, SID)
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(6), 1, 'x', 'y'), true)))
     await session.open()
-    const gate = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const gate = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule(gate.promise))
     const first = session.loadOlder()
     const second = session.loadOlder()
@@ -484,7 +484,7 @@ describe('prompt and cancel errors', () => {
     expect(steered).toEqual({ ok: true, value: { accepted: true } })
     expect(cancelled).toEqual({ ok: true, value: { accepted: true } })
     expect(mock.log.requests(FOLLOW)).toEqual([
-      { address: { kind: 'subagent', ...CHILD }, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
+      { address: { kind: 'subagent', ...CHILD }, activate: true, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
     ])
     expect(mock.log.requests(PAGE)).toEqual([])
     // The prompt mode crosses the wire as the request's delivery.
@@ -593,7 +593,7 @@ describe('prompt and cancel errors', () => {
     expect(mock.log.requests('subagents/prompt')).toMatchObject([CHILD])
     expect(mock.log.calls('subagents/interruptByParent').map(call => call.args)).toEqual([[SID, PARENT, 'continuable']])
     expect(mock.log.requests(FOLLOW)).toEqual([
-      { address: { kind: 'subagent', ...CHILD, mode: 'one-shot' }, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
+      { address: { kind: 'subagent', ...CHILD, mode: 'one-shot' }, activate: true, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
     ])
     expect(mock.log.requests(PAGE)).toEqual([])
     expect(mock.log.requests('session/cancel')).toEqual([])
@@ -878,7 +878,7 @@ describe('remaining branches', () => {
     const session = await sessionBench(mock, start, SID)
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(0), 0, 'a', 'b'))))
     await session.open()
-    const gate = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const gate = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule(gate.promise))
     await pushEvent(mock, ev.user(SessionSeq(9), '洞一'))
     await pushEvent(mock, ev.user(SessionSeq(10), '洞二'))
@@ -891,7 +891,7 @@ describe('remaining branches', () => {
 
   it('drops a stale opening without replacing the new generation error state', async ({ mock, start }) => {
     const session = await sessionBench(mock, start, SID)
-    const stale = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const stale = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.stream(FOLLOW, followScript(() => stale.promise))
     const opening = session.open()
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(0), 0, 'a', 'b'))))
@@ -903,7 +903,7 @@ describe('remaining branches', () => {
 
   it('drops a stale doOpen whose history resolved successfully after resync superseded it', async ({ mock, start }) => {
     const session = await sessionBench(mock, start, SID)
-    const stale = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const stale = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.stream(FOLLOW, followScript(() => stale.promise))
     const opening = session.open()
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(6), 1, '新', '代'))))
@@ -917,7 +917,7 @@ describe('remaining branches', () => {
     const session = await sessionBench(mock, start, SID)
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(0), 0, 'a', 'b'))))
     await session.open()
-    const repairPull = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const repairPull = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.remote.session.page.mockImplementation(pageRule(repairPull.promise))
     await pushEvent(mock, ev.user(SessionSeq(9), '洞'))
     await vi.waitFor(() => { expect(mock.log.requests(PAGE)).toHaveLength(1) })
@@ -959,7 +959,7 @@ describe('resync', () => {
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(0), 0, '旧', '窗'))))
     await session.open()
     const oldWindow = session.eventSource.getSnapshot()
-    const replacement = Promise.withResolvers<SessionPage>()
+    const replacement = Promise.withResolvers<SessionEventsPage>()
     // The reconnect generation: its snapshot is gated, then two live frames land out of order right behind it.
     mock.stream(FOLLOW, async ([request], stream) => {
       stream.push(followSnapshot(await replacement.promise, request as SessionFollowRequest))
@@ -1018,7 +1018,7 @@ describe('resync', () => {
 
   it('drops a stale in-flight open superseded by resync (generation guard)', async ({ mock, start }) => {
     const session = await sessionBench(mock, start, SID)
-    const stale = Promise.withResolvers<RemoteResult<SessionPage>>()
+    const stale = Promise.withResolvers<RemoteResult<SessionEventsPage>>()
     mock.stream(FOLLOW, followScript(() => stale.promise))
     const firstOpen = session.open()
     mock.stream(FOLLOW, followScript(history(plainTurn(SessionSeq(6), 1, '新', '代'))))

packages/api/session-controller/tests/sessions-service.client.spec.ts

@@ -12,7 +12,7 @@ import { createClientTest, webApp } from '@deepseek-ai/dsh-client-test-runtime/s
 import { ClientSessions, SessionCreateError, SessionForkError } from '../src/client/sessions/service.ts'
 import { scopeOf } from '../src/client/scope.ts'
 import type {
-  SessionAssistantStreamBaseline, SessionFollowFrame, SessionFollowRequest,
+  SessionAssistantStreamBaseline, SessionFollowFrame, SessionFollowRequest, SessionHistoryRecord,
 } from '../src/types.ts'
 import { FOLLOW, err, followScript, sessionWorld } from './remote/session.client.ts'
 
@@ -174,7 +174,7 @@ describe('scope tree', () => {
   it('opens a conversation from follow projections without a second projection request', async ({ bench }) => {
     const b = bench()
     b.mock.stream(FOLLOW, followScript(ok({
-      records: [], hasMore: false,
+      cursor: 0, records: [], hasMore: false,
       projections: { asOfSeq: 0, values: { subagentCatalog: [{
         id: sid('child'), createdAt: 1, mode: 'one-shot',
       }] } },
@@ -277,7 +277,7 @@ describe('scope tree', () => {
   it('replaces an active assistant baseline on reconnect without duplicate chunks', async ({ bench }) => {
     const b = bench()
     const attemptId = LlmAttemptId('reconnect-attempt')
-    let records: never[] = []
+    let records: SessionHistoryRecord[] = []
     let assistantStreamBaseline: SessionAssistantStreamBaseline = {
       revision: 2,
       activeAttempt: {
@@ -287,7 +287,7 @@ describe('scope tree', () => {
       },
     }
     b.mock.stream(FOLLOW, followScript(
-      () => ok({ records, hasMore: false }),
+      () => ok({ cursor: records.at(-1)?.event.seq ?? -1, records, hasMore: false }),
       { assistantStream: () => assistantStreamBaseline },
     ))
     await feedList(b, [{ id: 's1' }])
@@ -361,6 +361,7 @@ describe('scope tree', () => {
       },
     }
     const history = ok({
+      cursor: priorMessage.event.seq,
       records: [priorMessage] as never[],
       hasMore: false,
     })
@@ -437,7 +438,7 @@ describe('scope tree', () => {
         surfaceOp: 'append' as const,
       },
     }
-    let records = [prior] as never[]
+    let records: SessionHistoryRecord[] = [prior]
     let assistantStreamBaseline: SessionAssistantStreamBaseline = {
       revision: 2,
       activeAttempt: {
@@ -450,7 +451,7 @@ describe('scope tree', () => {
       },
     }
     b.mock.stream(FOLLOW, followScript(
-      () => ok({ records, hasMore: false }),
+      () => ok({ cursor: records.at(-1)?.event.seq ?? -1, records, hasMore: false }),
       { assistantStream: () => assistantStreamBaseline },
     ))
     await feedList(b, [{ id: 's1' }])

packages/api/session-controller/tests/state-controller.host.spec.ts

@@ -0,0 +1,480 @@
+import { Context } from '@deepseek-ai/cordis'
+import type { Agent } from '@deepseek-ai/dsh-agent'
+import { createInboxStub } from '@deepseek-ai/dsh-agent-loop-testkit'
+import { brandString } from '@deepseek-ai/dsh-brand'
+import type AgentRegistry from '@deepseek-ai/dsh-agent'
+import type TypertGatewayService from '@deepseek-ai/dsh-api-gateway'
+import type { PendingRemoteEventInfo } from '@deepseek-ai/dsh-api-gateway'
+import { createUserMessage } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
+import type { SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { HistoryController } from '../src/history/controller.ts'
+import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
+import { StateController } from '../src/state/controller.ts'
+import { StateListener } from '../src/state/listener.ts'
+import type { SessionState, SessionRequestId } from '../src/types.ts'
+import { installSessionReadTestServices } from './test-remote.ts'
+
+const roots: Context[] = []
+const aborters: AbortController[] = []
+const operations: Promise<unknown>[] = []
+function cancellation(): AbortController {
+  const abort = new AbortController()
+  aborters.push(abort)
+  return abort
+}
+function tracked<T>(promise: Promise<T>): Promise<T> {
+  operations.push(promise)
+  void promise.catch(() => undefined)
+  return promise
+}
+function deferred<T>() {
+  let resolve!: (value: T) => void
+  const promise = new Promise<T>((settle) => { resolve = settle })
+  return { promise, resolve }
+}
+afterEach(async () => {
+  for (const abort of aborters.splice(0)) abort.abort()
+  await Promise.allSettled(operations.splice(0))
+  await Promise.all(roots.splice(0).map(ctx => ctx.fiber.dispose()))
+  vi.restoreAllMocks()
+  vi.useRealTimers()
+})
+
+async function fixture() {
+  const ctx = new Context()
+  roots.push(ctx)
+  await ctx.plugin(SessionStore)
+  installSessionReadTestServices(ctx)
+  const session = ctx.sessions.create(SessionId('state-target'), { meta: { cwd: '/state-test' } })
+  let agent: Agent | undefined
+  const get = vi.fn(() => agent)
+  const agents: Partial<AgentRegistry> = { get }
+  ctx.provide('agents', agents as AgentRegistry)
+  const pending: PendingRemoteEventInfo[] = []
+  const pendingRead = vi.fn(() => pending)
+  const gateway: Partial<TypertGatewayService> = { getPendingRemoteEvents: pendingRead }
+  ctx.provide('typertGateway', gateway as TypertGatewayService)
+  const history = new HistoryController(ctx)
+  const output = vi.spyOn(history, 'assistantOutput').mockResolvedValue({ records: [], truncated: false })
+  const controller = new StateController(ctx, history)
+  const abort = cancellation()
+  const request = { sessionId: session.id }
+  const live = (status: Agent['status'] = 'running', idle = Promise.resolve(undefined)) => {
+    const value = {
+      id: session.id, session, status,
+      inbox: createInboxStub(),
+      whenIdle: vi.fn(() => idle),
+    }
+    const partial: Partial<Agent> = value
+    agent = partial as Agent
+    return value
+  }
+  const addPending = (event: string, id = session.id) => {
+    pending.push({ event, agentId: brandString<PendingRemoteEventInfo['agentId']>(id) })
+  }
+  return {
+    ctx, session, controller, history, output, abort, request, live, get, pending, pendingRead, addPending,
+    setAgent: (value: Agent | undefined) => { agent = value },
+  }
+}
+const emptyState: SessionState = { cursor: -1, active: false, running: false, queue: null, pendingInput: [] }
+
+describe('StateController', () => {
+  it('re-reads a state whose cold read turned live', async () => {
+    const f = await fixture()
+    const agent = f.live('running')
+    // The first lookup misses; the Session is a live source by the time the cold read runs.
+    let lookups = 0
+    f.get.mockImplementation(() => (lookups++ === 0 ? undefined : agent as never))
+    const state = await f.controller.state(f.request, new AbortController().signal)
+    expect(state).toMatchObject({ active: true, running: true, queue: [] })
+    expect(state.cursor).toBe(f.session.seq - 1)
+    expect(lookups).toBeGreaterThan(1)
+  })
+
+  it('rejects a live state read whose Session records no cwd', async () => {
+    const f = await fixture()
+    const session = f.ctx.sessions.create(SessionId('no-cwd-live'))
+    const partial: Partial<Agent> = {
+      id: session.id, session, status: 'idle', inbox: createInboxStub(), whenIdle: vi.fn(() => Promise.resolve(undefined)),
+    }
+    f.setAgent(partial as Agent)
+    await expect(f.controller.state({ sessionId: session.id }, new AbortController().signal))
+      .rejects.toMatchObject({ code: 'session/not-found' })
+  })
+
+  it('rejects a cold state read whose stored header records no cwd', async () => {
+    const f = await fixture()
+    const header: SessionHeader = {
+      version: SESSION_FORMAT_VERSION, id: f.session.id, createdAt: 1, isSeeded: false,
+    }
+    const observation: SessionObservation = {
+      source: 'prepared',
+      header,
+      inheritedEventCount: SessionLogOffset(0),
+      events: [],
+      cursor: -1,
+      projections: { asOfSeq: -1, values: {} },
+      retain: () => observation,
+      [Symbol.dispose]: () => {},
+    }
+    vi.spyOn(f.history, 'usingObservation').mockImplementation(async (_request, _signal, callback) =>
+      callback?.(observation))
+    await expect(f.controller.state(f.request, new AbortController().signal))
+      .rejects.toMatchObject({ code: 'session/not-found' })
+  })
+
+  it('reads cold state without projections and reports classified, deduplicated pending input', async () => {
+    const f = await fixture()
+    const read = vi.spyOn(f.history, 'usingObservation')
+    f.addPending('approval/request')
+    f.addPending('user-questions/request')
+    f.addPending('fixture/request')
+    f.addPending('approval/request')
+    f.addPending('ignored', SessionId('other'))
+    expect(await f.controller.state(f.request, f.abort.signal)).toEqual({ ...emptyState, pendingInput: ['approval', 'question', 'other'] })
+    expect(read).toHaveBeenCalledWith({ sessionId: f.session.id, projection: false }, f.abort.signal, expect.any(Function))
+  })
+
+  it('preserves live queues, RPC identities and steering versus context placement', async () => {
+    const f = await fixture()
+    const queued = createUserMessage({ content: [{ type: 'text', text: 'queued' }], source: { kind: 'user' } })
+    const steering = createUserMessage({ content: [{ type: 'text', text: 'steer' }], source: { kind: 'user', rpcId: 'rpc' as SessionRequestId } })
+    const context = createUserMessage({ content: [{ type: 'text', text: 'context' }], source: { kind: 'user-approval' } })
+    const agent = f.live()
+    agent.inbox.append('next-turn', queued)
+    agent.inbox.append('next-step', steering)
+    agent.inbox.append('next-step', context)
+    expect(await f.controller.state(f.request, f.abort.signal)).toEqual({
+      cursor: -1, active: true, running: true, pendingInput: [],
+      queue: [
+        { id: queued.id, placement: 'queued', content: queued.content },
+        { id: steering.id, placement: 'steering', content: steering.content, rpcId: 'rpc' },
+        { id: context.id, placement: 'context', content: context.content },
+      ],
+    })
+  })
+
+  it('rejects already cancelled state reads', async () => {
+    const f = await fixture()
+    f.abort.abort(new Error('cancelled'))
+    await expect(f.controller.state(f.request, f.abort.signal)).rejects.toThrow('cancelled')
+  })
+
+  it('observes complete replacements and closes a pending read on cancellation', async () => {
+    const f = await fixture()
+    const states = f.controller.observeState(f.request, f.abort.signal)[Symbol.asyncIterator]()
+    expect(await states.next()).toMatchObject({ value: emptyState })
+    f.live()
+    const next = tracked(states.next())
+    f.session.append('turn/start', { turn: 1 })
+    expect(await next).toMatchObject({ value: { active: true, running: true, cursor: 0 } })
+    const end = tracked(states.next())
+    f.abort.abort()
+    expect(await end).toEqual({ done: true, value: undefined })
+    await states.return?.()
+  })
+
+  it('returns inactive idle state and delegates the exact output window with caller signal', async () => {
+    const f = await fixture()
+    const result = await f.controller.wait({ ...f.request, afterSeq: 3, maxMessages: 2 }, f.abort.signal)
+    expect(result).toEqual({ reason: 'idle', state: emptyState, output: { records: [], truncated: false } })
+    expect(f.output).toHaveBeenCalledWith(f.session.id, 3, -1, 2, f.abort.signal)
+  })
+
+  it('stops immediately for pending input even while the Agent runs', async () => {
+    const f = await fixture()
+    f.live()
+    f.addPending('approval/request')
+    expect(await f.controller.wait(f.request, f.abort.signal)).toMatchObject({ reason: 'input-required' })
+  })
+
+  it('confirms an active idle Agent with whenIdle', async () => {
+    const f = await fixture()
+    const agent = f.live('idle')
+    expect(await f.controller.wait(f.request, f.abort.signal)).toMatchObject({ reason: 'idle', state: { active: true } })
+    expect(agent.whenIdle).toHaveBeenCalledOnce()
+  })
+
+  it('polls pending input after one second without Gateway notifications', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    f.live()
+    const result = tracked(f.controller.wait(f.request, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(0)
+    expect(f.pendingRead).toHaveBeenCalledTimes(1)
+    f.addPending('user-questions/request')
+    await vi.advanceTimersByTimeAsync(999)
+    expect(f.pendingRead).toHaveBeenCalledTimes(1)
+    await vi.advanceTimersByTimeAsync(1)
+    expect(await result).toMatchObject({ reason: 'input-required', state: { pendingInput: ['question'] } })
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('uses the latest accepted state on timeout and still reads output with the caller signal', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    f.live()
+    const result = tracked(f.controller.wait({ ...f.request, timeoutMs: 1500 }, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(1500)
+    expect(await result).toMatchObject({ reason: 'timeout', state: { running: true } })
+    expect(f.output).toHaveBeenCalledWith(f.session.id, -1, -1, 50, f.abort.signal)
+    expect(f.abort.signal.aborted).toBe(false)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('responds to a standard event before the polling tick and races idle against new state', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    const idle = deferred<undefined>()
+    f.live('idle', idle.promise)
+    const result = tracked(f.controller.wait(f.request, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(0)
+    f.addPending('approval/request')
+    f.session.append('turn/start', { turn: 1 })
+    await vi.advanceTimersByTimeAsync(0)
+    expect(await result).toMatchObject({ reason: 'input-required', state: { cursor: 0 } })
+    idle.resolve(undefined)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('propagates cancellation while waiting and cleans periodic reads', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    f.live()
+    const result = tracked(f.controller.wait(f.request, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(0)
+    f.abort.abort(new Error('caller cancelled'))
+    await expect(result).rejects.toThrow('caller cancelled')
+    expect(f.output).not.toHaveBeenCalled()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('propagates an idle-driver failure without treating it as a deadline', async () => {
+    const f = await fixture()
+    const idle = deferred<undefined>()
+    const agent = f.live('idle', idle.promise)
+    agent.whenIdle.mockImplementation(() => Promise.reject(new Error('driver failed')))
+    await expect(f.controller.wait(f.request, f.abort.signal)).rejects.toThrow('driver failed')
+    idle.resolve(undefined)
+  })
+
+  it('propagates a structured read failure raised before the deadline', async () => {
+    const f = await fixture()
+    const failure = deferred<undefined>()
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      await failure.promise
+      throw new RemoteError('gateway/internal', 'persistent read failed', {})
+    })
+    const result = tracked(f.controller.wait({ ...f.request, timeoutMs: 5_000 }, f.abort.signal))
+    failure.resolve(undefined)
+    await expect(result).rejects.toMatchObject({ code: 'gateway/internal' })
+  })
+
+  it('keeps a structured read failure raised after the deadline', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    let rejectRead!: (error: Error) => void
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      await new Promise<void>((_resolve, reject) => { rejectRead = reject })
+      yield* []
+    })
+    const result = tracked(f.controller.wait({ ...f.request, timeoutMs: 20 }, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(20)
+    rejectRead(new RemoteError('gateway/internal', 'persistent read failed', {}))
+    await expect(result).rejects.toMatchObject({ code: 'gateway/internal' })
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('reports timeout when its first cold read never yields a state', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    vi.spyOn(f.ctx.sessionQuery, 'observeSession').mockImplementation((_id, options) => new Promise((_resolve, reject) => {
+      const abort = options?.signal
+      abort?.addEventListener('abort', () => {
+        const reason: unknown = abort.reason
+        reject(Object.assign(new Error('session observation was aborted'), {
+          code: 'SESSION_QUERY_ABORTED', cause: reason,
+        }))
+      }, { once: true })
+    }))
+    const result = tracked(f.controller.wait({ ...f.request, timeoutMs: 20 }, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(20)
+    await expect(result).rejects.toMatchObject({ code: 'session/wait-timeout' })
+    expect(f.output).not.toHaveBeenCalled()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('rejects a state stream that ends before a stopping observation', async () => {
+    const f = await fixture()
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {})
+    await expect(f.controller.wait(f.request, f.abort.signal)).rejects.toMatchObject({ code: 'session/observation-ended' })
+  })
+
+  it('settles an already-armed failed read when the preceding observation stops the wait', async () => {
+    const f = await fixture()
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      yield { state: emptyState, revision: 0 }
+      throw new Error('next read failed')
+    })
+    expect(await f.controller.wait(f.request, f.abort.signal)).toMatchObject({ reason: 'idle' })
+  })
+
+  it('ignores a yielded state invalidated before wait accepts it', async () => {
+    const f = await fixture()
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      yield { state: { ...emptyState, running: true }, revision: 0 }
+      yield { state: emptyState, revision: 1 }
+    })
+    vi.spyOn(StateListener.prototype, 'isCurrent').mockReturnValueOnce(false).mockReturnValue(true)
+    expect(await f.controller.wait(f.request, f.abort.signal)).toMatchObject({ reason: 'idle', state: emptyState })
+  })
+
+  it('times out with the last accepted state, not a later invalidated sample', async () => {
+    const f = await fixture()
+    vi.useFakeTimers()
+    const accepted = { ...emptyState, running: true }
+    const stale = { ...emptyState, running: true, cursor: 7 }
+    const never = deferred<undefined>()
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      yield { state: accepted, revision: 0 }
+      yield { state: stale, revision: 0 }
+      await never.promise
+      yield* []
+    })
+    vi.spyOn(StateListener.prototype, 'isCurrent').mockReturnValueOnce(true).mockReturnValue(false)
+    const result = tracked(f.controller.wait({ ...f.request, timeoutMs: 20 }, f.abort.signal))
+    await vi.advanceTimersByTimeAsync(20)
+    never.resolve(undefined)
+    await expect(result).resolves.toMatchObject({ reason: 'timeout', state: { cursor: -1 } })
+  })
+
+  it.each(['missing-agent', 'different-session'] as const)('retries active state when its owner is %s', async (condition) => {
+    const f = await fixture()
+    const agent = f.live('idle')
+    if (condition === 'missing-agent') f.setAgent(undefined)
+    else Object.assign(agent, { session: f.ctx.sessions.create(SessionId('replacement')) })
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      yield { state: { ...emptyState, active: true, queue: [] }, revision: 0 }
+      yield { state: emptyState, revision: 0 }
+    })
+    expect(await f.controller.wait(f.request, f.abort.signal)).toMatchObject({ reason: 'idle', state: emptyState })
+    expect(agent.whenIdle).not.toHaveBeenCalled()
+  })
+
+  it.each(['revision', 'agent', 'session'] as const)('rechecks %s after the idle promise settles', async (changed) => {
+    const f = await fixture()
+    const idle = deferred<undefined>()
+    const entered = deferred<undefined>()
+    const next = deferred<undefined>()
+    const agent = f.live('idle', idle.promise)
+    agent.whenIdle.mockImplementation(() => { entered.resolve(undefined); return idle.promise })
+    vi.spyOn(StateListener.prototype, 'observeStates').mockImplementation(async function* () {
+      yield { state: { ...emptyState, active: true, queue: [] }, revision: 0 }
+      await next.promise
+      yield { state: emptyState, revision: 0 }
+    })
+    const rechecked = deferred<undefined>()
+    let checks = 0
+    vi.spyOn(StateListener.prototype, 'isCurrent').mockImplementation(() => {
+      checks++
+      if (checks === 2) rechecked.resolve(undefined)
+      return !(changed === 'revision' && checks === 2)
+    })
+    const result = tracked(f.controller.wait(f.request, f.abort.signal))
+    try {
+      await entered.promise
+      if (changed === 'agent') f.setAgent(undefined)
+      if (changed === 'session') Object.assign(agent, { session: f.ctx.sessions.create(SessionId('new-session')) })
+      idle.resolve(undefined)
+      await rechecked.promise
+      next.resolve(undefined)
+      expect(await result).toMatchObject({ reason: 'idle', state: emptyState })
+    } finally {
+      idle.resolve(undefined)
+      next.resolve(undefined)
+      f.abort.abort()
+      await Promise.allSettled([result])
+    }
+  })
+
+  it.each([
+    { afterSeq: -2 }, { afterSeq: -0 }, { afterSeq: 1.5 },
+    { maxMessages: 0 }, { maxMessages: 1.5 },
+    { timeoutMs: 0 }, { timeoutMs: 2147483648 }, { timeoutMs: 1.5 },
+  ])('rejects unsupported wait bounds %j', async (bounds) => {
+    const f = await fixture()
+    await expect(f.controller.wait({ ...f.request, ...bounds }, f.abort.signal)).rejects.toMatchObject({ code: 'gateway/bad-request' })
+  })
+})
+
+describe('StateListener', () => {
+  it('discards an async stale read and coalesces repeated invalidations', async () => {
+    const f = await fixture()
+    const first = deferred<SessionState>()
+    const reading = deferred<undefined>()
+    const read = vi.fn()
+      .mockImplementationOnce(() => { reading.resolve(undefined); return first.promise })
+      .mockResolvedValue({ ...emptyState, cursor: 1 })
+    using listener = new StateListener(f.ctx, f.session.id, f.abort.signal, read)
+    const states = listener.observeStates()
+    const pending = tracked(states.next())
+    await reading.promise
+    f.session.append('turn/start', { turn: 1 })
+    f.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    first.resolve(emptyState)
+    expect(await pending).toEqual({ done: false, value: { state: { ...emptyState, cursor: 1 }, revision: 2 } })
+    expect(read).toHaveBeenCalledTimes(2)
+    expect(listener.isCurrent(0)).toBe(false)
+    expect(listener.isCurrent(2)).toBe(true)
+    await states.return(undefined)
+    expect(listener.isCurrent(2)).toBe(false)
+  })
+
+  it('ignores other Sessions and ends when its Context is disposed', async () => {
+    const f = await fixture()
+    const read = vi.fn(async () => emptyState)
+    const listener = new StateListener(f.ctx, f.session.id, f.abort.signal, read)
+    const states = listener.observeStates()
+    await states.next()
+    const pending = tracked(states.next())
+    f.ctx.sessions.create(SessionId('unrelated'))
+    expect(read).toHaveBeenCalledOnce()
+    await f.ctx.fiber.dispose()
+    expect(await pending).toEqual({ done: true, value: undefined })
+    listener[Symbol.dispose]()
+  })
+
+  it('invalidates for each Agent lifecycle notification', async () => {
+    const f = await fixture()
+    f.live('idle')
+    const agent = f.get()!
+    using listener = new StateListener(f.ctx, f.session.id, f.abort.signal, async () => emptyState)
+    const states = listener.observeStates()
+    await states.next()
+    for (const emit of [
+      () => { f.ctx.emit('agent/status', { agent, status: 'running' }) },
+      () => { f.ctx.emit('agent/created', { agent, source: 'startup' }) },
+      () => { f.ctx.emit('agent/disposed', { agent }) },
+    ]) {
+      const pending = tracked(states.next())
+      emit()
+      expect(await pending).toMatchObject({ done: false, value: { state: emptyState } })
+    }
+    await states.return(undefined)
+  })
+
+  it('rejects cancelled construction and contains setup failure by removing abort listeners', async () => {
+    const f = await fixture()
+    f.abort.abort()
+    expect(() => new StateListener(f.ctx, f.session.id, f.abort.signal, async () => emptyState)).toThrow()
+    const abort = cancellation()
+    const remove = vi.spyOn(abort.signal, 'removeEventListener')
+    vi.spyOn(f.ctx, 'on').mockImplementation(() => { throw new Error('registration failed') })
+    expect(() => new StateListener(f.ctx, f.session.id, abort.signal, async () => emptyState)).toThrow('registration failed')
+    expect(remove).toHaveBeenCalledWith('abort', expect.any(Function))
+  })
+})

packages/api/session-controller/tests/subagent-catalog.host.spec.ts

@@ -105,7 +105,7 @@ describe('SessionController subagent catalog', () => {
       kind: 'sequenced', values: { subagentCatalog: [{ id: CHILD, mode: 'continuable', label: 'worker' }] },
     })
     expect(stat).toHaveBeenCalledWith(PARENT, { signal })
-    expect(observe).toHaveBeenCalledWith(PARENT, { signal })
+    expect(observe).toHaveBeenCalledWith(PARENT, { signal, projectionMode: 'all' })
     expect(open).toHaveBeenCalledWith(PARENT, 'read', { signal })
     expect(resume).not.toHaveBeenCalled()
     expect(ctx.sessions.get(PARENT)).toBeUndefined()

packages/api/session-controller/tests/test-remote.ts

@@ -51,7 +51,7 @@ import type {
   SessionOpenWorkspacePathRequest,
   SessionOpenWorkspacePathValue,
   SessionWorkspacePathApplication,
-  SessionPage,
+  SessionEventsPage,
   SessionPageRequest,
   SessionPromptRequest,
   SessionPromptValue,
@@ -66,7 +66,7 @@ import type {
   SessionUpdateQueueValue,
 } from '../src/types.ts'
 
-/** Direct test face matching the generated `ctx.remote.session` unary methods. */
+/** Direct test face for the generated `ctx.remote.session` unary methods these tests exercise. */
 export interface TestSessionRemote {
   workspacePathApplications(
     request: { readonly path: string }, signal?: AbortSignal,
@@ -87,7 +87,7 @@ export interface TestSessionRemote {
     request: SessionOpenWorkspacePathRequest,
     signal?: AbortSignal,
   ): Promise<RemoteResult<SessionOpenWorkspacePathValue>>
-  page(request: SessionPageRequest, signal?: AbortSignal): Promise<RemoteResult<SessionPage>>
+  page(request: SessionPageRequest, signal?: AbortSignal): Promise<RemoteResult<SessionEventsPage>>
   projections(request: SessionProjectionsRequest, signal?: AbortSignal): Promise<RemoteResult<SessionProjectionsValue>>
   follow(request: SessionFollowRequest, signal?: AbortSignal): AsyncIterable<SessionFollowFrame>
   control(signal?: AbortSignal): AsyncIterable<SessionControlFrame>
@@ -388,6 +388,7 @@ export async function createSessionTestRemote(
       signal,
     ),
     page: (request, signal = new AbortController().signal) => remoteResult(
+      // oxlint-disable-next-line typescript/no-deprecated -- Test face keeps the legacy page entry covered.
       () => direct.page(request, signal),
       signal,
     ),

packages/api/session-controller/tests/transport.client.spec.ts

@@ -25,7 +25,7 @@ import type {
   SessionFollowFrame,
   SessionFollowRequest,
   SessionHistoryRecord,
-  SessionPage,
+  SessionEventsPage,
   SessionPageRequest,
   SessionWireEvent,
 } from '../src/types.ts'
@@ -44,8 +44,8 @@ function entry(seq: number): SessionEventEntry {
   return { type: 'event', event: { type: 'turn/start', seq, time: seq, data: { turn: seq } } }
 }
 
-function page(records: readonly SessionHistoryRecord[], hasMore = false): SessionPage {
-  return { records, hasMore }
+function page(records: readonly SessionHistoryRecord[], hasMore = false): SessionEventsPage {
+  return { cursor: records.at(-1)?.event?.seq ?? -1, records, hasMore }
 }
 
 function snapshot(
@@ -102,7 +102,7 @@ class ScriptedSessionRemote implements SessionTransportRemote {
 
   constructor(
     private readonly generations: FollowGeneration[],
-    private readonly pages: RemoteResult<SessionPage>[],
+    private readonly pages: RemoteResult<SessionEventsPage>[],
     private readonly controlFrames: readonly SessionControlFrame[] = [],
     private readonly holdControl = true,
   ) {}
@@ -126,7 +126,7 @@ class ScriptedSessionRemote implements SessionTransportRemote {
     }
   }
 
-  page(request: SessionPageRequest): Promise<RemoteResult<SessionPage>> {
+  page(request: SessionPageRequest): Promise<RemoteResult<SessionEventsPage>> {
     this.pageRequests.push(request)
     const result = this.pages.shift()
     if (result === undefined) throw new Error('no scripted Session page')
@@ -326,7 +326,7 @@ describe('Session Client stream adapters', () => {
     await stream.open({})
     await vi.waitFor(() => { expect(changes).toHaveLength(2) })
 
-    expect(remote.followRequests).toEqual([{ address: ADDRESS, assistantStream: true }])
+    expect(remote.followRequests).toEqual([{ address: ADDRESS, activate: true, assistantStream: true }])
     expect(changes).toMatchObject([
       { type: 'replace', page: { assistantStream: baseline } },
       { type: 'assistant-stream', frame },
@@ -382,7 +382,7 @@ describe('Session Client stream adapters', () => {
         code: 'gateway/internal',
         message: 'session event stream emitted an entry before its opening cursor',
       })
-      expect(remote.followRequests).toEqual([{ address: ADDRESS, assistantStream: true }])
+      expect(remote.followRequests).toEqual([{ address: ADDRESS, activate: true, assistantStream: true }])
     } finally {
       await stream.dispose()
     }
@@ -547,7 +547,7 @@ describe('Session Client stream adapters', () => {
     await stream.prepend({ beforeSeq: 2, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } })
 
     expect(remote.followRequests).toEqual([{
-      address: ADDRESS, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 },
+      address: ADDRESS, activate: true, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 },
     }])
     expect(remote.pageRequests).toEqual([
       { address: ADDRESS, throughSeq: 4, beforeSeq: 2, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
@@ -585,8 +585,8 @@ describe('Session Client stream adapters', () => {
     await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) })
 
     expect(remote.followRequests).toEqual([
-      { address: ADDRESS, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
-      { address: ADDRESS, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
+      { address: ADDRESS, activate: true, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
+      { address: ADDRESS, activate: true, assistantStream: true, maxMessages: 500, turnWindow: { minMessages: 50, minTurns: 2 } },
     ])
     expect(remote.pageRequests).toEqual([])
     expect(changes.map(change => change.type)).toEqual(['replace', 'append', 'replace'])
@@ -616,8 +616,8 @@ describe('Session Client stream adapters', () => {
     finish.resolve(undefined)
     await vi.waitFor(() => { expect(remote.followRequests).toHaveLength(2) })
     expect(remote.followRequests).toEqual([
-      { address: ADDRESS, assistantStream: true },
-      { address: ADDRESS, assistantStream: true },
+      { address: ADDRESS, activate: true, assistantStream: true },
+      { address: ADDRESS, activate: true, assistantStream: true },
     ])
     expect(remote.pageRequests).toEqual([])
     await stream.dispose()

packages/api/session-controller/tests/transport.host.spec.ts

@@ -5,8 +5,9 @@ import type { Session, SessionEvent, SessionHeader, SurfaceIntent } from '@deeps
 import type { SessionObservation } from '@deepseek-ai/dsh-session-query'
 import { snapshotSubagentDescriptor } from '@deepseek-ai/dsh-subagent'
 import { subagentIdentityProjectionDefinition } from '@deepseek-ai/dsh-subagent/src/projection.ts'
-import { describe, expect, it, vi } from 'vitest'
-import { SessionHistoryController } from '../src/history.ts'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { HistoryController } from '../src/history/controller.ts'
+import { WebController } from '../src/web/controller.ts'
 import { installSessionReadTestServices, testSessionPersistence } from './test-remote.ts'
 
 const signal = (): AbortSignal => new AbortController().signal
@@ -72,22 +73,39 @@ function deferred<T>(): Deferred<T> {
   return { promise, resolve }
 }
 
-async function setup(): Promise<{ ctx: Context; transport: SessionHistoryController }> {
+const roots = new Set<Context>()
+afterEach(async () => {
+  await Promise.all([...roots].map(ctx => ctx.fiber.dispose()))
+  roots.clear()
+})
+
+function controllers(
+  ctx: Context, promote: (observation: SessionObservation) => void = (observation) => { observation[Symbol.dispose]() },
+): {
+  history: HistoryController
+  web: WebController
+} {
+  roots.add(ctx.root)
+  installSessionReadTestServices(ctx)
+  const history = new HistoryController(ctx)
+  return { history, web: new WebController(ctx, history, promote) }
+}
+
+async function setup(): Promise<{ ctx: Context; history: HistoryController; web: WebController }> {
   const ctx = new Context()
   await ctx.plugin(SessionStore)
   installSessionReadTestServices(ctx)
   await ctx.sessionProjections.register(subagentIdentityProjectionDefinition)
-  const transport = new SessionHistoryController(ctx, (observation) => { observation[Symbol.dispose]() })
-  return { ctx, transport }
+  return { ctx, ...controllers(ctx) }
 }
 
-describe('SessionHistoryController', () => {
+describe('Session history and Web follow', () => {
   it('opens at the current cursor and follows later events from an ordinary Session', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history, web } = await setup()
     const session = ctx.sessions.create(SessionId('ordinary'), { meta: { cwd: '/workspace' } })
     session.append('turn/start', { turn: 1 })
     const abort = new AbortController()
-    const iterator = transport.follow(
+    const iterator = web.follow(
       { address: { kind: 'session', sessionId: session.id } },
       abort.signal,
     )[Symbol.asyncIterator]()
@@ -99,7 +117,7 @@ describe('SessionHistoryController', () => {
       value: { type: 'event', event: { type: 'turn/end', seq: 1 } },
     })
 
-    const page = await transport.page(
+    const page = await history.history(
       { address: { kind: 'session', sessionId: session.id }, throughSeq: 1 },
       new AbortController().signal,
     )
@@ -113,16 +131,16 @@ describe('SessionHistoryController', () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     installSessionReadTestServices(ctx)
-    let transport!: SessionHistoryController
+    let web!: WebController
     const owner = ctx.plugin(Object.assign(
       (inner: Context) => {
-        transport = new SessionHistoryController(inner, (observation) => { observation[Symbol.dispose]() })
+        web = controllers(inner).web
       },
       { inject: ['sessions', 'sessionQuery'] },
     ))
     await owner.await()
     const session = ctx.sessions.create(SessionId('controller-unload'), { meta: { cwd: '/workspace' } })
-    const iterator = transport.follow(
+    const iterator = web.follow(
       { address: { kind: 'session', sessionId: session.id } },
       new AbortController().signal,
     )[Symbol.asyncIterator]()
@@ -138,13 +156,13 @@ describe('SessionHistoryController', () => {
   })
 
   it('reconnects with a complete replacement snapshot before later live events', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, web } = await setup()
     const session = ctx.sessions.create(SessionId('resume'), { meta: { cwd: '/workspace' } })
     session.append('turn/start', { turn: 1 })
     session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
     session.append('turn/start', { turn: 2 })
     const abort = new AbortController()
-    const iterator = transport.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
     }, abort.signal)[Symbol.asyncIterator]()
 
@@ -168,7 +186,7 @@ describe('SessionHistoryController', () => {
   })
 
   it('subscribes before a cold read and ignores unrelated and replayed buffered events', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, web } = await setup()
     const sessionId = SessionId('cold-race')
     const header: SessionHeader = {
       version: SESSION_FORMAT_VERSION,
@@ -187,7 +205,7 @@ describe('SessionHistoryController', () => {
       inspect: () => inspected.promise,
     }) as never)
     const abort = new AbortController()
-    const iterator = transport.follow({ address: { kind: 'session', sessionId } }, abort.signal)
+    const iterator = web.follow({ address: { kind: 'session', sessionId } }, abort.signal)
       [Symbol.asyncIterator]()
     const opening = iterator.next()
 
@@ -220,9 +238,9 @@ describe('SessionHistoryController', () => {
     }
     const observed = deferred<SessionObservation>()
     ctx.provide('sessionQuery', { observeSession: () => observed.promise } as never)
-    const transport = new SessionHistoryController(ctx, vi.fn())
+    const { web } = controllers(ctx, vi.fn())
     const abort = new AbortController()
-    const iterator = transport.follow({ address: { kind: 'session', sessionId } }, abort.signal)
+    const iterator = web.follow({ address: { kind: 'session', sessionId } }, abort.signal)
       [Symbol.asyncIterator]()
     const opening = iterator.next()
 
@@ -256,11 +274,11 @@ describe('SessionHistoryController', () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     installSessionReadTestServices(ctx)
-    let transport!: SessionHistoryController
+    let web!: WebController
     let agentCtx!: Context
     await ctx.plugin(Object.assign(
       (inner: Context) => {
-        transport = new SessionHistoryController(inner, (observation) => { observation[Symbol.dispose]() })
+        web = controllers(inner).web
       },
       { inject: ['sessions', 'sessionQuery'] },
     ))
@@ -279,7 +297,7 @@ describe('SessionHistoryController', () => {
     const seed = [event('fixture/start', SessionSeq(0))]
     cold(ctx, header, seed)
     const abort = new AbortController()
-    const iterator = transport.follow({ address: { kind: 'session', sessionId } }, abort.signal)
+    const iterator = web.follow({ address: { kind: 'session', sessionId } }, abort.signal)
       [Symbol.asyncIterator]()
 
     await expect(iterator.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: 0 } })
@@ -322,7 +340,7 @@ describe('SessionHistoryController', () => {
       isSeeded: false,
     }
     cold(replay.ctx, replayHeader, [event('fixture/start', SessionSeq(0)), event('fixture/gap', SessionSeq(2))])
-    const replayed = replay.transport.follow({
+    const replayed = replay.web.follow({
       address: { kind: 'session', sessionId: replayId },
     }, signal())[Symbol.asyncIterator]()
     await expect(replayed.next()).rejects.toMatchObject({ code: 'SESSION_QUERY_CORRUPT_SESSION' })
@@ -331,7 +349,7 @@ describe('SessionHistoryController', () => {
     const session = live.ctx.sessions.create(SessionId('live-gap'), { meta: { cwd: '/workspace' } })
     append(session, 'fixture/start', {})
     live.ctx.provide('agents', { get: () => ({ id: session.id }) } as never)
-    const followed = live.transport.follow({
+    const followed = live.web.follow({
       address: { kind: 'session', sessionId: session.id },
     }, signal())[Symbol.asyncIterator]()
     await expect(followed.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: 0 } })
@@ -345,21 +363,21 @@ describe('SessionHistoryController', () => {
   })
 
   it('opens an empty source at cursor -1', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history, web } = await setup()
     const session = ctx.sessions.create(SessionId('empty-follow'), { meta: { cwd: '/workspace' } })
     const abort = new AbortController()
-    const iterator = transport.follow({
+    const iterator = web.follow({
       address: { kind: 'session', sessionId: session.id },
     }, abort.signal)[Symbol.asyncIterator]()
     await expect(iterator.next()).resolves.toMatchObject({ done: false, value: { type: 'snapshot', cursor: -1 } })
-    await expect(transport.page({
+    await expect(history.history({
       address: { kind: 'session', sessionId: session.id }, throughSeq: -1,
     }, signal())).resolves.toMatchObject({ records: [], hasMore: false })
     abort.abort()
     await expect(iterator.next()).resolves.toMatchObject({ done: true })
   })
 
-  it('publishes an empty projection baseline when the query has no registry', async () => {
+  it('fails explicitly when a projected observation omits its baseline', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     const sessionId = SessionId('projectionless-follow')
@@ -380,19 +398,83 @@ describe('SessionHistoryController', () => {
         retain: vi.fn(), [Symbol.dispose]: vi.fn(),
       } satisfies SessionObservation),
     } as never)
-    const history = new SessionHistoryController(ctx, vi.fn())
+    const { web } = controllers(ctx, vi.fn())
     const abort = new AbortController()
-    const iterator = history.follow({ address: { kind: 'session', sessionId } }, abort.signal)
+    const iterator = web.follow({ address: { kind: 'session', sessionId } }, abort.signal)
       [Symbol.asyncIterator]()
 
-    await expect(iterator.next()).resolves.toMatchObject({
-      value: { type: 'snapshot', projections: { asOfSeq: -1, values: {} } },
-    })
+    await expect(iterator.next()).rejects.toMatchObject({ code: 'session/projections-unavailable' })
     abort.abort()
     await expect(iterator.next()).resolves.toMatchObject({ done: true })
     await ctx.fiber.dispose()
   })
 
+  it('does not activate a prepared Session when the request omits activate', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const sessionId = SessionId('passive-activation-default')
+    const header: SessionHeader = {
+      version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false,
+    }
+    const disposeSource = vi.fn()
+    const source: SessionObservation = {
+      source: 'prepared',
+      header,
+      inheritedEventCount: SessionLogOffset(0),
+      events: [],
+      cursor: -1,
+      projections: { asOfSeq: -1, values: {} },
+      retain: () => source,
+      [Symbol.dispose]: disposeSource,
+    }
+    ctx.provide('sessionQuery', { observeSession: () => Promise.resolve(source) } as never)
+    const promote = vi.fn()
+    const { web } = controllers(ctx, promote)
+    const abort = new AbortController()
+    const iterator = web.follow({ address: { kind: 'session', sessionId } }, abort.signal)[Symbol.asyncIterator]()
+    await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'snapshot' } })
+    expect(promote).not.toHaveBeenCalled()
+    abort.abort()
+    await expect(iterator.next()).resolves.toMatchObject({ done: true })
+    expect(disposeSource).toHaveBeenCalled()
+    await ctx.fiber.dispose()
+  })
+
+  it('starts activation for a consumer that stops after the opening snapshot', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const sessionId = SessionId('activation-after-opening')
+    const header: SessionHeader = {
+      version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, cwd: '/workspace', isSeeded: false,
+    }
+    const disposed: string[] = []
+    const promotion: SessionObservation = {
+      source: 'prepared',
+      header,
+      inheritedEventCount: SessionLogOffset(0),
+      events: [],
+      cursor: -1,
+      projections: { asOfSeq: -1, values: {} },
+      retain: vi.fn(),
+      [Symbol.dispose]: () => { disposed.push('promotion') },
+    }
+    const source: SessionObservation = {
+      ...promotion,
+      retain: () => promotion,
+      [Symbol.dispose]: () => { disposed.push('source') },
+    }
+    ctx.provide('sessionQuery', { observeSession: () => Promise.resolve(source) } as never)
+    const promote = vi.fn((observation: SessionObservation) => { observation[Symbol.dispose]() })
+    const { web } = controllers(ctx, promote)
+    const iterator = web.follow({ address: { kind: 'session', sessionId }, activate: true }, signal())
+      [Symbol.asyncIterator]()
+    await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'snapshot' } })
+    await iterator.return?.(undefined)
+    expect(promote).toHaveBeenCalledOnce()
+    expect(disposed).toContain('promotion')
+    await ctx.fiber.dispose()
+  })
+
   it('disposes a retained promotion when background activation rejects synchronously', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -412,18 +494,17 @@ describe('SessionHistoryController', () => {
     ctx.provide('sessionQuery', {
       observeSession: () => Promise.resolve(source),
     } as never)
-    const history = new SessionHistoryController(ctx, () => { throw new Error('activation failed') })
-    const iterator = history.follow({ address: { kind: 'session', sessionId } }, signal())
+    const { web } = controllers(ctx, () => { throw new Error('activation failed') })
+    const iterator = web.follow({ address: { kind: 'session', sessionId }, activate: true }, signal())
       [Symbol.asyncIterator]()
 
-    await expect(iterator.next()).resolves.toMatchObject({ value: { type: 'snapshot' } })
     await expect(iterator.next()).rejects.toThrow('activation failed')
     expect(disposePromotion).toHaveBeenCalledOnce()
     await ctx.fiber.dispose()
   })
 
   it('requires the durable parent and mode for a direct subagent address', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history } = await setup()
     const parentSessionId = SessionId('parent')
     const childSessionId = SessionId('child')
     ctx.sessions.create(parentSessionId, { meta: { cwd: '/workspace' } })
@@ -437,7 +518,7 @@ describe('SessionHistoryController', () => {
     }))
     const signal = new AbortController().signal
 
-    await expect(transport.page({
+    await expect(history.history({
       address: {
         kind: 'subagent',
         parentSessionId,
@@ -448,17 +529,17 @@ describe('SessionHistoryController', () => {
     }, signal)).resolves.toMatchObject({
       records: [{ type: 'event', event: { type: 'subagent/descriptor' } }],
     })
-    await expect(transport.page({
+    await expect(history.history({
       address: { kind: 'subagent', parentSessionId, childSessionId, mode: 'unknown' },
       throughSeq: 0,
     }, signal)).resolves.toMatchObject({
       records: [{ type: 'event', event: { type: 'subagent/descriptor' } }],
     })
-    await expect(transport.page({
+    await expect(history.history({
       address: { kind: 'subagent', parentSessionId: SessionId('other-parent'), childSessionId, mode: 'unknown' },
       throughSeq: 0,
     }, signal)).rejects.toMatchObject({ code: 'subagent/unauthorized' })
-    await expect(transport.page({
+    await expect(history.history({
       address: {
         kind: 'subagent',
         parentSessionId: SessionId('other-parent'),
@@ -467,7 +548,7 @@ describe('SessionHistoryController', () => {
       },
       throughSeq: 0,
     }, signal)).rejects.toMatchObject({ code: 'subagent/unauthorized' })
-    await expect(transport.page({
+    await expect(history.history({
       address: {
         kind: 'subagent',
         parentSessionId,
@@ -476,14 +557,14 @@ describe('SessionHistoryController', () => {
       },
       throughSeq: 0,
     }, signal)).rejects.toMatchObject({ code: 'subagent/unauthorized' })
-    await expect(transport.page({
+    await expect(history.history({
       address: { kind: 'session', sessionId: childSessionId },
       throughSeq: 0,
     }, signal)).rejects.toMatchObject({ code: 'session/agent-busy' })
   })
 
   it('preserves a cold inspection failure for the Gateway error branch', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history } = await setup()
     const sessionId = SessionId('corrupt-cold')
     const failure = new Error('cold log is corrupt')
     const header: SessionHeader = { version: SESSION_FORMAT_VERSION, id: sessionId, createdAt: 1, isSeeded: false, cwd: '/workspace' }
@@ -492,7 +573,7 @@ describe('SessionHistoryController', () => {
       inspect: () => Promise.reject(failure),
     }) as never)
 
-    await expect(transport.page({
+    await expect(history.history({
       address: { kind: 'session', sessionId },
       throughSeq: -1,
     }, new AbortController().signal)).rejects.toMatchObject({
@@ -502,7 +583,7 @@ describe('SessionHistoryController', () => {
   })
 
   it('rejects malformed page and follow cursors at the service boundary', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history, web } = await setup()
     const session = ctx.sessions.create(SessionId('validation'), { meta: { cwd: '/workspace' } })
     const address = { kind: 'session' as const, sessionId: session.id }
     for (const request of [
@@ -521,9 +602,9 @@ describe('SessionHistoryController', () => {
       { address, throughSeq: -1, turnWindow: { minMessages: 50, minTurns: 0 } },
       { address, throughSeq: -1, turnWindow: { minMessages: 50, minTurns: 1.5 } },
     ]) {
-      await expect(transport.page(request, signal())).rejects.toMatchObject({ code: 'gateway/bad-request' })
+      await expect(history.history(request, signal())).rejects.toMatchObject({ code: 'gateway/bad-request' })
     }
-    await expect(transport.page({ address, throughSeq: 0 }, signal()))
+    await expect(history.history({ address, throughSeq: 0 }, signal()))
       .rejects.toMatchObject({ code: 'gateway/bad-request' })
 
     const corrupt = await setup()
@@ -533,11 +614,11 @@ describe('SessionHistoryController', () => {
       { version: SESSION_FORMAT_VERSION, id: corruptId, createdAt: 1, cwd: '/workspace', isSeeded: false },
       [event('fixture/start', SessionSeq(0)), event('fixture/gap', SessionSeq(2))],
     )
-    await expect(corrupt.transport.page({
+    await expect(corrupt.history.history({
       address: { kind: 'session', sessionId: corruptId }, throughSeq: 1,
     }, signal())).rejects.toMatchObject({ code: 'SESSION_QUERY_CORRUPT_SESSION' })
     for (const maxMessages of [0, 0.5]) {
-      const iterator = transport.follow({ address, maxMessages }, signal())[Symbol.asyncIterator]()
+      const iterator = web.follow({ address, maxMessages }, signal())[Symbol.asyncIterator]()
       await expect(iterator.next()).rejects.toMatchObject({ code: 'gateway/bad-request' })
     }
     for (const turnWindow of [
@@ -547,15 +628,15 @@ describe('SessionHistoryController', () => {
       { minMessages: 50, minTurns: 0 },
       { minMessages: 50, minTurns: 1.5 },
     ]) {
-      const iterator = transport.follow({ address, turnWindow }, signal())[Symbol.asyncIterator]()
+      const iterator = web.follow({ address, turnWindow }, signal())[Symbol.asyncIterator]()
       await expect(iterator.next()).rejects.toMatchObject({ code: 'gateway/bad-request' })
     }
   })
 
   it('reports missing ordinary and subagent sources without fabricating inspection failures', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history } = await setup()
     const ordinary = { kind: 'session' as const, sessionId: SessionId('missing') }
-    await expect(transport.page({ address: ordinary, throughSeq: -1 }, signal()))
+    await expect(history.history({ address: ordinary, throughSeq: -1 }, signal()))
       .rejects.toMatchObject({ code: 'session/not-found' })
 
     const inspect = vi.fn(() => Promise.resolve(undefined))
@@ -565,9 +646,9 @@ describe('SessionHistoryController', () => {
       stat,
       inspect,
     }) as never)
-    await expect(transport.page({ address: ordinary, throughSeq: -1 }, signal()))
+    await expect(history.history({ address: ordinary, throughSeq: -1 }, signal()))
       .rejects.toMatchObject({ code: 'session/not-found' })
-    await expect(transport.page({
+    await expect(history.history({
       address: {
         kind: 'subagent',
         parentSessionId: SessionId('parent'),
@@ -594,7 +675,7 @@ describe('SessionHistoryController', () => {
         events: [],
       }),
     }) as never)
-    await expect(first.transport.page({ address, throughSeq: -1 }, signal()))
+    await expect(first.history.history({ address, throughSeq: -1 }, signal()))
       .rejects.toMatchObject({ code: 'session/not-found' })
 
     const second = await setup()
@@ -608,7 +689,7 @@ describe('SessionHistoryController', () => {
         events: [],
       }),
     }) as never)
-    await expect(second.transport.page({ address, throughSeq: -1 }, signal()))
+    await expect(second.history.history({ address, throughSeq: -1 }, signal()))
       .rejects.toMatchObject({ code: 'session/not-found' })
   })
 
@@ -623,7 +704,7 @@ describe('SessionHistoryController', () => {
       isSeeded: false,
     }
     cold(ordinaryBench.ctx, ordinaryHeader, [event('turn/start', SessionSeq(0), { turn: 1 })])
-    await expect(ordinaryBench.transport.page({
+    await expect(ordinaryBench.history.history({
       address: { kind: 'session', sessionId: ordinaryId },
       throughSeq: 0,
     }, signal())).resolves.toMatchObject({
@@ -649,18 +730,18 @@ describe('SessionHistoryController', () => {
     }
     const missing = await setup()
     cold(missing.ctx, childHeader, [])
-    await expect(missing.transport.page({ address: childAddress, throughSeq: -1 }, signal()))
+    await expect(missing.history.history({ address: childAddress, throughSeq: -1 }, signal()))
       .rejects.toMatchObject({ code: 'subagent/catalog-diagnostic', details: { reason: 'corrupt' } })
 
     const corrupt = await setup()
     cold(corrupt.ctx, childHeader, [event('subagent/descriptor', SessionSeq(0), { version: 'bad' })])
-    await expect(corrupt.transport.page({ address: childAddress, throughSeq: 0 }, signal()))
+    await expect(corrupt.history.history({ address: childAddress, throughSeq: 0 }, signal()))
       .rejects.toMatchObject({ code: 'subagent/catalog-diagnostic', details: { reason: 'corrupt' } })
 
     const ordinaryChild = await setup()
     const { origin: _origin, ...ordinaryChildHeader } = childHeader
     cold(ordinaryChild.ctx, ordinaryChildHeader, [])
-    await expect(ordinaryChild.transport.page({ address: childAddress, throughSeq: -1 }, signal()))
+    await expect(ordinaryChild.history.history({ address: childAddress, throughSeq: -1 }, signal()))
       .rejects.toMatchObject({ code: 'subagent/unauthorized' })
   })
 
@@ -689,9 +770,9 @@ describe('SessionHistoryController', () => {
         retain: vi.fn(), [Symbol.dispose]: vi.fn(),
       } as unknown as SessionObservation),
     } as never)
-    const history = new SessionHistoryController(ctx, vi.fn())
+    const { history } = controllers(ctx, vi.fn())
 
-    await expect(history.page({
+    await expect(history.history({
       address: {
         kind: 'subagent',
         parentSessionId,
@@ -710,7 +791,7 @@ describe('SessionHistoryController', () => {
     const session = ordinary.ctx.sessions.create(SessionId('projected'), { meta: { cwd: '/workspace' } })
     session.append('turn/start', { turn: 1 })
     const ordinarySnapshot = vi.spyOn(ordinary.ctx.sessionProjections, 'snapshot')
-    const ordinaryPage = await ordinary.transport.page({
+    const ordinaryPage = await ordinary.history.history({
       address: { kind: 'session', sessionId: session.id },
       throughSeq: 0,
     }, signal())
@@ -727,7 +808,7 @@ describe('SessionHistoryController', () => {
       mode: 'continuable', provider: 'test', label: 'child',
     }))
     const childSnapshot = vi.spyOn(child.ctx.sessionProjections, 'snapshot')
-    const page = await child.transport.page({
+    const page = await child.history.history({
       address: {
         kind: 'subagent',
         parentSessionId,
@@ -741,7 +822,7 @@ describe('SessionHistoryController', () => {
   })
 
   it('keeps message-aligned pagination contiguous across replacement source-event references', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history } = await setup()
     const session = ctx.sessions.create(SessionId('pagination'), { meta: { cwd: '/workspace' } })
     session.append('turn/start', { turn: 1 })
     append(session, 'user/message', { content: [], source: { kind: 'user' } }, { surfaceOp: 'append' })
@@ -754,27 +835,27 @@ describe('SessionHistoryController', () => {
       sourceEventSeqs: [SessionSeq(1), firstReply.seq, SessionSeq(3), SessionSeq(4), summary.seq],
     })
 
-    const page = await transport.page({
+    const page = await history.history({
       address: { kind: 'session', sessionId: session.id }, throughSeq: replacement.seq, maxMessages: 2,
     }, signal())
     expect(page.records.map(entry => entry.event.seq))
       .toEqual([3, 4, 5, replacement.seq])
     expect(page.hasMore).toBe(true)
-    const before = await transport.page({
+    const before = await history.history({
       address: { kind: 'session', sessionId: session.id }, throughSeq: replacement.seq, beforeSeq: 3, maxMessages: 1,
     }, signal())
     expect(before.records.map(entry => entry.event.seq)).toEqual([2])
   })
 
   it('keeps cited source events in the page that owns their appended message', async () => {
-    const { ctx, transport } = await setup()
+    const { ctx, history } = await setup()
     const session = ctx.sessions.create(SessionId('pagination-sources'), { meta: { cwd: '/workspace' } })
     const source = append(session, 'fixture/source', {})
     append(session, 'user/message', { content: [], source: { kind: 'plugin' } }, {
       surfaceOp: 'append', sourceEventSeqs: [source.seq],
     })
 
-    const page = await transport.page({
+    const page = await history.history({
       address: { kind: 'session', sessionId: session.id }, throughSeq: 1, maxMessages: 1,
     }, signal())
     expect(page.records.map(entry => entry.event.seq)).toEqual([0, 1])