test(api-session): bind history probes to follow cursors

apps/web/tests/seeded-history.e2e.ts

@@ -182,6 +182,7 @@ describe('web e2e: seeded history renders through cold resume', () => {
   let browser: Browser
   let page: Page
   let tripwire: ReturnType<typeof watchConsole>
+  let seededThroughSeq = -1
 
   beforeAll(async () => {
     scaffold = await launchWebScaffold({})
@@ -201,6 +202,7 @@ describe('web e2e: seeded history renders through cold resume', () => {
       const meter = scaffold.ctx.get('tokenMeter')
       if (meter === undefined) throw new Error('seeded-history requires the host token meter')
       const realizedWithCompaction = withCompaction(realizeSeedFixture(scaffold, raw, SEED_ID), meter)
+      seededThroughSeq = parseSeedFixture(realizedWithCompaction).events.at(-1)?.seq ?? -1
       await seedSession(scaffold, realizedWithCompaction, SEED_ID)
     }
     browser = await chromium.launch()
@@ -238,7 +240,10 @@ describe('web e2e: seeded history renders through cold resume', () => {
       body: JSON.stringify({
         type: 'client-request', rpcId: 'seeded-projections', method: 'session/page',
         payload: {
-          args: { request: { address: { kind: 'session', sessionId: SEED_ID } } },
+          args: { request: {
+            address: { kind: 'session', sessionId: SEED_ID },
+            throughSeq: seededThroughSeq,
+          } },
         },
       }),
     })

apps/web/tests/smoke-real.e2e.ts

@@ -70,6 +70,85 @@ async function remoteRpc<T>(baseUrl: string, endpoint: string, args: object): Pr
   return body.result.value
 }
 
+/** Read the explicit page cut from a freshly opened Session follow stream. */
+async function sessionCursor(baseUrl: string, sessionId: string): Promise<number> {
+  const socket = new WebSocket(`${baseUrl.replace(/^http/, 'ws')}/api/remote.mux`)
+  const streamId = `smoke-history-${randomUUID()}`
+  try {
+    await new Promise<void>((resolve, reject) => {
+      const cleanup = (): void => {
+        socket.removeEventListener('open', opened)
+        socket.removeEventListener('error', failed)
+        socket.removeEventListener('close', closed)
+      }
+      const opened = (): void => {
+        cleanup()
+        resolve()
+      }
+      const failed = (): void => {
+        cleanup()
+        reject(new Error('session/follow carrier failed before opening'))
+      }
+      const closed = (): void => {
+        cleanup()
+        reject(new Error('session/follow carrier closed before opening'))
+      }
+      socket.addEventListener('open', opened)
+      socket.addEventListener('error', failed)
+      socket.addEventListener('close', closed)
+    })
+    return await new Promise<number>((resolve, reject) => {
+      const timer = setTimeout(() => { finish(new Error('session/follow did not publish an opening cursor')) }, 10_000)
+      const cleanup = (): void => {
+        clearTimeout(timer)
+        socket.removeEventListener('message', message)
+        socket.removeEventListener('error', failed)
+        socket.removeEventListener('close', closed)
+      }
+      const finish = (error: Error | undefined, cursor?: number): void => {
+        cleanup()
+        if (error !== undefined) reject(error)
+        else resolve(cursor ?? -1)
+      }
+      const message = (event: MessageEvent<unknown>): void => {
+        try {
+          if (typeof event.data !== 'string') throw new Error('session/follow published a non-text frame')
+          const frame: unknown = JSON.parse(event.data)
+          if (!isRecord(frame) || frame.streamId !== streamId) return
+          if (frame.type === 'error') {
+            finish(new Error(`session/follow failed: ${JSON.stringify(frame.error)}`))
+            return
+          }
+          if (frame.type === 'end') {
+            finish(new Error('session/follow ended before its opening cursor'))
+            return
+          }
+          const value = frame.value
+          if (frame.type === 'item' && isRecord(value)
+            && value.type === 'opened' && Number.isSafeInteger(value.cursor)) {
+            finish(undefined, value.cursor as number)
+          }
+        } catch (error) {
+          finish(error instanceof Error ? error : new Error(String(error)))
+        }
+      }
+      const failed = (): void => { finish(new Error('session/follow carrier failed before its opening cursor')) }
+      const closed = (): void => { finish(new Error('session/follow carrier closed before its opening cursor')) }
+      socket.addEventListener('message', message)
+      socket.addEventListener('error', failed)
+      socket.addEventListener('close', closed)
+      socket.send(JSON.stringify({
+        type: 'open',
+        streamId,
+        endpoint: 'session/follow',
+        payload: { args: { request: { address: { kind: 'session', sessionId } } } },
+      }))
+    })
+  } finally {
+    socket.close()
+  }
+}
+
 interface HistoryPage {
   events: { event: { type: string; data: unknown } }[]
   hasMore: boolean
@@ -102,8 +181,9 @@ function hasAssistantMarker(page: HistoryPage, marker: string): boolean {
 }
 
 async function history(baseUrl: string, sessionId: string): Promise<HistoryPage> {
+  const throughSeq = await sessionCursor(baseUrl, sessionId)
   return remoteRpc<HistoryPage>(baseUrl, 'session/page', {
-    request: { address: { kind: 'session', sessionId }, maxMessages: 10 },
+    request: { address: { kind: 'session', sessionId }, throughSeq, maxMessages: 10 },
   })
 }
 
@@ -390,7 +470,7 @@ describe('dsh web keyless CLI smoke', () => {
       await new Promise<void>(resolveClose => provider.close(() => { resolveClose() }))
       rmSync(workspace, { recursive: true, force: true })
     }
-  }, 30_000)
+  }, 120_000)
 
   it('DSH_TOOLS_MODE=code collapses the provider wire tools to run_code with the SDK prompt section', async () => {
     requireDist()

packages/client/connection/src/client/fixture.ts

@@ -82,6 +82,7 @@ interface FixtureFollowRequest {
 
 interface FixturePageRequest {
   readonly address: FixtureSessionAddress
+  readonly throughSeq: number
   readonly beforeSeq?: number
   readonly maxMessages?: number
 }
@@ -193,6 +194,7 @@ interface FixtureSessionApi {
   fork(request: { readonly sessionId: SessionId; readonly atSeq?: number }): Promise<ConnectionRpcResult<unknown>>
   history(request: {
     readonly sessionId: SessionId
+    readonly throughSeq?: number
     readonly beforeSeq?: number
     readonly maxMessages?: number
   }): Promise<ConnectionRpcResult<unknown>>
@@ -2698,13 +2700,15 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
     },
     history: async (request) => {
       const log = logs.get(request.sessionId) ?? []
+      const throughSeq = request.throughSeq ?? log.length - 1
+      const boundedLog = log.slice(0, throughSeq + 1)
       // Snapshot at request time, then deliver after the transit delay.
-      const page = pageOf(log, request.beforeSeq, request.maxMessages ?? 50)
+      const page = pageOf(boundedLog, request.beforeSeq, request.maxMessages ?? 50)
       // Tail page carries the projections block (host parallel: one consistent
       // cut over the registered units; asOfSeq = window tail seq, -1 on an
       // empty log — the host's session.seq-1 convention).
       const projections = request.beforeSeq === undefined
-        ? { asOfSeq: log.length - 1, values: projectionValuesOf(log) }
+        ? { asOfSeq: throughSeq, values: projectionValuesOf(boundedLog) }
         : undefined
       const doomed = failNextHistory
       failNextHistory = false
@@ -3520,6 +3524,7 @@ function createFixtureWorld(options: FixtureOptions): FixtureWorld {
             : page.address.childSessionId
           return sessionApi.history({
             sessionId: pageSessionId,
+            throughSeq: page.throughSeq,
             ...page.beforeSeq === undefined ? {} : { beforeSeq: page.beforeSeq },
             ...page.maxMessages === undefined ? {} : { maxMessages: page.maxMessages },
           })