fix(native-chat): keep an older page from punching a hole in the transcript (#19845)

mobile/src/session/use-mobile-structured-agent-state.ts

@@ -16,6 +16,8 @@ import type { RpcClient } from '../transport/rpc-client'
 import { callAgentSession } from './mobile-structured-agent-session-rpc'
 
 const MAX_RETAINED_SESSION_STATES = 32
+/** Bounded so a busy stream cannot turn one Load-earlier tap into an endless read chain. */
+const OLDER_PAGE_ANCHOR_ATTEMPTS = 3
 
 function isSubscribeEvent(value: unknown): value is AgentSessionSubscribeEvent {
   if (typeof value !== 'object' || value === null) {
@@ -154,41 +156,45 @@ export function useMobileStructuredAgentState(args: {
     if (!client || !sessionId || !sessionKey || loadingOlder || !current.hasOlder) {
       return
     }
-    const cursor = oldestStructuredAgentSessionCursor(current)
-    if (!cursor) {
+    if (!oldestStructuredAgentSessionCursor(current)) {
       return
     }
     const requestSessionKey = sessionKey
     const requestGeneration = streamGenerationRef.current
+    const isCurrentRead = (): boolean =>
+      sessionKeyRef.current === requestSessionKey &&
+      streamGenerationRef.current === requestGeneration
     setLoadingOlder(true)
-    void callAgentSession<AgentSessionHistoryResult>(client, 'agentSession.history', {
-      sessionId,
-      direction: 'before',
-      cursor,
-      limit: AGENT_SESSION_HISTORY_MAX_LIMIT
-    })
-      .then((result) => {
-        if (
-          result.ok &&
-          sessionKeyRef.current === requestSessionKey &&
-          streamGenerationRef.current === requestGeneration
-        ) {
-          apply({ type: 'older-page', requestedEpoch: cursor.epoch, page: result.page })
+    void (async () => {
+      // A live batch can head-trim past the anchor mid-read, and the reducer drops that
+      // page rather than leave a hole in the transcript. Re-anchor and retry.
+      for (let attempt = 0; attempt < OLDER_PAGE_ANCHOR_ATTEMPTS; attempt += 1) {
+        const cursor = oldestStructuredAgentSessionCursor(stateRef.current)
+        if (!cursor || !isCurrentRead()) {
+          return
         }
-      })
+        const result = await callAgentSession<AgentSessionHistoryResult>(
+          client,
+          'agentSession.history',
+          { sessionId, direction: 'before', cursor, limit: AGENT_SESSION_HISTORY_MAX_LIMIT }
+        )
+        if (!result.ok || !isCurrentRead()) {
+          return
+        }
+        // The reducer drops a page whose anchor slid, so only an intact anchor lands.
+        if (oldestStructuredAgentSessionCursor(stateRef.current)?.sequence === cursor.sequence) {
+          apply({ type: 'older-page', requestedCursor: cursor, page: result.page })
+          return
+        }
+      }
+    })()
       .catch((error: unknown) => {
-        if (
-          sessionKeyRef.current === requestSessionKey &&
-          streamGenerationRef.current === requestGeneration
-        ) {
+        if (isCurrentRead()) {
           apply({ type: 'error', message: error instanceof Error ? error.message : String(error) })
         }
       })
       .finally(() => {
-        if (
-          sessionKeyRef.current === requestSessionKey &&
-          streamGenerationRef.current === requestGeneration
-        ) {
+        if (isCurrentRead()) {
           setLoadingOlder(false)
         }
       })

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

@@ -34,6 +34,9 @@ export type StructuredAgentSessionReadOwner = {
 
 const owners = new Map<string, StructuredAgentSessionReadOwner>()
 
+/** Bounded so a busy stream cannot turn one scroll-to-top into an endless read chain. */
+const OLDER_PAGE_ANCHOR_ATTEMPTS = 3
+
 function countsTowardInitialHistory(item: AgentJournalRenderItem): boolean {
   return item.body.kind !== 'status' || !item.body.providerFrame
 }
@@ -121,6 +124,7 @@ function createReadOwner(
       return
     }
     let restored = snapshot.state.items.filter(countsTowardInitialHistory).length
+    let anchorSlides = 0
     while (snapshot.state.hasOlder && restored < NATIVE_CHAT_INITIAL_LIMIT) {
       const oldest = oldestStructuredAgentSessionCursor(snapshot.state)
       if (!oldest || shouldStop()) {
@@ -146,7 +150,16 @@ function createReadOwner(
       if (shouldStop()) {
         return
       }
-      apply({ type: 'older-page', requestedEpoch: oldest.epoch, page: older.page })
+      // A live batch that head-trimmed past the anchor makes this page discontiguous;
+      // the reducer drops it, so re-anchor rather than chase a moving window forever.
+      if (oldestStructuredAgentSessionCursor(snapshot.state)?.sequence !== oldest.sequence) {
+        anchorSlides += 1
+        if (anchorSlides >= OLDER_PAGE_ANCHOR_ATTEMPTS) {
+          break
+        }
+        continue
+      }
+      apply({ type: 'older-page', requestedCursor: oldest, page: older.page })
       if (shouldStop()) {
         return
       }
@@ -209,25 +222,35 @@ function createReadOwner(
       if (shouldStop()) {
         return
       }
-      const cursor = oldestStructuredAgentSessionCursor(snapshot.state)
-      if (!cursor || !snapshot.state.hasOlder || snapshot.loadingOlder) {
-        return
-      }
-      if (shouldStop()) {
+      if (
+        !oldestStructuredAgentSessionCursor(snapshot.state) ||
+        !snapshot.state.hasOlder ||
+        snapshot.loadingOlder
+      ) {
         return
       }
       setSnapshot({ ...snapshot, loadingOlder: true })
       try {
-        const result = await callStructuredAgentSession<AgentSessionHistoryResult>(
-          target,
-          'agentSession.history',
-          { sessionId, direction: 'before', cursor, limit: AGENT_SESSION_HISTORY_MAX_LIMIT }
-        )
-        if (shouldStop()) {
-          return
-        }
-        if (result.ok && !shouldStop()) {
-          apply({ type: 'older-page', requestedEpoch: cursor.epoch, page: result.page })
+        // A live batch can head-trim past the anchor mid-read, and the reducer drops
+        // that page rather than leave a hole in the transcript. Re-anchor and retry.
+        for (let attempt = 0; attempt < OLDER_PAGE_ANCHOR_ATTEMPTS; attempt += 1) {
+          const cursor = oldestStructuredAgentSessionCursor(snapshot.state)
+          if (!cursor || shouldStop()) {
+            return
+          }
+          const result = await callStructuredAgentSession<AgentSessionHistoryResult>(
+            target,
+            'agentSession.history',
+            { sessionId, direction: 'before', cursor, limit: AGENT_SESSION_HISTORY_MAX_LIMIT }
+          )
+          if (shouldStop() || !result.ok) {
+            return
+          }
+          // The reducer drops a page whose anchor slid, so only an intact anchor lands.
+          if (oldestStructuredAgentSessionCursor(snapshot.state)?.sequence === cursor.sequence) {
+            apply({ type: 'older-page', requestedCursor: cursor, page: result.page })
+            return
+          }
         }
       } catch (error) {
         if (!shouldStop()) {

src/shared/structured-agent-session-item-retention.test.ts

@@ -114,7 +114,7 @@ describe('structured agent session item retention', () => {
     )
     const older = reduceStructuredAgentSession(streamed, {
       type: 'older-page',
-      requestedEpoch: 'epoch-a',
+      requestedCursor: { epoch: 'epoch-a', sequence: streamed.items[0]?.sequence ?? 0 },
       page: page(
         Array.from({ length: 300 }, (_, index) => item(index + 700)),
         true
@@ -130,6 +130,48 @@ describe('structured agent session item retention', () => {
     expect(afterLive.items.some((entry) => entry.sequence === 800)).toBe(true)
   })
 
+  it('drops an older page whose anchor a live batch trimmed past', () => {
+    const streamed = streamItems(
+      hydrate([item(0)], false),
+      Array.from({ length: CAP + 200 }, (_, index) => index + 1)
+    )
+    // The read captures this cursor, then a live batch trims three items off the head.
+    const anchor = oldestStructuredAgentSessionCursor(streamed)
+    const slid = streamItems(streamed, [CAP + 201, CAP + 202, CAP + 203])
+    expect(slid.items[0]?.sequence).toBeGreaterThan(anchor?.sequence ?? 0)
+
+    const merged = reduceStructuredAgentSession(slid, {
+      type: 'older-page',
+      requestedCursor: anchor ?? { epoch: 'epoch-a', sequence: 0 },
+      page: page(
+        Array.from({ length: 200 }, (_, index) => item((anchor?.sequence ?? 0) - 200 + index)),
+        true
+      )
+    })
+
+    // Merging would have left a hole between the page and the retained window.
+    expect(merged).toBe(slid)
+  })
+
+  it('accepts an older page whose anchor still matches the retained head', () => {
+    const streamed = streamItems(
+      hydrate([item(0)], false),
+      Array.from({ length: CAP + 200 }, (_, index) => index + 1)
+    )
+    const anchor = oldestStructuredAgentSessionCursor(streamed)
+    const merged = reduceStructuredAgentSession(streamed, {
+      type: 'older-page',
+      requestedCursor: anchor ?? { epoch: 'epoch-a', sequence: 0 },
+      page: page(
+        Array.from({ length: 200 }, (_, index) => item((anchor?.sequence ?? 0) - 200 + index)),
+        true
+      )
+    })
+
+    expect(merged.items).toHaveLength(CAP + 200)
+    expect(merged.items[0]?.sequence).toBe((anchor?.sequence ?? 0) - 200)
+  })
+
   it('keeps item identity stable when a batch carries no journal change', () => {
     const hydrated = hydrate([item(0)])
     const unchanged = reduceStructuredAgentSession(hydrated, {

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

@@ -149,7 +149,7 @@ describe('structured agent session reducer', () => {
     })
     const withOlder = reduceStructuredAgentSession(snapshot, {
       type: 'older-page',
-      requestedEpoch: 'epoch-a',
+      requestedCursor: { epoch: 'epoch-a', sequence: 50 },
       page: {
         sessionId: 'session-a',
         epoch: 'epoch-a',

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

@@ -36,7 +36,7 @@ export type StructuredAgentSessionAction =
   | { type: 'handoff'; handoff: AgentSessionHandoffStatus }
   | { type: 'event'; event: AgentSessionSubscribeEvent }
   | { type: 'tail-page'; page: AgentSessionHistoryPage }
-  | { type: 'older-page'; requestedEpoch: string; page: AgentSessionHistoryPage }
+  | { type: 'older-page'; requestedCursor: AgentJournalCursor; page: AgentSessionHistoryPage }
 
 const MAX_RETAINED_SUBMISSIONS = 256
 // Well above the renderer's initial read window (300) plus a page, so only genuinely
@@ -207,7 +207,15 @@ export function reduceStructuredAgentSession(
     }
   }
   if (action.type === 'older-page') {
-    if (state.epoch !== action.requestedEpoch || action.page.epoch !== action.requestedEpoch) {
+    const requested = action.requestedCursor
+    if (state.epoch !== requested.epoch || action.page.epoch !== requested.epoch) {
+      return state
+    }
+    const head = state.items[0]
+    // A live batch head-trimmed past the anchor while this read was in flight, so the
+    // page no longer abuts the retained window; merging it would leave a silent hole.
+    // The caller re-anchors on the new head and asks again.
+    if (head && head.sequence > requested.sequence) {
       return state
     }
     const paged = mergeItems(state.items, action.page.items, action.page.removedItemIds)