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() + } + ) +})