diff --git a/mobile/src/session/mobile-structured-agent-session-reducer.test.ts b/mobile/src/session/mobile-structured-agent-session-reducer.test.ts index 31a1a589470..ee53c1d62e7 100644 --- a/mobile/src/session/mobile-structured-agent-session-reducer.test.ts +++ b/mobile/src/session/mobile-structured-agent-session-reducer.test.ts @@ -84,7 +84,7 @@ describe('mobile structured session reducer', () => { { owner: 'native', direction: null, phase: 'idle', stage: null, operationId: null } ] let state = EMPTY_MOBILE_STRUCTURED_AGENT_SESSION - const projected: Array> = [] + const projected: Pick[] = [] for (const [index, handoff] of handoffs.entries()) { state = reduceMobileStructuredAgentSession(state, { diff --git a/mobile/src/session/use-mobile-structured-agent-session.test.ts b/mobile/src/session/use-mobile-structured-agent-session.test.ts new file mode 100644 index 00000000000..6642e598a0c --- /dev/null +++ b/mobile/src/session/use-mobile-structured-agent-session.test.ts @@ -0,0 +1,140 @@ +import { createElement } from 'react' +import { act, create, type ReactTestRenderer } from 'react-test-renderer' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { + AgentJournalCursor, + AgentJournalRenderItem +} from '../../../src/shared/agent-session-journal-types' +import type { AgentSessionSubscribeEvent } from '../../../src/shared/agent-session-wire' +import { STRUCTURED_AGENT_SESSION_CLIENT_COALESCE_MS } from '../../../src/shared/structured-agent-session-coalescer' +import type { SubscribeOptions } from '../transport/rpc-client-contract' +import type { RpcClient } from '../transport/rpc-client' +import type { RpcResponse } from '../transport/types' +import { useMobileStructuredAgentSession } from './use-mobile-structured-agent-session' + +vi.mock('react-native', () => ({ + AppState: { addEventListener: () => ({ remove: () => {} }) } +})) + +const SESSION_ID = 'session-a' +const EPOCH = 'epoch-a' + +function cursor(sequence: number): AgentJournalCursor { + return { epoch: EPOCH, sequence } +} + +// Assistant-message batches are the ones the coalescer holds; anything else +// bypasses it and would dispatch synchronously, hiding the window under test. +function assistantItem(sequence: number): AgentJournalRenderItem { + return { + itemId: `item-${sequence}`, + revision: 1, + sequence, + observedAt: sequence, + body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: `m${sequence}` }] } + } +} + +function batchEvent(sequence: number): AgentSessionSubscribeEvent { + return { + type: 'batch', + sessionId: SESSION_ID, + batch: { + cursor: cursor(sequence), + items: [assistantItem(sequence)], + removedItemIds: [], + submissions: [] + } + } +} + +function tailResponse(liveCursor: AgentJournalCursor): RpcResponse { + return { + id: 'req-1', + ok: true, + result: { + ok: true, + page: { + sessionId: SESSION_ID, + epoch: EPOCH, + fence: 1, + direction: 'tail', + items: [assistantItem(liveCursor.sequence)], + removedItemIds: [], + submissions: [], + window: { oldest: liveCursor, newest: liveCursor, nextCursor: liveCursor }, + liveCursor, + hasOlder: false, + hasNewer: false + } + }, + _meta: { runtimeId: 'runtime-1' } + } +} + +describe('mobile structured agent session resume cursor', () => { + let renderer: ReactTestRenderer | null = null + let onEvent: ((raw: unknown) => void) | null = null + let paramsForReconnect: (() => unknown) | null = null + let client: RpcClient + + function Harness(props: { tick: number }): null { + void props.tick + useMobileStructuredAgentSession({ client, sessionId: SESSION_ID }) + return null + } + + function resumeCursor(): AgentJournalCursor | undefined { + return (paramsForReconnect?.() as { cursor?: AgentJournalCursor } | undefined)?.cursor + } + + beforeEach(async () => { + vi.useFakeTimers() + onEvent = null + paramsForReconnect = null + client = { + sendRequest: vi.fn(async () => tailResponse(cursor(10))), + subscribe: vi.fn( + ( + _method: string, + _params: unknown, + onData: (raw: unknown) => void, + options?: SubscribeOptions + ) => { + onEvent = onData + paramsForReconnect = options?.paramsForReconnect ?? null + return () => {} + } + ) + } as unknown as RpcClient + await act(async () => { + renderer = create(createElement(Harness, { tick: 0 })) + }) + }) + + afterEach(() => { + act(() => renderer?.unmount()) + renderer = null + vi.useRealTimers() + }) + + it('resumes from the committed reducer cursor, not the newest one the stream handler saw', async () => { + expect(resumeCursor()).toEqual(cursor(10)) + + // Received but still sitting in the coalescer: the reducer has not applied it. + onEvent?.(batchEvent(11)) + expect(resumeCursor()).toEqual(cursor(10)) + + // A re-render between the handler and the flush must not change the answer. + await act(async () => { + renderer?.update(createElement(Harness, { tick: 1 })) + }) + expect(resumeCursor()).toEqual(cursor(10)) + + // Once the batch is committed, the resume cursor advances with it. + await act(async () => { + vi.advanceTimersByTime(STRUCTURED_AGENT_SESSION_CLIENT_COALESCE_MS) + }) + expect(resumeCursor()).toEqual(cursor(11)) + }) +}) diff --git a/mobile/src/session/use-mobile-structured-agent-session.ts b/mobile/src/session/use-mobile-structured-agent-session.ts index cf8fe08522c..7cabb9f4429 100644 --- a/mobile/src/session/use-mobile-structured-agent-session.ts +++ b/mobile/src/session/use-mobile-structured-agent-session.ts @@ -11,7 +11,6 @@ import { EMPTY_STRUCTURED_AGENT_SESSION, oldestStructuredAgentSessionCursor, reduceStructuredAgentSession, - shouldAdvanceStructuredResumeCursor, type StructuredAgentSessionState } from '../../../src/shared/structured-agent-session-reducer' import type { RpcClient } from '../transport/rpc-client' @@ -61,13 +60,16 @@ export function useMobileStructuredAgentSession(args: { const { client, sessionId } = args const [state, dispatch] = useReducer(reduceStructuredAgentSession, EMPTY_STRUCTURED_AGENT_SESSION) const stateRef = useRef(state) - stateRef.current = state - const resumeCursorRef = useRef(state.cursor) - resumeCursorRef.current = state.cursor const [loadingOlder, setLoadingOlder] = useState(false) const reconnectRef = useRef(createMobileStructuredReconnectState()) const cancelLongevityRef = useRef<() => void>(() => {}) + // The reducer owns the resume cursor, not the stream handlers: mirroring it + // post-commit means a reconnect can only replay, never skip undelivered items. + useEffect(() => { + stateRef.current = state + }, [state]) + useEffect(() => { dispatch({ type: 'loading' }) setLoadingOlder(false) @@ -107,7 +109,7 @@ export function useMobileStructuredAgentSession(args: { client, sessionId, cursor, - resumeCursor: () => resumeCursorRef.current, + resumeCursor: () => stateRef.current.cursor, onEvent: (raw) => { if (closed) { return @@ -123,12 +125,6 @@ export function useMobileStructuredAgentSession(args: { } if (event.type === 'snapshot' || event.type === 'reset') { setLoadingOlder(false) - resumeCursorRef.current = event.snapshot.cursor - } else if (event.type === 'batch') { - const current = resumeCursorRef.current - if (shouldAdvanceStructuredResumeCursor(current, event.batch.cursor)) { - resumeCursorRef.current = event.batch.cursor - } } coalescer.push(event) } @@ -146,7 +142,6 @@ export function useMobileStructuredAgentSession(args: { const result = response.result as AgentSessionHistoryResult if (result.ok) { dispatch({ type: 'tail-page', page: result.page }) - resumeCursorRef.current = result.page.liveCursor ?? null openSubscription(result.page.liveCursor ?? null) return } @@ -160,7 +155,6 @@ export function useMobileStructuredAgentSession(args: { fence: result.fence ?? 0 } }) - resumeCursorRef.current = result.snapshot.cursor openSubscription(result.snapshot.cursor) }) .catch((error: unknown) => { diff --git a/mobile/src/transport/rpc-client-test-websocket.ts b/mobile/src/transport/rpc-client-test-websocket.ts index 0fdbafa026c..f3958a78187 100644 --- a/mobile/src/transport/rpc-client-test-websocket.ts +++ b/mobile/src/transport/rpc-client-test-websocket.ts @@ -63,8 +63,8 @@ export function sentRequest( export function sentRequests( socket: MockWebSocket, method: string -): Array<{ id: string; params?: unknown }> { - const requests: Array<{ id: string; params?: unknown }> = [] +): { id: string; params?: unknown }[] { + const requests: { id: string; params?: unknown }[] = [] for (const payload of socket.sent) { const decoded = JSON.parse(payload.replace(/^encrypted:/, '')) as { id: string