fix(native-chat): bound desktop catch-up to newest page

src/main/ipc/desktop-renderer-runtime-capabilities.ts

@@ -3,6 +3,7 @@ import {
   AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY,
   AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY,
   AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY,
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   AGENT_SESSION_TURN_ITEM_CAPABILITY,
   CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
   PI_STRUCTURED_DIALOGS_RUNTIME_CAPABILITY,
@@ -39,6 +40,8 @@ export const DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES: readonly RuntimeCapab
   AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY,
   AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY,
   AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY,
+  // The renderer settles acknowledged echoes from the read cursor even when a snapshot skips them.
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   AGENT_SESSION_TURN_ITEM_CAPABILITY,
   AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY,
   AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY,

src/main/native-chat/agent-session-wire/agent-session-history-forward-read-budget.test.ts

@@ -8,7 +8,6 @@ import {
   type AgentSessionJournalIdentity
 } from '../../../shared/agent-session-journal-types'
 import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire'
-import * as journalReducer from '../agent-session-journal/journal-reducer'
 import * as rowSchema from '../agent-session-journal/journal-row-schema'
 import {
   createTrackedJournalOpener,
@@ -101,8 +100,7 @@ function observeForwardReads() {
 }
 
 describe('forward history SQL read budget', () => {
-  it('reconnects through every page with one lookahead row per page', async () => {
-    const count = 2_000
+  it.each([200, 2_000, 20_000])('bounds reconnect reads after %i missed rows', async (count) => {
     const journal = await seedJournal(count)
     const { returnedRows, parse } = observeForwardReads()
     const events: AgentSessionSubscribeEvent[] = []
@@ -114,35 +112,16 @@ describe('forward history SQL read budget', () => {
       cursor: { epoch: journal.epoch, sequence: 1 },
       emit: (event) => events.push(event)
     })
-    const batches = events.filter((event) => event.type === 'batch')
-    expect(batches.flatMap((event) => event.batch.items.map((item) => item.itemId))).toEqual(
-      Array.from({ length: count }, (_, index) => `item-${index}`)
+    expect(events).toHaveLength(1)
+    const event = events[0]!
+    expect(event.type).toBe(count > 200 ? 'snapshot' : 'batch')
+    const items =
+      event.type === 'snapshot' ? event.page.items : event.type === 'batch' ? event.batch.items : []
+    expect(items.map((item) => item.itemId)).toEqual(
+      Array.from({ length: 200 }, (_, index) => `item-${count - 200 + index}`)
     )
-    expect(batches.at(-1)?.batch.cursor).toEqual(journal.cursor())
-    // Only the last page says the reader is caught up.
-    expect(batches.map((event) => event.hasNewer === true)).toEqual([
-      ...Array<boolean>(9).fill(true),
-      false
-    ])
-    expect(returnedRows).toEqual([...Array<number>(9).fill(201), 200])
-    expect(parse).toHaveBeenCalledTimes(2_009)
-  })
-
-  it('reduces the timeline once for the whole catch-up, not once per page', async () => {
-    const journal = await seedJournal(2_000)
-    const render = vi.spyOn(journalReducer, 'renderJournalState')
-    const events: AgentSessionSubscribeEvent[] = []
-    new AgentSessionSubscribers().open({
-      id: 'reader',
-      sessionId: identity.sessionId,
-      journal,
-      fence: 1,
-      cursor: { epoch: journal.epoch, sequence: 1 },
-      emit: (event) => events.push(event)
-    })
-    // Catch-up is synchronous, so the reduced timeline cannot change between pages.
-    expect(events.filter((event) => event.type === 'batch')).toHaveLength(10)
-    expect(render).toHaveBeenCalledTimes(1)
+    expect(returnedRows).toEqual([Math.min(count, 201)])
+    expect(parse).toHaveBeenCalledTimes(Math.min(count, 201))
   })
 
   it('keeps an exact final page final and preserves unlimited journal readers', async () => {

src/main/native-chat/agent-session-wire/agent-session-subscriber-catch-up.ts

@@ -103,9 +103,8 @@ export function deliverToSubscriber(
       emitCaughtUp(port, subscriber, emitCheckpoint, shared, { journal, work })
       return
     }
-    // A phone's app ends a stream that outruns its unacked window, so a gap of more than one page
-    // arrives as the newest page instead; older rows load on scroll.
-    if (page.hasNewer && subscriber.boundedFrames) {
+    // Replaying an unbounded gap blocks the host; older rows remain available on scroll.
+    if (page.hasNewer && subscriber.snapshotCatchUp !== false) {
       const tail = withPublishedWork(
         readAgentSessionHydrationPage(journal, subscriber.fence, reader),
         work

src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts

@@ -36,6 +36,8 @@ export type AgentSessionSubscribeInput = {
   cursor?: AgentJournalCursor
   /** Every frame to fit a phone's bridge frame; the RPC sets it for every phone. */
   boundedFrames?: boolean
+  /** False for older desktops whose send echoes require exact row replay. */
+  snapshotCatchUp?: boolean
 }
 
 export type Subscriber = {
@@ -45,6 +47,7 @@ export type Subscriber = {
   cursor: AgentJournalCursor
   fence: number
   boundedFrames: boolean
+  snapshotCatchUp?: boolean
   commands?: AgentSessionSlashCommand[] | null
   /** The last draft list actually SENT — never advanced on a page that withheld
    *  it, or the final replacement would be suppressed by the identity dedup. */
@@ -95,6 +98,7 @@ export class AgentSessionSubscribers {
     emit: AgentSessionSubscriberEmit
     cursor?: AgentJournalCursor
     boundedFrames?: boolean
+    snapshotCatchUp?: boolean
   }): () => void {
     const liveCursor = input.journal.cursor()
     const subscriber: Subscriber = {
@@ -103,7 +107,8 @@ export class AgentSessionSubscribers {
       emit: input.emit,
       cursor: input.cursor ?? { epoch: liveCursor.epoch, sequence: 0 },
       fence: input.fence,
-      boundedFrames: input.boundedFrames === true
+      boundedFrames: input.boundedFrames === true,
+      snapshotCatchUp: input.snapshotCatchUp !== false
     }
     const session = this.bySession.get(input.sessionId) ?? new Map<string, Subscriber>()
     session.set(input.id, subscriber)

src/main/native-chat/agent-session-wire/structured-agent-session-wire-admission.test.ts

@@ -121,7 +121,7 @@ describe('structured agent-session outbound admission', () => {
     dispose()
   })
 
-  it('splits valid-cursor catch-up into admitted batch frames', () => {
+  it('replaces an over-budget catch-up with one admitted newest page', () => {
     const subscribers = new AgentSessionSubscribers()
     const catchup: AgentSessionSubscribeEvent[] = []
     const firstItemSequence = journal.snapshot().items[0]!.sequence
@@ -135,14 +135,11 @@ describe('structured agent-session outbound admission', () => {
       emit: (event) => catchup.push(event)
     })
 
-    expect(catchup.length).toBeGreaterThan(1)
+    expect(catchup).toHaveLength(1)
     catchup.forEach(expectAdmitted)
-    expect(
-      catchup.flatMap((event) => (event.type === 'batch' ? event.batch.items : []))
-    ).toHaveLength(19)
-    expect(catchup.at(-1)).toMatchObject({
-      type: 'batch',
-      batch: { cursor: journal.cursor() },
+    expect(catchup[0]).toMatchObject({
+      type: 'snapshot',
+      page: { hasOlder: true, hasNewer: false, liveCursor: journal.cursor() },
       fence: 2
     })
   })

src/main/runtime/rpc/methods/structured-agent-session-bounded-frames-rpc.test.ts

@@ -3,6 +3,8 @@
 
 import { afterEach, beforeEach, expect, it } from 'vitest'
 import type { AgentSessionSubscribeInput } from '../../../native-chat/agent-session-wire/structured-agent-session-subscribers'
+import { AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY } from '../../../../shared/protocol-version'
+import { DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES } from '../../../ipc/desktop-renderer-runtime-capabilities'
 import {
   call,
   clearStructuredHostStub,
@@ -16,12 +18,21 @@ import {
 beforeEach(installStructuredHostStub)
 afterEach(clearStructuredHostStub)
 
+const currentDesktop = {
+  ...STRUCTURED_CLIENT,
+  clientCapabilities: [
+    ...STRUCTURED_CLIENT.clientCapabilities,
+    AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY
+  ]
+}
+
 it.each([
   // Being a phone is enough: no capability or param is asked of it.
-  ['a phone of any version', STRUCTURED_MOBILE_CLIENT, true],
-  ['a paired desktop', STRUCTURED_CLIENT, false],
-  ['an in-process reader', undefined, false]
-] as const)('subscribes %s with bounded frames: %s', async (_label, client, bounded) => {
+  ['a phone of any version', STRUCTURED_MOBILE_CLIENT, true, true],
+  ['an older paired desktop', STRUCTURED_CLIENT, false, false],
+  ['an updated paired desktop', currentDesktop, false, true],
+  ['an in-process reader', undefined, false, true]
+] as const)('subscribes %s with bounded frames: %s', async (_label, client, bounded, snapshot) => {
   let input: AgentSessionSubscribeInput | undefined
   hostCalls.subscribe.mockImplementation((opened: AgentSessionSubscribeInput) => {
     input = opened
@@ -30,4 +41,22 @@ it.each([
   })
   await call('agentSession.subscribe', { sessionId: SESSION }, client)
   expect(input?.boundedFrames === true).toBe(bounded)
+  expect(input?.snapshotCatchUp).toBe(snapshot)
+})
+
+it('allows the local desktop renderer to replace a missed gap with a snapshot', async () => {
+  let input: AgentSessionSubscribeInput | undefined
+  hostCalls.subscribe.mockImplementation((opened: AgentSessionSubscribeInput) => {
+    input = opened
+    opened.emit({ type: 'end' })
+    return () => {}
+  })
+  const cursor = { epoch: 'epoch-a', sequence: 10 }
+  await call('agentSession.subscribe', { sessionId: SESSION, cursor }, {
+    clientId: 'desktop-renderer',
+    clientKind: 'runtime',
+    clientCapabilities: [...DESKTOP_RENDERER_RUNTIME_CLIENT_CAPABILITIES]
+  })
+  expect(input?.cursor).toEqual(cursor)
+  expect(input?.snapshotCatchUp).toBe(true)
 })

src/main/runtime/rpc/methods/structured-agent-session.ts

@@ -58,6 +58,7 @@ import {
   UnsubscribeParams
 } from './structured-agent-session-schemas'
 import { sendStructuredAgentSessionForClient } from './structured-agent-session-send-compatibility'
+import { AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY } from '../../../../shared/protocol-version'
 
 export const STRUCTURED_AGENT_SESSION_METHODS = [
   defineMethod({
@@ -182,6 +183,11 @@ export const STRUCTURED_AGENT_SESSION_METHODS = [
             )
           ),
         ...(params.cursor ? { cursor: params.cursor } : {}),
+        // Older desktops still need each missed submission to settle their local echoes.
+        snapshotCatchUp:
+          ctx.clientKind === undefined ||
+          ctx.clientKind === 'mobile' ||
+          ctx.clientCapabilities?.includes(AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY) === true,
         // Every phone's app, whatever its version, ends a stream over a frame its bridge cannot carry.
         ...(ctx.clientKind === 'mobile' ? { boundedFrames: true } : {})
       })

src/renderer/src/web/web-runtime-client-capabilities.ts

@@ -3,6 +3,7 @@ import {
   AGENT_SESSION_BOUNDARY_RUNTIME_CAPABILITY,
   REPO_SEARCH_QUALIFIED_REFS_RUNTIME_CAPABILITY,
   AGENT_SESSION_TURN_ITEM_CAPABILITY,
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY,
   SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY,
   SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY,
@@ -18,6 +19,7 @@ export const WEB_RUNTIME_CLIENT_CAPABILITIES = [
   AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY,
   AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY,
   AGENT_SESSION_TURN_ITEM_CAPABILITY,
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY,
   SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY,
   SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY,

src/renderer/src/web/web-runtime-client.test.ts

@@ -17,6 +17,7 @@ import type { RuntimeRpcResponse } from '../../../shared/runtime-rpc-envelope'
 import {
   AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY,
   AGENT_SESSION_TURN_ITEM_CAPABILITY,
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   AGENT_SESSION_BOUNDARY_RUNTIME_CAPABILITY,
   REPO_SEARCH_QUALIFIED_REFS_RUNTIME_CAPABILITY,
   SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY,
@@ -128,6 +129,7 @@ describe('WebRuntimeClient', () => {
         AGENT_SESSION_BACKGROUND_TASK_ROW_STOP_CAPABILITY,
         AGENT_SESSION_BACKGROUND_TASK_CHILD_VIEWS_CAPABILITY,
         AGENT_SESSION_TURN_ITEM_CAPABILITY,
+        AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
         SESSION_TAB_CLOSE_INTENT_RUNTIME_CAPABILITY,
         SESSION_TABS_SPLIT_GROUP_PLACEMENT_RUNTIME_CAPABILITY,
         SESSION_TABS_RETIREMENT_PROOF_DELTA_RUNTIME_CAPABILITY,

src/shared/electron-remote-runtime-client-capabilities.ts

@@ -1,6 +1,7 @@
 import {
   AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY,
   AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY,
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   BROWSER_CLIENT_HOST_RUNTIME_CAPABILITY,
   BROWSER_CLIENT_PAGE_METADATA_RUNTIME_CAPABILITY,
   CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
@@ -22,6 +23,7 @@ export const ELECTRON_REMOTE_RUNTIME_CLIENT_CAPABILITIES = [
   AGENT_SESSION_CREATE_MESSAGE_RUNTIME_CAPABILITY,
   AGENT_SESSION_PENDING_SEND_RESULT_RUNTIME_CAPABILITY,
   AGENT_SESSION_ACCEPTED_SEND_RUNTIME_CAPABILITY,
+  AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY,
   // The renderer that reads a paired host's structured chats is the one that reads its own.
   STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,
   CLAUDE_STRUCTURED_AGENT_SESSION_RUNTIME_CAPABILITY,

src/shared/runtime-capability-constants.ts

@@ -183,6 +183,9 @@ export const AGENT_SESSION_REWIND_RECOVERY_CAPABILITY = 'agent-session.rewind-re
 // status form to clients that do not advertise this. Transitional: drop the downgrade once no
 // supported release lacks the capability.
 export const AGENT_SESSION_TURN_ITEM_CAPABILITY = 'agent-session.turn-item.v1' as const
+/** The client retires acknowledged echoes even when a catch-up snapshot skips their rows. */
+export const AGENT_SESSION_SNAPSHOT_CATCH_UP_CLIENT_CAPABILITY =
+  'agent-session.snapshot-catch-up.v1' as const
 export const AGENT_SESSION_BACKGROUND_TASK_STOP_CAPABILITY =
   'agent-session.background-task-stop.v1' as const
 // Why: agentSession.respondToQuestion has a strict schema, so clients must not send structured

tests/e2e/cross-version-wire/cross-version-stop-note-scenarios.ts

@@ -111,7 +111,7 @@ export function describeReleasedStopNoteProjection(port: ScenarioPort): void {
         expect(render(state)).toEqual({ type: 'text', text: ordinary.text })
         expect(rig.journal.cursor()).toEqual(held)
       })
-      it('catches up through byte-bounded pages without losing the dependent note or cursor', async () => {
+      it('replays to a released desktop and pages a fresh snapshot back to the dependent note', async () => {
         const before = await seed()
         const held = rig.journal.cursor()
         for (const clientMessageId of ['large-a', 'large-b']) {
@@ -134,10 +134,22 @@ export function describeReleasedStopNoteProjection(port: ScenarioPort): void {
         const first = await page({ direction: 'after', cursor: held, limit: 3 })
         expect(first.hasNewer).toBe(true)
         const caughtUp = await subscribe(held)
-        const after = reduce(before, caughtUp)
+        expect(events(caughtUp).every((event) => event.type === 'batch')).toBe(true)
+        expect(reduce(before, caughtUp).cursor).toEqual(rig.journal.cursor())
+        const fresh = await subscribe(null)
+        let after = reduce(before, fresh)
+        expect(events(fresh)[0]).toMatchObject({ type: 'snapshot', page: { hasOlder: true } })
+        while (after.hasOlder && !after.items.some((item) => item.itemId === noteId)) {
+          const requestedCursor = { epoch: after.epoch!, sequence: after.items[0]!.sequence }
+          const older = await page({ direction: 'before', cursor: requestedCursor })
+          after = rig.client.reduceStructuredAgentSession(after, {
+            type: 'older-page',
+            requestedCursor,
+            page: older
+          })
+        }
         expect(render(after)).toEqual({ type: 'text', text: ordinary.text })
         expect(after.cursor).toEqual(rig.journal.cursor())
-        expect(events(caughtUp).filter((event) => event.type === 'batch').length).toBeGreaterThan(1)
       })
       it.each(['running', 'completed', 'unverifiable', 'missing', 'other-failure'] as const)(
         'keeps %s wording through the released page reader',

tests/e2e/cross-version-wire/cross-version-stop-note-test-rig.ts

@@ -14,6 +14,7 @@ import { agentSessionFailureFact } from '../../../src/shared/agent-session-failu
 import { agentSessionFailureWords } from '../../../src/shared/agent-session-failure-words'
 import {
   AGENT_JOURNAL_THREAD_SCOPE,
+  type AgentJournalCursor,
   type AgentJournalItemBody,
   type AgentJournalTurnLifecycle
 } from '../../../src/shared/agent-session-journal-types'
@@ -138,8 +139,8 @@ export function createReleasedStopNoteRig(port: ScenarioPort, ref: string) {
       items: result.page.items.map((item) => client.AgentJournalRenderItemSchema.parse(item))
     }
   }
-  async function subscribe(cursor = journal.cursor()): Promise<RpcReply[]> {
-    return call('agentSession.subscribe', { sessionId: SESSION, cursor })
+  async function subscribe(cursor: AgentJournalCursor | null = journal.cursor()): Promise<RpcReply[]> {
+    return call('agentSession.subscribe', { sessionId: SESSION, ...(cursor ? { cursor } : {}) })
   }
   function events(replies: RpcReply[]): AgentSessionSubscribeEvent[] {
     return replies.map((reply) => {

tests/e2e/structured-agent-session-read-owner.unit.test.ts

@@ -122,7 +122,7 @@ describe('structured session cursor/body regression', () => {
   })
 
   it.each([40, 401])(
-    'replays an off-page dispatch after %i missed rows without stranding pending state',
+    'catches up an off-page dispatch after %i missed rows without stranding pending state',
     async (missedRows) => {
       const { journal, appendOutput, accept } = await fixture()
       let hostSummary: AgentSessionStatusSummary | undefined
@@ -234,12 +234,18 @@ describe('structured session cursor/body regression', () => {
       expect(
         projectStructuredAgentSessionStatus(settled.items, settled.submissions, settled.fence)
       ).toBe(hostSummary?.status)
-      expect(settled.submissions).toEqual(journal.snapshot().submissions)
+      expect(settled.submissions).toEqual(
+        missedRows > 200 ? tail.submissions : journal.snapshot().submissions
+      )
       expect({
         cursor: caughtUp.cursor?.sequence,
         dispatch: caughtUp.submissions[0]?.dispatchState,
         unansweredDispatch: hasUnansweredStructuredAgentSessionDispatch(caughtUp.submissions, 1)
-      }).toEqual({ cursor: 100 + missedRows, dispatch: 'accepted', unansweredDispatch: false })
+      }).toEqual({
+        cursor: 100 + missedRows,
+        dispatch: missedRows > 200 ? undefined : 'accepted',
+        unansweredDispatch: false
+      })
       expect(mocks.call).toHaveBeenCalledTimes(1)
       expect(mocks.subscribe).toHaveBeenCalledTimes(2)
       expect(mocks.subscribe.mock.calls[1]?.[1]).toEqual({