diff --git a/mobile/src/session/use-mobile-structured-agent-state.ts b/mobile/src/session/use-mobile-structured-agent-state.ts index 52aefab24aa..82399cce3cc 100644 --- a/mobile/src/session/use-mobile-structured-agent-state.ts +++ b/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(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( + 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) } }) diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts index 4c43326acb1..d568fb4e73f 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts +++ b/src/renderer/src/components/native-chat/structured-agent-session-read-owner.ts @@ -34,6 +34,9 @@ export type StructuredAgentSessionReadOwner = { const owners = new Map() +/** 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( - 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( + 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()) { diff --git a/src/shared/structured-agent-session-item-retention.test.ts b/src/shared/structured-agent-session-item-retention.test.ts index fe65b882596..f9763a02cba 100644 --- a/src/shared/structured-agent-session-item-retention.test.ts +++ b/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, { diff --git a/src/shared/structured-agent-session-reducer.test.ts b/src/shared/structured-agent-session-reducer.test.ts index 759d42f01d5..f617db975f5 100644 --- a/src/shared/structured-agent-session-reducer.test.ts +++ b/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', diff --git a/src/shared/structured-agent-session-reducer.ts b/src/shared/structured-agent-session-reducer.ts index 52555d103a8..ea32587178b 100644 --- a/src/shared/structured-agent-session-reducer.ts +++ b/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)