From 8b2601b2f2492f72e12726dd56d30798e2ea4faa Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Wed, 12 Aug 2026 18:26:18 -0700 Subject: [PATCH] fix mobile structured first-send hydration race --- .../mobile-structured-outbox-hydration.ts | 34 +++++++++++++ ...se-mobile-structured-outbox-persistence.ts | 15 ++++++ .../use-mobile-structured-session-options.ts | 6 ++- ...e-mobile-structured-session-writes.test.ts | 38 +++++++++++++++ .../use-mobile-structured-session-writes.ts | 48 +++++++++---------- ...ructured-agent-session-integration.test.ts | 39 +++++++++++++++ 6 files changed, 154 insertions(+), 26 deletions(-) create mode 100644 mobile/src/session/mobile-structured-outbox-hydration.ts create mode 100644 mobile/src/session/use-mobile-structured-outbox-persistence.ts diff --git a/mobile/src/session/mobile-structured-outbox-hydration.ts b/mobile/src/session/mobile-structured-outbox-hydration.ts new file mode 100644 index 00000000000..77c0ba7315a --- /dev/null +++ b/mobile/src/session/mobile-structured-outbox-hydration.ts @@ -0,0 +1,34 @@ +import { + loadMobileStructuredOutbox, + type MobileStructuredOutboxEntry +} from './mobile-structured-outbox-store' + +export type MobileStructuredOutboxHydration = { + sessionId: string + promise: Promise + 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 { + if (!hydration || hydration.sessionId !== sessionId) { + return false + } + await hydration.promise + return true +} diff --git a/mobile/src/session/use-mobile-structured-outbox-persistence.ts b/mobile/src/session/use-mobile-structured-outbox-persistence.ts new file mode 100644 index 00000000000..1494fe5b357 --- /dev/null +++ b/mobile/src/session/use-mobile-structured-outbox-persistence.ts @@ -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.resolve()) + + return useCallback((sessionId: string, entries: MobileStructuredOutboxEntry[]) => { + const write = persistTailRef.current.then(() => saveMobileStructuredOutbox(sessionId, entries)) + persistTailRef.current = write.catch(() => {}) + return write + }, []) +} diff --git a/mobile/src/session/use-mobile-structured-session-options.ts b/mobile/src/session/use-mobile-structured-session-options.ts index ceade28479a..dbf8d92d523 100644 --- a/mobile/src/session/use-mobile-structured-session-options.ts +++ b/mobile/src/session/use-mobile-structured-session-options.ts @@ -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' diff --git a/mobile/src/session/use-mobile-structured-session-writes.test.ts b/mobile/src/session/use-mobile-structured-session-writes.test.ts index a01c0181ed8..ecfa6487cf6 100644 --- a/mobile/src/session/use-mobile-structured-session-writes.test.ts +++ b/mobile/src/session/use-mobile-structured-session-writes.test.ts @@ -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() + 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 + 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) => { diff --git a/mobile/src/session/use-mobile-structured-session-writes.ts b/mobile/src/session/use-mobile-structured-session-writes.ts index 296605362b6..fb565e3696e 100644 --- a/mobile/src/session/use-mobile-structured-session-writes.ts +++ b/mobile/src/session/use-mobile-structured-session-writes.ts @@ -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(null) const [dispatchVersion, setDispatchVersion] = useState(0) const outboxRef = useRef([]) - const persistTailRef = useRef>(Promise.resolve()) + const hydrationRef = useRef(null) const dispatchingRef = useRef(false) const dispatchGenerationRef = useRef(0) const blockedIdRef = useRef(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 { diff --git a/src/main/runtime/structured-agent-session-integration.test.ts b/src/main/runtime/structured-agent-session-integration.test.ts index 6a34527aab1..dae5155effd 100644 --- a/src/main/runtime/structured-agent-session-integration.test.ts +++ b/src/main/runtime/structured-agent-session-integration.test.ts @@ -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.