fix mobile structured first-send hydration race

This commit is contained in:
Brennan Benson
2026-08-12 18:26:18 -07:00
parent d53823acc7
commit 8b2601b2f2
6 changed files with 154 additions and 26 deletions
@@ -0,0 +1,34 @@
import {
loadMobileStructuredOutbox,
type MobileStructuredOutboxEntry
} from './mobile-structured-outbox-store'
export type MobileStructuredOutboxHydration = {
sessionId: string
promise: Promise<void>
cancel: () => void
}
export function startMobileStructuredOutboxHydration(
sessionId: string,
onHydrated: (entries: MobileStructuredOutboxEntry[]) => void
): MobileStructuredOutboxHydration {
let stale = false
const promise = loadMobileStructuredOutbox(sessionId).then((entries) => {
if (!stale) {
onHydrated(entries)
}
})
return { sessionId, promise, cancel: () => (stale = true) }
}
export async function waitForMobileStructuredOutboxHydration(
hydration: MobileStructuredOutboxHydration | null,
sessionId: string
): Promise<boolean> {
if (!hydration || hydration.sessionId !== sessionId) {
return false
}
await hydration.promise
return true
}
@@ -0,0 +1,15 @@
import { useCallback, useRef } from 'react'
import {
saveMobileStructuredOutbox,
type MobileStructuredOutboxEntry
} from './mobile-structured-outbox-store'
export function useMobileStructuredOutboxPersistence() {
const persistTailRef = useRef<Promise<void>>(Promise.resolve())
return useCallback((sessionId: string, entries: MobileStructuredOutboxEntry[]) => {
const write = persistTailRef.current.then(() => saveMobileStructuredOutbox(sessionId, entries))
persistTailRef.current = write.catch(() => {})
return write
}, [])
}
@@ -8,8 +8,10 @@ import {
structuredAgentSessionOptionSnapshot
} from '../../../src/shared/structured-agent-session-options'
import type { SessionOptionValue } from '../../../src/shared/native-chat-session-options'
import type { AgentSessionOptionsResult } from '../../../src/shared/agent-session-wire'
import type { AgentSessionOptionResult } from '../../../src/shared/agent-session-wire'
import type {
AgentSessionOptionResult,
AgentSessionOptionsResult
} from '../../../src/shared/agent-session-wire'
import type { RpcClient } from '../transport/rpc-client'
import type { MobileNativeChatSessionOptionsController } from './use-mobile-native-chat-session-options'
@@ -144,6 +144,44 @@ describe('useMobileStructuredSessionWrites', () => {
expect(sendRequest).toHaveBeenCalledTimes(2)
})
it('dispatches the first send after a fresh session finishes outbox hydration', async () => {
act(() => renderer?.unmount())
renderer = null
api = null
const hydration = deferred<MobileStructuredOutboxStore.MobileStructuredOutboxEntry[]>()
vi.mocked(loadMobileStructuredOutbox).mockReset().mockReturnValue(hydration.promise)
sendRequest.mockImplementation(async (_method, params) =>
accepted((params as { envelope: { clientOperationId: string } }).envelope.clientOperationId)
)
await act(async () => {
renderer = create(createElement(Probe))
await Promise.resolve()
})
let sending!: Promise<boolean>
act(() => {
sending = api!.send('first message')
})
expect(sendRequest).not.toHaveBeenCalled()
expect(saveMobileStructuredOutbox).not.toHaveBeenCalled()
await act(async () => {
hydration.resolve([])
await expect(sending).resolves.toBe(true)
await Promise.resolve()
await Promise.resolve()
})
expect(sendRequest).toHaveBeenCalledTimes(1)
expect(sendRequest).toHaveBeenCalledWith(
'agentSession.send',
expect.objectContaining({
envelope: expect.objectContaining({ sessionId: 'mobile_1', expectedRuntimeFence: 3 })
})
)
expect(api!.outbox).toEqual([])
})
it.each(['agent_session_operation_conflict', 'agent_session_operation_expired'] as const)(
'rotates a send operation after %s',
async (code) => {
@@ -11,16 +11,18 @@ import {
} from '../../../src/shared/structured-agent-session-outbox'
import type { RpcClient } from '../transport/rpc-client'
import type { PendingNativeChatImage } from './mobile-native-chat-image-attachment'
import {
startMobileStructuredOutboxHydration,
waitForMobileStructuredOutboxHydration,
type MobileStructuredOutboxHydration
} from './mobile-structured-outbox-hydration'
import {
createQueuedMobileStructuredOutboxEntry,
isMobileStructuredDeliveryUnknown,
requeueMobileStructuredSendRefusal
} from './mobile-structured-outbox-entry'
import {
loadMobileStructuredOutbox,
saveMobileStructuredOutbox,
type MobileStructuredOutboxEntry
} from './mobile-structured-outbox-store'
import type { MobileStructuredOutboxEntry } from './mobile-structured-outbox-store'
import { useMobileStructuredOutboxPersistence } from './use-mobile-structured-outbox-persistence'
import { useMobileStructuredSessionMutations } from './use-mobile-structured-session-mutations'
import type { MobileStructuredSessionWrites } from './mobile-structured-session-write-types'
@@ -40,12 +42,13 @@ export function useMobileStructuredSessionWrites(args: {
const [error, setError] = useState<string | null>(null)
const [dispatchVersion, setDispatchVersion] = useState(0)
const outboxRef = useRef<MobileStructuredOutboxEntry[]>([])
const persistTailRef = useRef<Promise<void>>(Promise.resolve())
const hydrationRef = useRef<MobileStructuredOutboxHydration | null>(null)
const dispatchingRef = useRef(false)
const dispatchGenerationRef = useRef(0)
const blockedIdRef = useRef<string | null>(null)
const activeSessionRef = useRef(sessionId)
const activeFenceRef = useRef(fence)
const persist = useMobileStructuredOutboxPersistence()
const mutations = useMobileStructuredSessionMutations({
client,
sessionId,
@@ -54,14 +57,6 @@ export function useMobileStructuredSessionWrites(args: {
onRefusal: setError
})
const persist = useCallback((targetSessionId: string, entries: MobileStructuredOutboxEntry[]) => {
const write = persistTailRef.current.then(() =>
saveMobileStructuredOutbox(targetSessionId, entries)
)
persistTailRef.current = write.catch(() => {})
return write
}, [])
useEffect(() => {
activeSessionRef.current = sessionId
dispatchGenerationRef.current += 1
@@ -71,20 +66,17 @@ export function useMobileStructuredSessionWrites(args: {
setHydrated(false)
setError(null)
blockedIdRef.current = null
hydrationRef.current = null
if (!sessionId) {
return
}
let stale = false
void loadMobileStructuredOutbox(sessionId).then((entries) => {
if (!stale) {
outboxRef.current = entries
setOutbox(entries)
setHydrated(true)
}
const hydration = startMobileStructuredOutboxHydration(sessionId, (entries) => {
outboxRef.current = entries
setOutbox(entries)
setHydrated(true)
})
return () => {
stale = true
}
hydrationRef.current = hydration
return hydration.cancel
}, [sessionId])
useEffect(() => {
@@ -246,6 +238,14 @@ export function useMobileStructuredSessionWrites(args: {
if (!sessionId || (!text.trim() && attachments.length === 0)) {
return false
}
const hydration = hydrationRef.current
if (!(await waitForMobileStructuredOutboxHydration(hydration, sessionId))) {
setError('Chat session is still loading')
return false
}
if (activeSessionRef.current !== sessionId || hydrationRef.current !== hydration) {
return false
}
const entry = createQueuedMobileStructuredOutboxEntry({ sessionId, text, attachments })
const next = [...outboxRef.current, entry]
try {
@@ -380,6 +380,45 @@ describe('a structured codex session over agentSession.*', () => {
})
})
it('dispatches and streams a plain first send from a fresh session', async () => {
const created = await ok<{ fence: number; snapshot: { items: unknown[] } }>(
'agentSession.create',
createIntentParams()
)
expect(created.snapshot.items).toEqual([])
const stream = await subscribe('sub-first-send')
const body = { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }
const sent = await ok<{
clientMessageId: string
submission: { dispatchState: string; providerItemId: string | null }
}>('agentSession.send', {
envelope: envelope('agentSession.send', { body }, created.fence),
body
})
expect(sent.submission).toMatchObject({
dispatchState: 'accepted',
providerItemId: `codex:${THREAD}:${TURN}:0`
})
expect(codex.live().calls.at(-1)).toMatchObject({
method: 'turn/start',
params: { threadId: THREAD, clientUserMessageId: sent.clientMessageId }
})
codex.notify('turn/started', { turn: { id: TURN } })
codex.notify('item/completed', {
item: { type: 'userMessage', id: 'item-0', content: [{ type: 'text', text: 'hi' }] }
})
codex.notify('item/started', { item: { type: 'agentMessage', id: 'item-1', text: '' } })
codex.notify('item/agentMessage/delta', { itemId: 'item-1', delta: 'Hello.' })
codex.notify('item/completed', {
item: { type: 'agentMessage', id: 'item-1', text: 'Hello.' }
})
await drainStreamedEvents()
expect(itemsOf(stream).map(textOf).filter(Boolean)).toEqual(['hi', 'Hello.'])
})
it('runs create → send → stream → approval → cancel → reconnect → page history', async () => {
// ── create ──────────────────────────────────────────────────────────────
// No host exists yet; `create` is the call that builds one.