fix(native-chat): stop a bounded tail read from moving the chat cursor past unapplied rows (#20581)

* fix(native-chat): stop a bounded tail read from moving the chat cursor past unapplied rows

A structured chat pane could latch "Working for N" forever after the agent had
finished, showing the send arrow rather than Stop, while the sidebar and
`worktree ps` correctly read idle.

The client replica has one position (`state.cursor`) and one body. Two
operations keep those consistent: replace (both from one host snapshot) and
append (rows contiguous with the cursor). The `tail-page` branch was a third
thing: it took the cursor from the journal head, the items from a bounded page
(200 items, byte-capped), then merged retained client submissions over the
page's. Under continuous journal writes the client is always slightly behind,
so the branch ran on every window focus and on every pane re-activation. When
more than a page of rows had landed since a send, that send's user item fell
off the page, its submission was not carried, the retained `pending` survived,
and the cursor jumped past the dispatch-acceptance row. Nothing re-sends it: a
batch carries only touched items and that submission is never touched again.

Delete the third operation rather than guard it. A live subscription is now the
only thing that moves the cursor, and `subscribe({ cursor })` already replays
exactly the missed rows.

- remove the window `focus` listener and the owner/transport `refresh` contract
- skip warm hydration: a retained owner subscribes at its applied cursor
- cold hydration keeps its history read, applied as the existing `snapshot`
  (replace) event rather than `tail-page`
- delete the `tail-page` action and its reducer branch
- delete `resumeCursor` and `shouldAdvanceStructuredResumeCursor`; two cursors
  with two advancement rules were how position and body drifted apart

`older-page`/`loadOlder`, the unattached-refusal grace, generation guards and
the coalescer are unchanged. No host, wire or schema change.

Also fixes a second cost of the same branch: focus during a busy turn discarded
paged-in older items, shrinking the transcript to one bounded page mid-turn.

* fix(native-chat): preserve unavailable mixed-version session fences

src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts

@@ -29,7 +29,6 @@ export type StructuredAgentSessionReadOwner = {
   dispose: () => void
   getSnapshot: () => StructuredAgentSessionReadSnapshot
   loadOlder: () => Promise<void>
-  refresh: () => void
   subscribe: (listener: () => void) => () => void
 }
 
@@ -57,7 +56,6 @@ function createReadOwner(
     loadingOlder: false
   }
   let stopActiveRun: (() => void) | null = null
-  let refreshActiveRun = (): void => {}
   const retiredHistoryRead = (): boolean => true
   let captureActiveHistoryReadGuard = (): (() => boolean) => retiredHistoryRead
   const activations = new Set<symbol>()
@@ -91,7 +89,7 @@ function createReadOwner(
       setSnapshot({ ...snapshot, loadingOlder: false })
     }
   }
-  const refreshTail = async (shouldStop: () => boolean): Promise<void> => {
+  const hydrate = async (shouldStop: () => boolean): Promise<void> => {
     const result = await callStructuredAgentSession<AgentSessionHistoryResult>(
       target,
       'agentSession.history',
@@ -120,7 +118,7 @@ function createReadOwner(
     if (shouldStop()) {
       return
     }
-    apply({ type: 'tail-page', page: result.page })
+    apply({ type: 'history-page', page: result.page })
     if (shouldStop()) {
       return
     }
@@ -177,15 +175,13 @@ function createReadOwner(
       applyError: (message) => apply({ type: 'error', message }),
       getCursor: () => snapshot.state.cursor,
       onHistoryReadInvalidated: clearLoadingOlder,
-      refreshTail,
+      hydrate: snapshot.state.epoch === null ? hydrate : undefined,
       sessionId,
       target
     })
     captureActiveHistoryReadGuard = transport.captureHistoryReadGuard
-    refreshActiveRun = transport.refresh
     stopActiveRun = () => {
       captureActiveHistoryReadGuard = () => retiredHistoryRead
-      refreshActiveRun = (): void => {}
       transport.dispose()
       stopActiveRun = null
     }
@@ -266,7 +262,6 @@ function createReadOwner(
         }
       }
     },
-    refresh: () => refreshActiveRun(),
     subscribe: (listener) => {
       listeners.add(listener)
       return () => {

src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts

@@ -1,4 +1,8 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
+import {
+  EMPTY_STRUCTURED_AGENT_SESSION,
+  reduceStructuredAgentSession
+} from '../../../../shared/structured-agent-session-reducer'
 import type { AgentJournalCursor } from '../../../../shared/agent-session-journal-types'
 import type {
   AgentSessionHistoryPage,
@@ -70,12 +74,56 @@ describe('structured agent-session read transport generations', () => {
       applyError,
       getCursor: () => null,
       onHistoryReadInvalidated: () => undefined,
-      refreshTail: async () => undefined,
+      hydrate: async () => undefined,
       sessionId: 'session-a',
       target
     })
   }
 
+  it('flushes queued rows before reading the applied cursor for reconnect', async () => {
+    vi.useFakeTimers()
+    try {
+      let state = EMPTY_STRUCTURED_AGENT_SESSION
+      const transport = startStructuredAgentSessionReadTransport({
+        applyEvent: (event) => {
+          state = reduceStructuredAgentSession(state, { type: 'event', event })
+        },
+        applyError: vi.fn(),
+        getCursor: () => state.cursor,
+        onHistoryReadInvalidated: () => undefined,
+        sessionId: 'session-a',
+        target
+      })
+      attempts[0].onEvent(snapshot(100))
+      attempts[0].closed.resolve({ unsubscribe: attempts[0].unsubscribe })
+      await flushPromises()
+      attempts[0].onClose()
+      await vi.advanceTimersByTimeAsync(720)
+      attempts[0].onEvent({
+        type: 'batch',
+        sessionId: 'session-a',
+        batch: {
+          cursor: { epoch: 'epoch-a', sequence: 101 },
+          items: [],
+          removedItemIds: [],
+          submissions: []
+        }
+      })
+      expect(state.cursor?.sequence).toBe(100)
+      await vi.advanceTimersByTimeAsync(30)
+      expect(state.cursor?.sequence).toBe(101)
+      expect(mocks.subscribe.mock.calls[1]?.[1]).toEqual({
+        sessionId: 'session-a',
+        cursor: { epoch: 'epoch-a', sequence: 101 }
+      })
+      attempts[1].closed.resolve({ unsubscribe: attempts[1].unsubscribe })
+      await flushPromises()
+      transport.dispose()
+    } finally {
+      vi.useRealTimers()
+    }
+  })
+
   it('ignores opening frames after disposal and a replacement transport starts', async () => {
     const applyEvent = vi.fn()
     const applyError = vi.fn()
@@ -159,8 +207,8 @@ describe('structured agent-session read transport unattached refusals', () => {
     })
   })
 
-  function startWithTail(
-    refreshTail: () => Promise<void>,
+  function startWithHydration(
+    hydrate: () => Promise<void>,
     applyError: (message: string) => void,
     applyEvent = vi.fn()
   ) {
@@ -169,7 +217,7 @@ describe('structured agent-session read transport unattached refusals', () => {
       applyError,
       getCursor: () => null,
       onHistoryReadInvalidated: () => undefined,
-      refreshTail,
+      hydrate,
       sessionId: 'session-a',
       target
     })
@@ -186,7 +234,7 @@ describe('structured agent-session read transport unattached refusals', () => {
     vi.useFakeTimers()
     try {
       const applyError = vi.fn()
-      const transport = startWithTail(async () => {
+      const transport = startWithHydration(async () => {
         throw rpcRefusal(UNATTACHED)
       }, applyError)
       await flushPromises()
@@ -204,7 +252,7 @@ describe('structured agent-session read transport unattached refusals', () => {
     vi.useFakeTimers()
     try {
       const applyError = vi.fn()
-      const transport = startWithTail(async () => {
+      const transport = startWithHydration(async () => {
         throw rpcRefusal(UNATTACHED)
       }, applyError)
       await flushPromises()
@@ -232,7 +280,7 @@ describe('structured agent-session read transport unattached refusals', () => {
     vi.useFakeTimers()
     try {
       const applyError = vi.fn()
-      const transport = startWithTail(async () => {
+      const transport = startWithHydration(async () => {
         throw new Error('journal read failed')
       }, applyError)
       await flushPromises()
@@ -247,7 +295,7 @@ describe('structured agent-session read transport unattached refusals', () => {
     vi.useFakeTimers()
     try {
       const applyError = vi.fn()
-      const transport = startWithTail(async () => undefined, applyError)
+      const transport = startWithHydration(async () => undefined, applyError)
       await flushPromises()
       expect(attempts).toHaveLength(1)
 
@@ -267,7 +315,7 @@ describe('structured agent-session read transport unattached refusals', () => {
     try {
       const applyError = vi.fn()
       const applyEvent = vi.fn()
-      const transport = startWithTail(async () => undefined, applyError, applyEvent)
+      const transport = startWithHydration(async () => undefined, applyError, applyEvent)
       await flushPromises()
       expect(attempts).toHaveLength(1)
 

src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts

@@ -5,7 +5,6 @@ import {
   AGENT_SESSION_UNATTACHED_READ_GRACE_MS,
   isUnattachedAgentSessionReadRefusal
 } from '../../../../shared/structured-agent-session-read-refusal'
-import { shouldAdvanceStructuredResumeCursor } from '../../../../shared/structured-agent-session-reducer'
 import type { RuntimeClientTarget } from '@/runtime/runtime-rpc-client'
 import { subscribeStructuredAgentSession } from '@/runtime/structured-agent-session-client'
 
@@ -37,13 +36,12 @@ export function startStructuredAgentSessionReadTransport(args: {
   applyError: (message: string) => void
   getCursor: () => AgentJournalCursor | null
   onHistoryReadInvalidated: () => void
-  refreshTail: (shouldStop: () => boolean) => Promise<void>
+  hydrate?: (shouldStop: () => boolean) => Promise<void>
   sessionId: string
   target: RuntimeClientTarget
 }): {
   captureHistoryReadGuard: () => () => boolean
   dispose: () => void
-  refresh: () => void
 } {
   let stopped = false
   let connected = false
@@ -52,7 +50,6 @@ export function startStructuredAgentSessionReadTransport(args: {
   let openGeneration = 0
   let stateGeneration = 0
   let unsubscribe = (): void => {}
-  let resumeCursor = args.getCursor()
   let shouldStopCoalescedEvent = (): boolean => true
   const coalescer = createStructuredAgentSessionEventCoalescer((event) => {
     if (!shouldStopCoalescedEvent()) {
@@ -112,12 +109,6 @@ export function startStructuredAgentSessionReadTransport(args: {
       if (!isCurrentOpenGeneration(eventOpenGeneration)) {
         return
       }
-      resumeCursor = event.page.liveCursor ?? event.page.window.nextCursor
-    } else if (
-      event.type === 'batch' &&
-      shouldAdvanceStructuredResumeCursor(resumeCursor, event.batch.cursor)
-    ) {
-      resumeCursor = event.batch.cursor
     } else if (event.type === 'end') {
       connected = false
       reconnectScheduler.schedule()
@@ -148,9 +139,10 @@ export function startStructuredAgentSessionReadTransport(args: {
         return
       }
       let closedDuringOpen = false
+      const cursor = args.getCursor()
       const handle = await subscribeStructuredAgentSession(
         args.target,
-        { sessionId: args.sessionId, ...(resumeCursor ? { cursor: resumeCursor } : {}) },
+        { sessionId: args.sessionId, ...(cursor ? { cursor } : {}) },
         (event) => handleEvent(event, currentOpenGeneration),
         (error) => {
           if (!isCurrentOpenGeneration(currentOpenGeneration)) {
@@ -192,43 +184,26 @@ export function startStructuredAgentSessionReadTransport(args: {
       }
     }
   }
-  const refresh = (): void => {
-    const shouldStop = captureHistoryReadGuard()
+  if (args.hydrate) {
+    const shouldStopInitialRead = captureHistoryReadGuard()
     void args
-      .refreshTail(shouldStop)
+      .hydrate(shouldStopInitialRead)
       .then(() => {
-        if (shouldStop()) {
+        if (shouldStopInitialRead()) {
           return
         }
         clearUnattachedReadGrace()
-        resumeCursor = args.getCursor()
-        if (!connected) {
-          reconnectScheduler.schedule(0)
-        }
+        return open()
       })
       .catch((error) => {
-        if (!shouldStop()) {
+        if (!shouldStopInitialRead()) {
           reportReadFailure(error)
+          reconnectScheduler.schedule()
         }
       })
+  } else {
+    void open()
   }
-  const shouldStopInitialRead = captureHistoryReadGuard()
-  void args
-    .refreshTail(shouldStopInitialRead)
-    .then(() => {
-      if (shouldStopInitialRead()) {
-        return
-      }
-      clearUnattachedReadGrace()
-      resumeCursor = args.getCursor()
-      return open()
-    })
-    .catch((error) => {
-      if (!shouldStopInitialRead()) {
-        reportReadFailure(error)
-        reconnectScheduler.schedule()
-      }
-    })
   return {
     captureHistoryReadGuard,
     dispose: () => {
@@ -238,7 +213,6 @@ export function startStructuredAgentSessionReadTransport(args: {
       reconnectScheduler.dispose()
       coalescer.dispose()
       unsubscribe()
-    },
-    refresh
+    }
   }
 }

src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx

@@ -8,8 +8,7 @@ import type {
 } from '../../../../shared/agent-session-journal-types'
 import {
   AGENT_SESSION_HISTORY_MAX_LIMIT,
-  type AgentSessionHistoryPage,
-  type AgentSessionSubscribeEvent
+  type AgentSessionHistoryPage
 } from '../../../../shared/agent-session-wire'
 
 const mocks = vi.hoisted(() => ({ call: vi.fn(), subscribe: vi.fn() }))
@@ -124,6 +123,17 @@ describe('useStructuredAgentSessionRead history window', () => {
     })
   })
 
+  it('does not invent a writable fence for a mixed-version history page', async () => {
+    mocks.call.mockResolvedValueOnce({ ok: true, page: page('tail', [], false) })
+
+    const { result } = renderHook(() =>
+      useStructuredAgentSessionRead({ sessionId: 'session-a', target: LOCAL_TARGET })
+    )
+
+    await waitFor(() => expect(result.current.state.status).toBe('ready'))
+    expect(result.current.state.fence).toBeNull()
+  })
+
   it('loads each earlier page at the wire maximum', async () => {
     const tailItems = Array.from({ length: 200 }, (_, index) =>
       message(`tail-${index}`, 301 + index, 'assistant')
@@ -162,7 +172,7 @@ describe('useStructuredAgentSessionRead history window', () => {
     expect(result.current.state.items[0]?.itemId).toBe('oldest')
   })
 
-  it('refreshes only visible structured sessions when the app regains focus', async () => {
+  it('does no host work when the app regains focus', async () => {
     const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true)
     mocks.call.mockResolvedValue({ ok: true, page: page('tail', [], false) })
     const visible = renderHook(() =>
@@ -182,154 +192,15 @@ describe('useStructuredAgentSessionRead history window', () => {
     await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(1))
     expect(mocks.subscribe).toHaveBeenCalledTimes(1)
 
-    act(() => window.dispatchEvent(new Event('focus')))
+    await act(async () => window.dispatchEvent(new Event('focus')))
 
-    await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2))
-    expect(mocks.call).toHaveBeenLastCalledWith(LOCAL_TARGET, 'agentSession.history', {
-      sessionId: 'session-visible',
-      direction: 'tail',
-      limit: AGENT_SESSION_HISTORY_MAX_LIMIT
-    })
+    expect(mocks.call).toHaveBeenCalledTimes(1)
+    expect(mocks.subscribe).toHaveBeenCalledTimes(1)
     visible.unmount()
     hidden.unmount()
     hasFocus.mockRestore()
   })
 
-  it('drops a delayed refresh after reconnect without mutating state or provider session', async () => {
-    const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true)
-    const delayedRefresh = Promise.withResolvers<{
-      ok: true
-      page: AgentSessionHistoryPage
-      providerSession: { key: 'session_id'; id: string }
-    }>()
-    const closes: (() => void)[] = []
-    const initialProviderSession = { key: 'session_id', id: 'provider-initial' } as const
-    mocks.call
-      .mockResolvedValueOnce({
-        ok: true,
-        page: page('tail', [message('initial', 1, 'assistant')], false),
-        providerSession: initialProviderSession
-      })
-      .mockReturnValueOnce(delayedRefresh.promise)
-    mocks.subscribe.mockImplementation((_target, _params, _onEvent, _onError, onClose) => {
-      closes.push(onClose)
-      return Promise.resolve({ unsubscribe: vi.fn() })
-    })
-
-    const view = renderHook(() =>
-      useStructuredAgentSessionRead({ sessionId: 'session-a', target: LOCAL_TARGET })
-    )
-
-    try {
-      await waitFor(() => expect(mocks.subscribe).toHaveBeenCalledOnce())
-      expect(view.result.current.state.items[0]?.itemId).toBe('initial')
-      expect(view.result.current.providerSession).toBe(initialProviderSession)
-      const stateBeforeRefresh = view.result.current.state
-
-      act(() => window.dispatchEvent(new Event('focus')))
-      await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2))
-
-      vi.useFakeTimers()
-      act(() => closes[0]?.())
-      await act(async () => vi.advanceTimersByTimeAsync(750))
-      expect(mocks.subscribe).toHaveBeenCalledTimes(2)
-
-      await act(async () => {
-        delayedRefresh.resolve({
-          ok: true,
-          page: page('tail', [message('stale', 2, 'assistant')], false),
-          providerSession: { key: 'session_id', id: 'provider-stale' }
-        })
-        await delayedRefresh.promise
-        await Promise.resolve()
-      })
-
-      expect(view.result.current.state).toBe(stateBeforeRefresh)
-      expect(view.result.current.state.items[0]?.itemId).toBe('initial')
-      expect(view.result.current.providerSession).toBe(initialProviderSession)
-    } finally {
-      vi.useRealTimers()
-      view.unmount()
-      hasFocus.mockRestore()
-    }
-  })
-
-  it.each(['snapshot', 'reset'] as const)(
-    'drops a delayed refresh after a same-stream %s advances the epoch',
-    async (eventType) => {
-      const hasFocus = vi.spyOn(document, 'hasFocus').mockReturnValue(true)
-      const delayedRefresh = Promise.withResolvers<{
-        ok: true
-        page: AgentSessionHistoryPage
-        providerSession: { key: 'session_id'; id: string }
-      }>()
-      const onEvents: ((event: AgentSessionSubscribeEvent) => void)[] = []
-      const initialProviderSession = { key: 'session_id', id: 'provider-initial' } as const
-      mocks.call
-        .mockResolvedValueOnce({
-          ok: true,
-          page: page('tail', [message('initial', 1, 'assistant')], false),
-          providerSession: initialProviderSession
-        })
-        .mockReturnValueOnce(delayedRefresh.promise)
-      mocks.subscribe.mockImplementation((_target, _params, onEvent) => {
-        onEvents.push(onEvent)
-        return Promise.resolve({ unsubscribe: vi.fn() })
-      })
-
-      const view = renderHook(() =>
-        useStructuredAgentSessionRead({ sessionId: 'session-a', target: LOCAL_TARGET })
-      )
-
-      try {
-        await waitFor(() => expect(onEvents).toHaveLength(1))
-        act(() => window.dispatchEvent(new Event('focus')))
-        await waitFor(() => expect(mocks.call).toHaveBeenCalledTimes(2))
-
-        const replacementPage = page(
-          'tail',
-          [message('new-epoch', 2, 'assistant')],
-          false,
-          'epoch-b'
-        )
-        const replacementEvent: AgentSessionSubscribeEvent =
-          eventType === 'reset'
-            ? {
-                type: 'reset',
-                sessionId: 'session-a',
-                reset: 'epoch_changed',
-                page: replacementPage,
-                fence: 2
-              }
-            : { type: 'snapshot', sessionId: 'session-a', page: replacementPage, fence: 2 }
-        act(() => onEvents[0]?.(replacementEvent))
-
-        expect(view.result.current.state.epoch).toBe('epoch-b')
-        expect(view.result.current.state.items[0]?.itemId).toBe('new-epoch')
-        expect(view.result.current.providerSession).toBe(initialProviderSession)
-        const stateAfterReplacement = view.result.current.state
-
-        await act(async () => {
-          delayedRefresh.resolve({
-            ok: true,
-            page: page('tail', [message('stale-refresh', 3, 'assistant')], false),
-            providerSession: { key: 'session_id', id: 'provider-stale' }
-          })
-          await delayedRefresh.promise
-          await Promise.resolve()
-        })
-
-        expect(view.result.current.state).toBe(stateAfterReplacement)
-        expect(view.result.current.state.epoch).toBe('epoch-b')
-        expect(view.result.current.state.items[0]?.itemId).toBe('new-epoch')
-        expect(view.result.current.providerSession).toBe(initialProviderSession)
-      } finally {
-        view.unmount()
-        hasFocus.mockRestore()
-      }
-    }
-  )
-
   it('does no host work for retained inactive sessions', async () => {
     const first = renderHook(() =>
       useStructuredAgentSessionRead({
@@ -354,7 +225,7 @@ describe('useStructuredAgentSessionRead history window', () => {
     second.unmount()
   })
 
-  it('preserves cached state while switching away and refreshes once on re-entry', async () => {
+  it('preserves cached state and resumes at the applied cursor on re-entry', async () => {
     const unsubscribe = vi.fn()
     mocks.call.mockImplementation((_target, _method, params) => {
       const sessionId = (params as { sessionId: string }).sessionId
@@ -395,7 +266,11 @@ describe('useStructuredAgentSessionRead history window', () => {
     view.rerender({ active: 'first' })
     expect(view.result.current.first.state.items[0]?.itemId).toBe('session-switch-a-message')
     await waitFor(() => expect(mocks.subscribe).toHaveBeenCalledTimes(3))
-    expect(mocks.call).toHaveBeenCalledTimes(3)
+    expect(mocks.call).toHaveBeenCalledTimes(2)
+    expect(mocks.subscribe.mock.calls[2]?.[1]).toEqual({
+      sessionId: 'session-switch-a',
+      cursor: view.result.current.first.state.cursor
+    })
     expect(unsubscribe).toHaveBeenCalledTimes(2)
   })
 })

src/renderer/src/components/native-chat/use-structured-agent-session-read.ts

@@ -30,19 +30,6 @@ export function useStructuredAgentSessionRead(args: {
 
   useEffect(() => (isVisible ? owner.activate() : undefined), [isVisible, owner])
 
-  useEffect(() => {
-    if (!isVisible) {
-      return
-    }
-    const refresh = (): void => {
-      if (document.hasFocus()) {
-        owner.refresh()
-      }
-    }
-    window.addEventListener('focus', refresh)
-    return () => window.removeEventListener('focus', refresh)
-  }, [isVisible, owner])
-
   return {
     state: snapshot.state,
     loadingOlder: snapshot.loadingOlder,

src/shared/structured-agent-session-reducer.test.ts

@@ -144,186 +144,6 @@ describe('structured agent session reducer', () => {
     expect(restored.hasOlder).toBe(false)
   })
 
-  it('does not let a stale focus refresh replace newer streamed state', () => {
-    const streamed = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-      type: 'event',
-      event: {
-        type: 'snapshot',
-        sessionId: 'session-a',
-        fence: 1,
-        page: hydrationPage([item('streamed', 50)])
-      }
-    })
-    const afterRefresh = reduceStructuredAgentSession(streamed, {
-      type: 'tail-page',
-      page: {
-        sessionId: 'session-a',
-        epoch: 'epoch-a',
-        direction: 'tail',
-        items: [item('stale', 40)],
-        removedItemIds: [],
-        submissions: [],
-        window: {
-          oldest: { epoch: 'epoch-a', sequence: 40 },
-          newest: { epoch: 'epoch-a', sequence: 40 },
-          nextCursor: { epoch: 'epoch-a', sequence: 40 }
-        },
-        liveCursor: { epoch: 'epoch-a', sequence: 40 },
-        hasOlder: true,
-        hasNewer: false
-      }
-    })
-
-    expect(afterRefresh).toBe(streamed)
-  })
-
-  it('keeps paged-in older items when a focus refresh carries nothing new', () => {
-    const snapshot = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-      type: 'event',
-      event: {
-        type: 'snapshot',
-        sessionId: 'session-a',
-        fence: 1,
-        page: hydrationPage([item('newest', 50)])
-      }
-    })
-    const withOlder = reduceStructuredAgentSession(snapshot, {
-      type: 'older-page',
-      requestedCursor: { epoch: 'epoch-a', sequence: 50 },
-      page: {
-        sessionId: 'session-a',
-        epoch: 'epoch-a',
-        direction: 'before',
-        items: [item('older', 10)],
-        removedItemIds: [],
-        submissions: [],
-        window: {
-          oldest: { epoch: 'epoch-a', sequence: 10 },
-          newest: { epoch: 'epoch-a', sequence: 10 },
-          nextCursor: { epoch: 'epoch-a', sequence: 10 }
-        },
-        hasOlder: false,
-        hasNewer: true
-      }
-    })
-    const afterRefresh = reduceStructuredAgentSession(withOlder, {
-      type: 'tail-page',
-      page: {
-        sessionId: 'session-a',
-        epoch: 'epoch-a',
-        direction: 'tail',
-        items: [item('newest', 50)],
-        removedItemIds: [],
-        submissions: [],
-        window: {
-          oldest: { epoch: 'epoch-a', sequence: 50 },
-          newest: { epoch: 'epoch-a', sequence: 50 },
-          nextCursor: { epoch: 'epoch-a', sequence: 50 }
-        },
-        liveCursor: { epoch: 'epoch-a', sequence: 50 },
-        hasOlder: true,
-        hasNewer: false
-      }
-    })
-
-    expect(afterRefresh).toBe(withOlder)
-    expect(afterRefresh.items.map((entry) => entry.itemId)).toEqual(['older', 'newest'])
-  })
-
-  it('accepts a newer fence from an equal-cursor tail refresh', () => {
-    const initial = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-      type: 'event',
-      event: {
-        type: 'snapshot',
-        sessionId: 'session-a',
-        fence: 1,
-        page: hydrationPage([item('newest', 50)])
-      }
-    })
-    const page = { ...hydrationPage([item('newest', 50)]), fence: 2 }
-
-    const refreshed = reduceStructuredAgentSession(initial, { type: 'tail-page', page })
-
-    expect(refreshed.fence).toBe(2)
-    expect(refreshed.items).toBe(initial.items)
-  })
-
-  it('keeps rapid-send submissions when a newer tail refresh contains only the last one', () => {
-    const initial = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-      type: 'event',
-      event: {
-        type: 'snapshot',
-        sessionId: 'session-a',
-        fence: 1,
-        page: hydrationPage(
-          [item('first', 10)],
-          Array.from({ length: 8 }, (_, index) => submission(index))
-        )
-      }
-    })
-    const refreshed = reduceStructuredAgentSession(initial, {
-      type: 'tail-page',
-      page: {
-        sessionId: 'session-a',
-        epoch: 'epoch-a',
-        direction: 'tail',
-        items: [item('latest', 11)],
-        removedItemIds: [],
-        submissions: [submission(7)],
-        window: {
-          oldest: { epoch: 'epoch-a', sequence: 11 },
-          newest: { epoch: 'epoch-a', sequence: 11 },
-          nextCursor: { epoch: 'epoch-a', sequence: 11 }
-        },
-        liveCursor: { epoch: 'epoch-a', sequence: 11 },
-        hasOlder: true,
-        hasNewer: false
-      }
-    })
-
-    expect(refreshed.submissions.map((entry) => entry.clientMessageId)).toEqual(
-      Array.from({ length: 8 }, (_, index) => `client-${index}`)
-    )
-  })
-
-  it('bounds retained submission identities across repeated tail refreshes', () => {
-    let state = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-      type: 'event',
-      event: {
-        type: 'snapshot',
-        sessionId: 'session-a',
-        fence: 1,
-        page: hydrationPage([item('first', 1)])
-      }
-    })
-
-    for (let index = 0; index < 300; index += 1) {
-      state = reduceStructuredAgentSession(state, {
-        type: 'tail-page',
-        page: {
-          sessionId: 'session-a',
-          epoch: 'epoch-a',
-          direction: 'tail',
-          items: [item(`item-${index}`, index + 2)],
-          removedItemIds: [],
-          submissions: [submission(index)],
-          window: {
-            oldest: { epoch: 'epoch-a', sequence: index + 2 },
-            newest: { epoch: 'epoch-a', sequence: index + 2 },
-            nextCursor: { epoch: 'epoch-a', sequence: index + 2 }
-          },
-          liveCursor: { epoch: 'epoch-a', sequence: index + 2 },
-          hasOlder: true,
-          hasNewer: false
-        }
-      })
-    }
-
-    expect(state.submissions).toHaveLength(256)
-    expect(state.submissions[0]?.clientMessageId).toBe('client-44')
-    expect(state.submissions.at(-1)?.clientMessageId).toBe('client-299')
-  })
-
   it('projects additive background task state without changing transcript identity', () => {
     const initial = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
       type: 'event',
@@ -625,38 +445,6 @@ describe('structured agent session reducer', () => {
       9_800
     )
     expect(unstamped.hostClock).toEqual({ hostNow: 5_400, receivedAt: 9_400 })
-
-    const paged = reduceStructuredAgentSession(
-      unstamped,
-      { type: 'tail-page', page: { ...hydrationPage([item('fourth', 4)]), hostNow: 6_000 } },
-      10_000
-    )
-    expect(paged.hostClock).toEqual({ hostNow: 6_000, receivedAt: 10_000 })
-    expect(
-      reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-        type: 'tail-page',
-        page: hydrationPage([item('first', 1)])
-      }).hostClock
-    ).toBeUndefined()
-  })
-
-  it('retains same-epoch activity across a newer journal tail refresh', () => {
-    const active = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
-      type: 'event',
-      event: {
-        type: 'snapshot',
-        sessionId: 'session-a',
-        fence: 1,
-        page: hydrationPage([item('first', 1)]),
-        activity: { turnId: 'turn-1', text: 'Checking the renderer' }
-      }
-    })
-    const refreshed = reduceStructuredAgentSession(active, {
-      type: 'tail-page',
-      page: hydrationPage([item('latest', 2)])
-    })
-
-    expect(refreshed.activity).toEqual({ turnId: 'turn-1', text: 'Checking the renderer' })
   })
 })
 

src/shared/structured-agent-session-reducer.ts

@@ -45,7 +45,7 @@ export type StructuredAgentSessionAction =
   | { type: 'error'; message: string }
   | { type: 'handoff'; handoff: AgentSessionHandoffStatus }
   | { type: 'event'; event: AgentSessionSubscribeEvent }
-  | { type: 'tail-page'; page: AgentSessionHistoryPage }
+  | { type: 'history-page'; page: AgentSessionHistoryPage }
   | { type: 'older-page'; requestedCursor: AgentJournalCursor; page: AgentSessionHistoryPage }
 
 const MAX_RETAINED_SUBMISSIONS = 256
@@ -77,7 +77,7 @@ function hostClockField(
 
 function replacePage(
   page: AgentSessionHistoryPage,
-  fence: number,
+  fence: number | null,
   handoff?: AgentSessionHandoffStatus,
   backgroundTasks?: AgentSessionBackgroundTaskState | null,
   activity?: AgentSessionTurnActivity | null
@@ -165,56 +165,16 @@ export function reduceStructuredAgentSession(
   if (action.type === 'handoff') {
     return { ...state, handoff: action.handoff }
   }
-  if (action.type === 'tail-page') {
-    const pageCursor = action.page.liveCursor ?? action.page.window.newest
-    // An equal cursor means the page holds nothing the stream has not already
-    // delivered; replacing would throw away paged-in older items mid-scroll.
-    if (
-      state.epoch === action.page.epoch &&
-      state.cursor &&
-      (!pageCursor || pageCursor.sequence <= state.cursor.sequence)
-    ) {
-      const backgroundTasksChanged =
-        action.page.backgroundTasks !== undefined &&
-        !backgroundTaskStatesEqual(action.page.backgroundTasks, state.backgroundTasks)
-      if (
-        pageCursor?.sequence === state.cursor.sequence &&
-        ((action.page.fence !== undefined && action.page.fence !== state.fence) ||
-          backgroundTasksChanged)
-      ) {
-        return {
-          ...state,
-          ...(action.page.fence !== undefined ? { fence: action.page.fence } : {}),
-          ...(action.page.backgroundTasks !== undefined
-            ? { backgroundTasks: action.page.backgroundTasks }
-            : {}),
-          ...hostClockField(action.page.hostNow, receivedAt, state.hostClock),
-          status: 'ready',
-          error: undefined
-        }
-      }
-      return state
-    }
-    const sameEpoch = state.epoch === action.page.epoch
+  if (action.type === 'history-page') {
     return {
-      epoch: action.page.epoch,
-      cursor: action.page.liveCursor ?? null,
-      fence: action.page.fence ?? null,
-      items: action.page.items,
-      submissions: sameEpoch
-        ? mergeSubmissions(state.submissions, action.page.submissions, action.page.items)
-        : action.page.submissions,
-      retainedItemLimit: Math.max(MAX_RETAINED_ITEMS, action.page.items.length),
-      hasOlder: action.page.hasOlder,
-      status: 'ready',
-      handoff: state.handoff,
-      ...(sameEpoch ? { commands: state.commands } : {}),
-      ...(sameEpoch && state.activity !== undefined ? { activity: state.activity } : {}),
-      ...(action.page.backgroundTasks !== undefined
-        ? { backgroundTasks: action.page.backgroundTasks }
-        : state.backgroundTasks !== undefined
-          ? { backgroundTasks: state.backgroundTasks }
-          : {}),
+      ...replacePage(
+        action.page,
+        action.page.fence ?? null,
+        state.handoff ?? undefined,
+        state.backgroundTasks,
+        state.activity
+      ),
+      commands: state.commands,
       ...hostClockField(action.page.hostNow, receivedAt, state.hostClock)
     }
   }
@@ -309,12 +269,3 @@ export function oldestStructuredAgentSessionCursor(
   const oldest = state.items[0]
   return state.epoch && oldest ? { epoch: state.epoch, sequence: oldest.sequence } : null
 }
-
-export function shouldAdvanceStructuredResumeCursor(
-  current: AgentJournalCursor | null,
-  incoming: AgentJournalCursor
-): boolean {
-  return (
-    current === null || (current.epoch === incoming.epoch && incoming.sequence >= current.sequence)
-  )
-}

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

@@ -0,0 +1,251 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type {
+  AgentSessionHistoryRequest,
+  AgentSessionHistoryResult,
+  AgentSessionStatusSummary,
+  AgentSessionSubscribeEvent
+} from '../../src/shared/agent-session-wire'
+import type { AgentJournalCursor } from '../../src/shared/agent-session-journal-types'
+import {
+  EMPTY_STRUCTURED_AGENT_SESSION,
+  reduceStructuredAgentSession
+} from '../../src/shared/structured-agent-session-reducer'
+import {
+  hasUnansweredStructuredAgentSessionDispatch,
+  projectStructuredAgentSessionStatus
+} from '../../src/shared/structured-agent-session-projection'
+import { createTrackedJournalOpener } from '../../src/main/native-chat/agent-session-journal/journal-store-test-open'
+import { readAgentSessionHistory } from '../../src/main/native-chat/agent-session-wire/agent-session-history-page'
+import { AgentSessionSubscribers } from '../../src/main/native-chat/agent-session-wire/structured-agent-session-subscribers'
+import { StructuredAgentSessionStatusFeed } from '../../src/main/native-chat/agent-session-wire/structured-agent-session-status-feed'
+
+const mocks = vi.hoisted(() => ({ call: vi.fn(), subscribe: vi.fn() }))
+vi.mock('@/runtime/structured-agent-session-client', () => ({
+  callStructuredAgentSession: mocks.call,
+  subscribeStructuredAgentSession: mocks.subscribe
+}))
+
+import {
+  getStructuredAgentSessionReadOwner,
+  resetStructuredAgentSessionReadOwnersForTests
+} from '../../src/renderer/src/components/native-chat/structured-agent-session-read-owner'
+
+const SESSION = 'cursor-body-regression'
+const target = { kind: 'local' } as const
+const journals = createTrackedJournalOpener()
+let root: string
+
+beforeEach(async () => {
+  vi.resetAllMocks()
+  root = await mkdtemp(join(tmpdir(), 'orca-cursor-body-'))
+})
+afterEach(async () => {
+  resetStructuredAgentSessionReadOwnersForTests()
+  await journals.closeAll()
+  await rm(root, { recursive: true, force: true })
+})
+
+async function fixture() {
+  const journal = await journals.open({
+    identity: {
+      sessionId: SESSION,
+      workspaceId: 'folder-workspace',
+      hostId: 'local',
+      agent: 'codex',
+      providerHandle: { kind: 'codex', threadId: 'thread-1' }
+    },
+    journalDir: join(root, 'journal')
+  })
+  async function appendOutput(index: number) {
+    await journal.appendItem(
+      { provider: 'orca', clientMessageId: `output-${index}` },
+      { kind: 'status', text: `Tool output ${index}` },
+      { fence: 1 }
+    )
+  }
+  for (let index = 1; index < 99; index += 1) {
+    await appendOutput(index)
+  }
+  await journal.appendSubmission({
+    clientMessageId: 'pending-send',
+    payloadFingerprint: 'prompt',
+    body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'Run tools' }] },
+    fence: 1
+  })
+  expect(journal.cursor().sequence).toBe(100)
+  const initial = structuredClone(
+    readAgentSessionHistory(journal, { sessionId: SESSION, direction: 'tail' })
+  )
+  const accept = () =>
+    journal.resolveDispatch({
+      clientMessageId: 'pending-send',
+      fence: 1,
+      state: 'accepted',
+      providerIdentity: { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 0 }
+    })
+  return { journal, initial, appendOutput, accept }
+}
+
+describe('structured session cursor/body regression', () => {
+  it('replaces retained pending submissions together with a real bounded snapshot at 140', async () => {
+    const { journal, initial, appendOutput, accept } = await fixture()
+    const retained = reduceStructuredAgentSession(EMPTY_STRUCTURED_AGENT_SESSION, {
+      type: 'event',
+      event: { type: 'snapshot', sessionId: SESSION, page: initial.page, fence: 1 }
+    })
+    expect(retained.submissions[0]?.dispatchState).toBe('pending')
+    await accept()
+    for (let index = 102; index <= 140; index += 1) {
+      await appendOutput(index)
+    }
+    const bounded = readAgentSessionHistory(journal, {
+      sessionId: SESSION,
+      direction: 'tail',
+      limit: 1
+    }).page
+    expect(bounded.liveCursor?.sequence).toBe(140)
+    expect(bounded.items).not.toContainEqual(retained.items.at(-1))
+    expect(bounded.submissions).toEqual([])
+
+    const replaced = reduceStructuredAgentSession(retained, {
+      type: 'event',
+      event: { type: 'snapshot', sessionId: SESSION, page: bounded, fence: 1 }
+    })
+    expect(replaced.cursor).toEqual(bounded.liveCursor)
+    expect(replaced.submissions).toEqual(bounded.submissions)
+    expect(hasUnansweredStructuredAgentSessionDispatch(replaced.submissions, 1)).toBe(false)
+  })
+
+  it.each([40, 401])(
+    'replays an off-page dispatch after %i missed rows without stranding pending state',
+    async (missedRows) => {
+      const { journal, appendOutput, accept } = await fixture()
+      let hostSummary: AgentSessionStatusSummary | undefined
+      const feed = new StructuredAgentSessionStatusFeed({
+        sessions: new Map([
+          [
+            SESSION,
+            {
+              journal,
+              fence: 1,
+              params: { location: { workspaceId: 'folder-workspace' }, provider: 'codex' }
+            }
+          ]
+        ]),
+        getRecord: () => null,
+        now: () => 1_000,
+        onStatusChanged: (summary) => {
+          hostSummary = summary
+        }
+      })
+      const subscribers = new AgentSessionSubscribers({
+        onJournalPublished: (sessionId, published) => feed.publish(sessionId, published)
+      })
+      const delayedOlder = Promise.withResolvers<AgentSessionHistoryResult>()
+      let warm = false
+      mocks.call.mockImplementation((_target, _method, request: AgentSessionHistoryRequest) => {
+        // Hold the measured bounded page before its asynchronous older-page fill can mask it.
+        if (warm && missedRows === 40 && request.direction === 'before') {
+          return delayedOlder.promise
+        }
+        const result = readAgentSessionHistory(journal, {
+          ...request,
+          ...(warm && missedRows === 40 ? { limit: 1 } : {})
+        })
+        return Promise.resolve(
+          structuredClone({
+            ...result,
+            page: { ...result.page, fence: 1, hostNow: 1234 },
+            providerSession: { key: 'session_id', id: 'provider-1' }
+          })
+        )
+      })
+      mocks.subscribe.mockImplementation(
+        (
+          _target,
+          request: { cursor?: AgentJournalCursor },
+          onEvent: (event: AgentSessionSubscribeEvent) => void
+        ) =>
+          Promise.resolve({
+            unsubscribe: subscribers.open({
+              id: 'pane',
+              sessionId: SESSION,
+              journal,
+              fence: 1,
+              cursor: request.cursor,
+              emit: (event) => onEvent(structuredClone(event))
+            })
+          })
+      )
+      const owner = getStructuredAgentSessionReadOwner(SESSION, target)
+      const unlisten = owner.subscribe(() => {})
+      const deactivate = owner.activate()
+      await vi.waitFor(() => expect(mocks.subscribe).toHaveBeenCalledTimes(1))
+      expect(owner.getSnapshot().state.cursor?.sequence).toBe(100)
+      expect(owner.getSnapshot().state.items.at(-1)?.body).toMatchObject({ role: 'user' })
+      expect(owner.getSnapshot().state.submissions[0]?.dispatchState).toBe('pending')
+      expect(owner.getSnapshot().providerSession).toEqual({ key: 'session_id', id: 'provider-1' })
+      expect(owner.getSnapshot().state.hostClock?.hostNow).toBe(1234)
+      expect(mocks.call).toHaveBeenCalledTimes(1)
+      deactivate()
+
+      await accept()
+      for (let index = 102; index <= 100 + missedRows; index += 1) {
+        await appendOutput(index)
+      }
+      const tail = readAgentSessionHistory(journal, {
+        sessionId: SESSION,
+        direction: 'tail',
+        limit: missedRows === 40 ? 1 : 200
+      }).page
+      expect(tail.liveCursor?.sequence).toBe(100 + missedRows)
+      expect(tail.submissions).toEqual([])
+      feed.publish(SESSION, journal)
+      // IPC/RPC copies values; the journal mutates its own submission records in place.
+      expect(owner.getSnapshot().state.submissions[0]?.dispatchState).toBe('pending')
+      warm = true
+      const stop = owner.activate()
+
+      await vi.waitFor(() => expect(owner.getSnapshot().state.cursor).toEqual(journal.cursor()))
+      const caughtUp = owner.getSnapshot().state
+      if (missedRows === 40) {
+        expect({
+          cursor: caughtUp.cursor?.sequence,
+          dispatch: caughtUp.submissions[0]?.dispatchState,
+          unansweredDispatch: hasUnansweredStructuredAgentSessionDispatch(caughtUp.submissions, 1)
+        }).toEqual({ cursor: 140, dispatch: 'accepted', unansweredDispatch: false })
+      }
+
+      await journal.appendItem(
+        { provider: 'orca', clientMessageId: 'completed-turn' },
+        { kind: 'turn', turnId: 'turn-1', state: 'completed' },
+        { fence: 1 }
+      )
+      subscribers.publish(SESSION, journal)
+      await vi.waitFor(() => expect(owner.getSnapshot().state.cursor).toEqual(journal.cursor()))
+      const settled = owner.getSnapshot().state
+      expect(hostSummary?.status).toBe('idle')
+      expect(
+        projectStructuredAgentSessionStatus(settled.items, settled.submissions, settled.fence)
+      ).toBe(hostSummary?.status)
+      expect(settled.submissions).toEqual(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 })
+      expect(mocks.call).toHaveBeenCalledTimes(1)
+      expect(mocks.subscribe).toHaveBeenCalledTimes(2)
+      expect(mocks.subscribe.mock.calls[1]?.[1]).toEqual({
+        sessionId: SESSION,
+        cursor: { epoch: journal.cursor().epoch, sequence: 100 }
+      })
+
+      stop()
+      unlisten()
+    }
+  )
+})