From a2475cbe0197dc76cb0a8bf778b40da164964ad0 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 19 Aug 2026 14:11:16 -0700 Subject: [PATCH] fix(mobile): make the reducer the single owner of the structured resume cursor MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit React Doctor's no-ref-current-in-render flagged the two render-time ref writes in useMobileStructuredAgentSession. Moving them into an effect alone would not have been enough: resumeCursorRef was also written by the live stream handlers, so the render assignment was a continuous repair of a two-writer race. The reducer is the correct owner. Its cursor is the newest one whose items are actually committed, and it is the only place that applies the epoch and sequence guards that decide whether a batch counts at all. A handler-owned ref advances at receipt, which can point past items the reducer rejected or has not applied — resuming there drops messages silently. Resuming from the committed cursor can only replay, and mergeItems is idempotent by itemId and revision. So resumeCursorRef is gone; resumeCursor now reads stateRef.current.cursor, and stateRef is mirrored from a post-commit effect. Session switches reset for free because 'loading' clears state.cursor. Also switches three Array annotations on changed lines to T[]. --- ...e-structured-agent-session-reducer.test.ts | 2 +- ...se-mobile-structured-agent-session.test.ts | 140 ++++++++++++++++++ .../use-mobile-structured-agent-session.ts | 20 +-- .../transport/rpc-client-test-websocket.ts | 4 +- 4 files changed, 150 insertions(+), 16 deletions(-) create mode 100644 mobile/src/session/use-mobile-structured-agent-session.test.ts 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