From a4c11f18899cb47d28cbad2db6bf96e0a694bfcb Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 14 Sep 2026 10:28:16 -0700 Subject: [PATCH] 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 --- .../structured-agent-session-read-owner.ts | 11 +- ...tured-agent-session-read-transport.test.ts | 66 ++++- ...structured-agent-session-read-transport.ts | 52 +--- ...use-structured-agent-session-read.test.tsx | 169 ++---------- .../use-structured-agent-session-read.ts | 13 - .../structured-agent-session-reducer.test.ts | 212 --------------- .../structured-agent-session-reducer.ts | 71 +---- ...ured-agent-session-read-owner.unit.test.ts | 251 ++++++++++++++++++ 8 files changed, 357 insertions(+), 488 deletions(-) create mode 100644 tests/e2e/structured-agent-session-read-owner.unit.test.ts 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 82c231c9982..68e1e3bc250 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 @@ -29,7 +29,6 @@ export type StructuredAgentSessionReadOwner = { dispose: () => void getSnapshot: () => StructuredAgentSessionReadSnapshot loadOlder: () => Promise - 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() @@ -91,7 +89,7 @@ function createReadOwner( setSnapshot({ ...snapshot, loadingOlder: false }) } } - const refreshTail = async (shouldStop: () => boolean): Promise => { + const hydrate = async (shouldStop: () => boolean): Promise => { const result = await callStructuredAgentSession( 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 () => { diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts index 555c09acdc4..5e5be1b6b3b 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.test.ts +++ b/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, + function startWithHydration( + hydrate: () => Promise, 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) diff --git a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts b/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts index b380c89d577..f4310cec890 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-read-transport.ts +++ b/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 + hydrate?: (shouldStop: () => boolean) => Promise 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 + } } } diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx b/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx index cb42c013ac4..faf4e5f3b92 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-read.test.tsx +++ b/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) }) }) diff --git a/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts b/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts index 834d11d3fa1..25dc3777d16 100644 --- a/src/renderer/src/components/native-chat/use-structured-agent-session-read.ts +++ b/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, diff --git a/src/shared/structured-agent-session-reducer.test.ts b/src/shared/structured-agent-session-reducer.test.ts index ca3c8774e0d..aa90749ffdf 100644 --- a/src/shared/structured-agent-session-reducer.test.ts +++ b/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' }) }) }) diff --git a/src/shared/structured-agent-session-reducer.ts b/src/shared/structured-agent-session-reducer.ts index 15a2ad34608..b86e927c15d 100644 --- a/src/shared/structured-agent-session-reducer.ts +++ b/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) - ) -} diff --git a/tests/e2e/structured-agent-session-read-owner.unit.test.ts b/tests/e2e/structured-agent-session-read-owner.unit.test.ts new file mode 100644 index 00000000000..4e64a047efc --- /dev/null +++ b/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() + 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() + } + ) +})