fix(mobile): make the reducer the single owner of the structured resume cursor

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<T> annotations on changed lines to T[].
This commit is contained in:
Brennan Benson
2026-08-19 14:11:16 -07:00
parent 41e6078bb4
commit a2475cbe01
4 changed files with 150 additions and 16 deletions
@@ -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<Pick<AgentSessionHandoffStatus, 'owner' | 'phase' | 'stage'>> = []
const projected: Pick<AgentSessionHandoffStatus, 'owner' | 'phase' | 'stage'>[] = []
for (const [index, handoff] of handoffs.entries()) {
state = reduceMobileStructuredAgentSession(state, {
@@ -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))
})
})
@@ -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<AgentJournalCursor | null>(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) => {
@@ -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