From 8778fdedad333fb863342a8ea6f5ea87586f88b9 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sun, 9 Aug 2026 19:58:58 -0700 Subject: [PATCH] Harden structured mobile writes before RC --- .../app/h/[hostId]/session/[worktreeId].tsx | 1 + .../session/MobileNativeChatQuestion.test.tsx | 53 +++++ .../src/session/MobileNativeChatQuestion.tsx | 65 +++--- .../MobileStructuredAgentSessionView.tsx | 12 +- .../session/MobileStructuredPromptCard.tsx | 1 + .../mobile-image-source-picker.test.ts | 54 +++++ .../src/session/mobile-image-source-picker.ts | 33 +-- ...ile-structured-agent-session-projection.ts | 3 + .../mobile-structured-mutation-envelope.ts | 21 ++ .../session/mobile-structured-outbox-entry.ts | 45 +++- .../mobile-structured-outbox-store.test.ts | 18 +- .../session/mobile-structured-outbox-store.ts | 18 +- ...mobile-structured-session-timeline.test.ts | 34 ++- .../mobile-structured-session-timeline.ts | 9 +- .../mobile-structured-session-write-types.ts | 12 ++ .../use-mobile-structured-attachments.test.ts | 62 ++++++ .../use-mobile-structured-attachments.ts | 94 ++++++--- ...e-mobile-structured-composer-state.test.ts | 46 +++++ .../use-mobile-structured-composer-state.ts | 14 ++ .../use-mobile-structured-session-entry.ts | 2 +- ...use-mobile-structured-session-mutations.ts | 3 +- ...-mobile-structured-session-options.test.ts | 46 ++++- .../use-mobile-structured-session-options.ts | 25 ++- ...e-mobile-structured-session-writes.test.ts | 195 ++++++++++++++++-- .../use-mobile-structured-session-writes.ts | 151 ++++++++------ ...dex-structured-journal-translation.test.ts | 27 ++- .../codex-structured-journal-translation.ts | 37 ++++ .../codex/codex-structured-session-adapter.ts | 4 + src/shared/agent-session-journal-types.ts | 3 + 29 files changed, 904 insertions(+), 184 deletions(-) create mode 100644 mobile/src/session/MobileNativeChatQuestion.test.tsx create mode 100644 mobile/src/session/mobile-structured-session-write-types.ts create mode 100644 mobile/src/session/use-mobile-structured-attachments.test.ts create mode 100644 mobile/src/session/use-mobile-structured-composer-state.test.ts create mode 100644 mobile/src/session/use-mobile-structured-composer-state.ts diff --git a/mobile/app/h/[hostId]/session/[worktreeId].tsx b/mobile/app/h/[hostId]/session/[worktreeId].tsx index 5f17e2d29a9..2814a8a27fc 100644 --- a/mobile/app/h/[hostId]/session/[worktreeId].tsx +++ b/mobile/app/h/[hostId]/session/[worktreeId].tsx @@ -4634,6 +4634,7 @@ export default function SessionScreen() { ) : activeStructuredTab ? ( ({ + Pressable: 'Pressable', + StyleSheet: { create: (styles: unknown) => styles, hairlineWidth: 1 }, + Text: 'Text', + TextInput: 'TextInput', + View: 'View' +})) + +vi.mock('lucide-react-native', () => ({ + ArrowUp: 'ArrowUp', + Check: 'Check', + CircleHelp: 'CircleHelp' +})) + +const QUESTION = { + question: 'Choose one', + options: ['Allow', 'Deny'], + multiSelect: false, + optionTokens: [null, null] +} + +describe('MobileNativeChatQuestion', () => { + let renderer: ReactTestRenderer | null = null + + beforeEach(() => { + globalThis.IS_REACT_ACT_ENVIRONMENT = true + }) + + afterEach(() => { + act(() => renderer?.unmount()) + renderer = null + }) + + it('does not render a dead free-text control when the prompt disallows it', () => { + act(() => { + renderer = create( + createElement(MobileNativeChatQuestion, { + question: QUESTION, + allowFreeText: false, + onAnswer: vi.fn(async () => true) + }) + ) + }) + + expect(renderer!.root.findAllByType(TextInput)).toHaveLength(0) + }) +}) diff --git a/mobile/src/session/MobileNativeChatQuestion.tsx b/mobile/src/session/MobileNativeChatQuestion.tsx index f470214dbed..fe59815b758 100644 --- a/mobile/src/session/MobileNativeChatQuestion.tsx +++ b/mobile/src/session/MobileNativeChatQuestion.tsx @@ -7,13 +7,18 @@ import { formatQuestionAnswer, type MobileChatQuestion } from './mobile-native-c type Props = { question: MobileChatQuestion onAnswer: (text: string) => Promise + allowFreeText?: boolean } /** Renders an agent's choice prompt as a tappable card. Single-select answers * on tap; multi-select toggles then Submits; an always-present text entry lets * the user answer freely (the escape hatch) when the heuristic misreads the * options or none apply. */ -export function MobileNativeChatQuestion({ question, onAnswer }: Props): React.JSX.Element { +export function MobileNativeChatQuestion({ + question, + onAnswer, + allowFreeText = true +}: Props): React.JSX.Element { const [selected, setSelected] = useState([]) const [freeText, setFreeText] = useState('') const [sending, setSending] = useState(false) @@ -124,35 +129,37 @@ export function MobileNativeChatQuestion({ question, onAnswer }: Props): React.J ) : null} - - - [ - styles.freeSend, - !canSendFreeText && styles.freeSendDisabled, - pressed && canSendFreeText && styles.pressed - ]} - onPress={submitFreeText} - disabled={!canSendFreeText} - > - + - - + [ + styles.freeSend, + !canSendFreeText && styles.freeSendDisabled, + pressed && canSendFreeText && styles.pressed + ]} + onPress={submitFreeText} + disabled={!canSendFreeText} + > + + + + ) : null} ) } diff --git a/mobile/src/session/MobileStructuredAgentSessionView.tsx b/mobile/src/session/MobileStructuredAgentSessionView.tsx index 2e43a5c7305..6d31233aa2f 100644 --- a/mobile/src/session/MobileStructuredAgentSessionView.tsx +++ b/mobile/src/session/MobileStructuredAgentSessionView.tsx @@ -1,4 +1,4 @@ -import { useCallback, useMemo, useRef, useState } from 'react' +import { useCallback, useMemo, useRef } from 'react' import { ActivityIndicator, FlatList, @@ -23,7 +23,7 @@ import type { PendingNativeChatImage } from './mobile-native-chat-image-attachme import type { MobileStructuredOutboxEntry } from './mobile-structured-outbox-store' import { buildMobileStructuredTimeline, - latestMobileStructuredTurnId, + activeMobileStructuredTurnId, mobileStructuredOutboxText, restoreMobileStructuredAttachments, type MobileStructuredTimelineRow @@ -36,6 +36,7 @@ import { settleStructuredOlderPage } from './mobile-structured-history-pagination' import { styles } from './mobile-structured-agent-session-view-styles' +import { useMobileStructuredComposerState } from './use-mobile-structured-composer-state' type Props = { items: AgentJournalRenderItem[] @@ -64,9 +65,9 @@ export function MobileStructuredAgentSessionView(props: Props): React.JSX.Elemen const paginationRef = useRef(createMobileStructuredPaginationState()) const priorContentHeightRef = useRef(0) const androidAnchorOffsetRef = useRef(0) - const [composerText, setComposerText] = useState('') - const [restored, setRestored] = useState([]) - const turnId = latestMobileStructuredTurnId(props.items) + const { composerText, setComposerText, restored, setRestored } = + useMobileStructuredComposerState() + const turnId = activeMobileStructuredTurnId(props.items) const allAttachments = [...restored, ...props.attachments] const rows = useMemo( () => buildMobileStructuredTimeline(props.items, props.outbox), @@ -157,7 +158,6 @@ export function MobileStructuredAgentSessionView(props: Props): React.JSX.Elemen if (Platform.OS !== 'android' || priorHeight === 0 || height <= priorHeight) { return } - paginationRef.current.programmaticMomentum = true listRef.current?.scrollToOffset({ offset: androidAnchorOffsetRef.current + height - priorHeight, animated: false diff --git a/mobile/src/session/MobileStructuredPromptCard.tsx b/mobile/src/session/MobileStructuredPromptCard.tsx index 5b0cf900b0b..4b53af61d14 100644 --- a/mobile/src/session/MobileStructuredPromptCard.tsx +++ b/mobile/src/session/MobileStructuredPromptCard.tsx @@ -44,6 +44,7 @@ export function MobileStructuredPromptCard(props: { return ( { const offered = questionBody.options.find((option) => option.label === answer) const optionId = diff --git a/mobile/src/session/mobile-image-source-picker.test.ts b/mobile/src/session/mobile-image-source-picker.test.ts index bcaf75daa14..26f177e3d80 100644 --- a/mobile/src/session/mobile-image-source-picker.test.ts +++ b/mobile/src/session/mobile-image-source-picker.test.ts @@ -1,4 +1,5 @@ import { afterEach, describe, expect, it, vi } from 'vitest' +import { ImageManipulator } from 'expo-image-manipulator' import { CLIPBOARD_IMAGE_MAX_SOURCE_BYTES } from '../../../src/shared/clipboard-image' vi.mock('expo-image-picker', () => ({ @@ -20,6 +21,8 @@ import { ImageLibraryPermissionError, pickMobileImage, pickMobileImages, + MOBILE_IMAGE_DOWNSAMPLE_TARGET_BASE64, + prepareMobileImageForUpload, shouldDownsampleMobileImage, type PickedMobileImage } from './mobile-image-source-picker' @@ -242,4 +245,55 @@ describe('pickMobileImage', () => { height: 3000 }) }) + + it('never guesses a resize that can upscale an image with unknown dimensions', async () => { + const resize = vi.fn() + const release = vi.fn() + vi.mocked(ImageManipulator.manipulate).mockReturnValue({ + resize, + renderAsync: vi.fn().mockResolvedValue({ + saveAsync: vi.fn().mockResolvedValue({ base64: 'AAAA', uri: 'file:///compressed.png' }), + release + }), + release + } as never) + + await prepareMobileImageForUpload({ + uri: 'file:///unknown.png', + fileSize: 5 * 1024 * 1024 + }) + + expect(resize).not.toHaveBeenCalled() + }) + + it('keeps the final quality-rung preview file while deleting superseded rungs', async () => { + const oversized = 'A'.repeat(MOBILE_IMAGE_DOWNSAMPLE_TARGET_BASE64 + 1) + let attempt = 0 + vi.mocked(ImageManipulator.manipulate).mockImplementation(() => { + const uri = `file:///quality-${++attempt}.png` + const release = vi.fn() + return { + resize: vi.fn(), + renderAsync: vi.fn().mockResolvedValue({ + saveAsync: vi.fn().mockResolvedValue({ base64: oversized, uri }), + release + }), + release + } as never + }) + const deleted: string[] = [] + + const result = await prepareMobileImageForUpload( + { + uri: 'file:///large.png', + fileSize: 5 * 1024 * 1024, + width: 4000, + height: 3000 + }, + (uri) => deleted.push(uri) + ) + + expect(result?.uri).toBe('file:///quality-3.png') + expect(deleted).toEqual(['file:///quality-1.png', 'file:///quality-2.png']) + }) }) diff --git a/mobile/src/session/mobile-image-source-picker.ts b/mobile/src/session/mobile-image-source-picker.ts index d66c861c5e1..85e6c370ffd 100644 --- a/mobile/src/session/mobile-image-source-picker.ts +++ b/mobile/src/session/mobile-image-source-picker.ts @@ -68,12 +68,15 @@ export function shouldDownsampleMobileImage(input: { ) } -async function downsampleMobileImage(input: { - uri: string - fileSize?: number - width?: number - height?: number -}): Promise { +export async function prepareMobileImageForUpload( + input: { + uri: string + fileSize?: number + width?: number + height?: number + }, + deletePreparedFile: (uri: string) => void = (uri) => new FsFile(uri).delete() +): Promise { if (!shouldDownsampleMobileImage(input)) { return null } @@ -90,8 +93,6 @@ async function downsampleMobileImage(input: { width: Math.max(1, Math.round(sourceWidth * edgeScale)), height: Math.max(1, Math.round(sourceHeight * edgeScale)) }) - } else { - context.resize({ width, height: null }) } const rendered = await context.renderAsync() try { @@ -103,15 +104,17 @@ async function downsampleMobileImage(input: { if (!result.base64) { throw new Error('Failed to encode resized image') } + if (last) { + try { + deletePreparedFile(last.uri) + } catch { + // Cache cleanup is best effort; the OS reclaims it independently. + } + } last = { base64: result.base64, uri: result.uri } if (result.base64.length <= MOBILE_IMAGE_DOWNSAMPLE_TARGET_BASE64) { return last } - try { - new FsFile(result.uri).delete() - } catch { - // Cache cleanup is best effort; the OS reclaims it independently. - } } finally { rendered.release() context.release() @@ -167,7 +170,7 @@ async function* pickFromLibrary( requestPermission: typeof ImagePicker.requestMediaLibraryPermissionsAsync = ImagePicker.requestMediaLibraryPermissionsAsync, launch: typeof ImagePicker.launchImageLibraryAsync = ImagePicker.launchImageLibraryAsync, createFile: MobileImageFileFactory = defaultMobileImageFileFactory, - prepareImage: PrepareMobileImage = downsampleMobileImage + prepareImage: PrepareMobileImage = prepareMobileImageForUpload ): AsyncGenerator { const permission = await requestPermission() // Why: `granted` covers full + limited iOS access; only a hard denial blocks us. @@ -206,7 +209,7 @@ async function* pickFromFiles( multiple: boolean, launch: typeof DocumentPicker.getDocumentAsync = DocumentPicker.getDocumentAsync, createFile: MobileImageFileFactory = defaultMobileImageFileFactory, - prepareImage: PrepareMobileImage = downsampleMobileImage + prepareImage: PrepareMobileImage = prepareMobileImageForUpload ): AsyncGenerator { const result = await launch({ type: 'image/*', diff --git a/mobile/src/session/mobile-structured-agent-session-projection.ts b/mobile/src/session/mobile-structured-agent-session-projection.ts index 28c8329f67a..0d10acf891c 100644 --- a/mobile/src/session/mobile-structured-agent-session-projection.ts +++ b/mobile/src/session/mobile-structured-agent-session-projection.ts @@ -59,6 +59,9 @@ function itemBlocks(item: AgentJournalRenderItem): { blocks: [{ type: 'text', text: `${body.question}\n${choices}`.trim() }] } } + if (body.turnLifecycle) { + return null + } return { role: 'system', blocks: [{ type: 'text', text: body.text }] } } diff --git a/mobile/src/session/mobile-structured-mutation-envelope.ts b/mobile/src/session/mobile-structured-mutation-envelope.ts index 8ba9096a66b..7dad042ef61 100644 --- a/mobile/src/session/mobile-structured-mutation-envelope.ts +++ b/mobile/src/session/mobile-structured-mutation-envelope.ts @@ -1,4 +1,5 @@ import { sha256 } from '@noble/hashes/sha256' +import type { MobileStructuredOutboxEntry } from './mobile-structured-outbox-store' function canonicalize(value: unknown): string { if (value === null || typeof value !== 'object') { @@ -33,3 +34,23 @@ export function createMobileStructuredOperationId( ): string { return `${prefix}:${now.toString(36)}:${randomUuid()}` } + +export function mobileStructuredSendRequest( + entry: MobileStructuredOutboxEntry, + expectedRuntimeFence: number +): Record { + const fields = { body: entry.body } + return { + envelope: { + sessionId: entry.sessionId, + clientOperationId: entry.clientMessageId, + expectedRuntimeFence, + payloadFingerprint: mobileStructuredPayloadFingerprint({ + method: 'agentSession.send', + sessionId: entry.sessionId, + fields + }) + }, + ...fields + } +} diff --git a/mobile/src/session/mobile-structured-outbox-entry.ts b/mobile/src/session/mobile-structured-outbox-entry.ts index 3b77e9cafc8..0bf1690072f 100644 --- a/mobile/src/session/mobile-structured-outbox-entry.ts +++ b/mobile/src/session/mobile-structured-outbox-entry.ts @@ -1,4 +1,7 @@ -import type { AgentJournalMessageItem } from '../../../src/shared/agent-session-journal-types' +import type { + AgentJournalMessageItem, + AgentJournalSubmission +} from '../../../src/shared/agent-session-journal-types' import { isRpcDeliveryUnknown } from '../transport/rpc-delivery-ambiguity' import { isLogicalClientCutoverError } from '../transport/stable-logical-rpc-client' import type { PendingNativeChatImage } from './mobile-native-chat-image-attachment' @@ -32,6 +35,46 @@ export function updateMobileStructuredOutboxEntry( }) } +export function createMobileStructuredOutboxEntry(args: { + clientMessageId: string + sessionId: string + text: string + attachments: readonly PendingNativeChatImage[] + queuedAt: number +}): MobileStructuredOutboxEntry { + return { + clientMessageId: args.clientMessageId, + sessionId: args.sessionId, + body: mobileStructuredSendBody(args.text, args.attachments), + previewUris: args.attachments.map((attachment) => attachment.previewUri), + state: 'queued', + queuedAt: args.queuedAt, + lastAttemptAt: null, + retryAfterUnknownSubmittedAt: null + } +} + +export function reconcileMobileStructuredOutbox( + entries: readonly MobileStructuredOutboxEntry[], + submissions: readonly AgentJournalSubmission[] +): MobileStructuredOutboxEntry[] { + const settled = new Map(submissions.map((entry) => [entry.clientMessageId, entry])) + return entries.flatMap((entry) => { + const submission = settled.get(entry.clientMessageId) + if (submission?.dispatchState === 'accepted') { + return [] + } + if ( + submission?.dispatchState === 'unknown' && + entry.retryAfterUnknownSubmittedAt !== -1 && + entry.retryAfterUnknownSubmittedAt !== submission.submittedAt + ) { + return [{ ...entry, state: 'unconfirmed' as const }] + } + return [entry] + }) +} + export function isMobileStructuredDeliveryUnknown(error: unknown): boolean { return isRpcDeliveryUnknown(error) || isLogicalClientCutoverError(error) } diff --git a/mobile/src/session/mobile-structured-outbox-store.test.ts b/mobile/src/session/mobile-structured-outbox-store.test.ts index 85fb44ea0dd..e9da349e257 100644 --- a/mobile/src/session/mobile-structured-outbox-store.test.ts +++ b/mobile/src/session/mobile-structured-outbox-store.test.ts @@ -2,6 +2,7 @@ import AsyncStorage from '@react-native-async-storage/async-storage' import { beforeEach, describe, expect, it, vi } from 'vitest' import { loadMobileStructuredOutbox, + MAX_MOBILE_STRUCTURED_OUTBOX_ENTRIES, saveMobileStructuredOutbox, type MobileStructuredOutboxEntry } from './mobile-structured-outbox-store' @@ -17,7 +18,8 @@ const ENTRY: MobileStructuredOutboxEntry = { previewUris: [], state: 'queued', queuedAt: 1, - lastAttemptAt: null + lastAttemptAt: null, + retryAfterUnknownSubmittedAt: null } describe('mobile structured outbox store', () => { @@ -48,4 +50,18 @@ describe('mobile structured outbox store', () => { await expect(loadMobileStructuredOutbox('mobile_1')).resolves.toEqual([ENTRY]) }) + + it('rejects overflow instead of silently deleting older queued messages', async () => { + const entries = Array.from( + { length: MAX_MOBILE_STRUCTURED_OUTBOX_ENTRIES + 1 }, + (_, index) => ({ + ...ENTRY, + clientMessageId: `mobile-send:${index}:id`, + queuedAt: index + }) + ) + + await expect(saveMobileStructuredOutbox('mobile_1', entries)).rejects.toThrow('outbox is full') + expect(AsyncStorage.setItem).not.toHaveBeenCalled() + }) }) diff --git a/mobile/src/session/mobile-structured-outbox-store.ts b/mobile/src/session/mobile-structured-outbox-store.ts index 7916af6dd2d..ded8679af8d 100644 --- a/mobile/src/session/mobile-structured-outbox-store.ts +++ b/mobile/src/session/mobile-structured-outbox-store.ts @@ -11,10 +11,11 @@ export type MobileStructuredOutboxEntry = { state: MobileStructuredOutboxState queuedAt: number lastAttemptAt: number | null + retryAfterUnknownSubmittedAt: number | null } const STORAGE_PREFIX = 'orca:structuredAgentSessionOutbox:v1:' -const MAX_OUTBOX_ENTRIES = 64 +export const MAX_MOBILE_STRUCTURED_OUTBOX_ENTRIES = 64 function storageKey(sessionId: string): string { return STORAGE_PREFIX + encodeURIComponent(sessionId) @@ -47,7 +48,11 @@ function parseEntry(value: unknown, sessionId: string): MobileStructuredOutboxEn previewUris: entry.previewUris, state: entry.state as MobileStructuredOutboxState, queuedAt: entry.queuedAt, - lastAttemptAt: typeof entry.lastAttemptAt === 'number' ? entry.lastAttemptAt : null + lastAttemptAt: typeof entry.lastAttemptAt === 'number' ? entry.lastAttemptAt : null, + retryAfterUnknownSubmittedAt: + typeof entry.retryAfterUnknownSubmittedAt === 'number' + ? entry.retryAfterUnknownSubmittedAt + : null } } @@ -60,7 +65,7 @@ export async function loadMobileStructuredOutbox( return [] } return parsed - .slice(-MAX_OUTBOX_ENTRIES) + .slice(-MAX_MOBILE_STRUCTURED_OUTBOX_ENTRIES) .map((entry) => parseEntry(entry, sessionId)) .filter((entry): entry is MobileStructuredOutboxEntry => entry !== null) .map((entry) => @@ -76,9 +81,10 @@ export async function saveMobileStructuredOutbox( sessionId: string, entries: readonly MobileStructuredOutboxEntry[] ): Promise { - const bounded = entries - .filter((entry) => entry.sessionId === sessionId) - .slice(-MAX_OUTBOX_ENTRIES) + const bounded = entries.filter((entry) => entry.sessionId === sessionId) + if (bounded.length > MAX_MOBILE_STRUCTURED_OUTBOX_ENTRIES) { + throw new Error(`Structured outbox is full (${MAX_MOBILE_STRUCTURED_OUTBOX_ENTRIES} messages)`) + } if (bounded.length === 0) { await AsyncStorage.removeItem(storageKey(sessionId)) return diff --git a/mobile/src/session/mobile-structured-session-timeline.test.ts b/mobile/src/session/mobile-structured-session-timeline.test.ts index 77b4244d8fc..bb0b041df08 100644 --- a/mobile/src/session/mobile-structured-session-timeline.test.ts +++ b/mobile/src/session/mobile-structured-session-timeline.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest' import type { AgentJournalRenderItem } from '../../../src/shared/agent-session-journal-types' import { buildMobileStructuredTimeline, - latestMobileStructuredTurnId, + activeMobileStructuredTurnId, restoreMobileStructuredAttachments } from './mobile-structured-session-timeline' import type { MobileStructuredOutboxEntry } from './mobile-structured-outbox-store' @@ -35,7 +35,8 @@ const OUTBOX: MobileStructuredOutboxEntry = { previewUris: ['file:///preview.png'], state: 'unconfirmed', queuedAt: 3, - lastAttemptAt: 4 + lastAttemptAt: 4, + retryAfterUnknownSubmittedAt: null } describe('mobile structured session timeline', () => { @@ -61,12 +62,33 @@ describe('mobile structured session timeline', () => { ]) }) - it('derives the cancellable turn from the latest Codex item identity', () => { + it('shows cancellation only while the durable root-turn lifecycle is running', () => { + const lifecycle: AgentJournalRenderItem = { + itemId: 'legacy:codex:mobile_1:turn-lifecycle%3Aturn-7', + revision: 1, + sequence: 5, + observedAt: 1, + body: { + kind: 'status', + text: '', + turnLifecycle: { turnId: 'turn-7', state: 'running' } + } + } + expect(activeMobileStructuredTurnId([APPROVAL, lifecycle])).toBe('turn-7') expect( - latestMobileStructuredTurnId([ + activeMobileStructuredTurnId([ APPROVAL, - { ...APPROVAL, itemId: 'codex:thread-1:turn-7:1', sequence: 5 } + { + ...lifecycle, + revision: 2, + body: { + kind: 'status', + text: '', + turnLifecycle: { turnId: 'turn-7', state: 'completed' } + } + } ]) - ).toBe('turn-7') + ).toBeNull() + expect(buildMobileStructuredTimeline([lifecycle], [])).toEqual([]) }) }) diff --git a/mobile/src/session/mobile-structured-session-timeline.ts b/mobile/src/session/mobile-structured-session-timeline.ts index b362c26d5bb..f40b9c97ce6 100644 --- a/mobile/src/session/mobile-structured-session-timeline.ts +++ b/mobile/src/session/mobile-structured-session-timeline.ts @@ -1,5 +1,4 @@ import type { AgentJournalRenderItem } from '../../../src/shared/agent-session-journal-types' -import { parseAgentJournalItemKey } from '../../../src/shared/agent-session-journal-item-key' import type { NativeChatMessage } from '../../../src/shared/native-chat-types' import type { MobileStructuredPromptItem } from './MobileStructuredPromptCard' import type { PendingNativeChatImage } from './mobile-native-chat-image-attachment' @@ -63,13 +62,13 @@ export function buildMobileStructuredTimeline( ] } -export function latestMobileStructuredTurnId( +export function activeMobileStructuredTurnId( items: readonly AgentJournalRenderItem[] ): string | null { for (let index = items.length - 1; index >= 0; index -= 1) { - const identity = parseAgentJournalItemKey(items[index]?.itemId ?? '') - if (identity?.provider === 'codex') { - return identity.turnId + const body = items[index]?.body + if (body?.kind === 'status' && body.turnLifecycle?.state === 'running') { + return body.turnLifecycle.turnId } } return null diff --git a/mobile/src/session/mobile-structured-session-write-types.ts b/mobile/src/session/mobile-structured-session-write-types.ts new file mode 100644 index 00000000000..e7d0faa86c7 --- /dev/null +++ b/mobile/src/session/mobile-structured-session-write-types.ts @@ -0,0 +1,12 @@ +import type { PendingNativeChatImage } from './mobile-native-chat-image-attachment' +import type { MobileStructuredOutboxEntry } from './mobile-structured-outbox-store' +import type { MobileStructuredSessionMutations } from './use-mobile-structured-session-mutations' + +export type MobileStructuredSessionWrites = MobileStructuredSessionMutations & { + outbox: MobileStructuredOutboxEntry[] + hydrated: boolean + error: string | null + send: (text: string, attachments?: readonly PendingNativeChatImage[]) => Promise + takeQueuedForEdit: (clientMessageId: string) => Promise + retry: (clientMessageId: string) => Promise +} diff --git a/mobile/src/session/use-mobile-structured-attachments.test.ts b/mobile/src/session/use-mobile-structured-attachments.test.ts new file mode 100644 index 00000000000..e3e2034e908 --- /dev/null +++ b/mobile/src/session/use-mobile-structured-attachments.test.ts @@ -0,0 +1,62 @@ +import { createElement } from 'react' +import { act, create, type ReactTestRenderer } from 'react-test-renderer' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { RpcClient } from '../transport/rpc-client' +import { uploadMobileNativeChatImages } from './mobile-native-chat-image-attachment' +import { useMobileStructuredAttachments } from './use-mobile-structured-attachments' + +vi.mock('./mobile-native-chat-image-attachment', async (importOriginal) => { + const original = await importOriginal() + return { ...original, uploadMobileNativeChatImages: vi.fn() } +}) +vi.mock('./mobile-image-source-picker', () => ({ + ImageLibraryPermissionError: class ImageLibraryPermissionError extends Error {}, + pickMobileImages: vi.fn() +})) + +describe('useMobileStructuredAttachments', () => { + let renderer: ReactTestRenderer | null = null + let controller: ReturnType | null = null + const onError = vi.fn() + + function Probe(): null { + controller = useMobileStructuredAttachments({ + client: {} as RpcClient, + sessionId: 'mobile_1', + getConnectionId: async () => null, + onError + }) + return null + } + + beforeEach(() => { + globalThis.IS_REACT_ACT_ENVIRONMENT = true + onError.mockReset() + vi.mocked(uploadMobileNativeChatImages).mockReset() + act(() => { + renderer = create(createElement(Probe)) + }) + }) + + afterEach(() => { + act(() => renderer?.unmount()) + renderer = null + controller = null + }) + + it('retains completed uploads when a later selected image fails', async () => { + vi.mocked(uploadMobileNativeChatImages).mockImplementation(async (_source, deps) => { + deps.onImageUploaded?.({ path: '/host/first.png', previewUri: 'file:///first.png' }) + throw new Error('second upload failed') + }) + + await act(async () => { + await controller!.attach('library') + }) + + expect(controller!.attachments).toEqual([ + { id: 'img-1', path: '/host/first.png', previewUri: 'file:///first.png' } + ]) + expect(onError).toHaveBeenCalledWith('Attach failed') + }) +}) diff --git a/mobile/src/session/use-mobile-structured-attachments.ts b/mobile/src/session/use-mobile-structured-attachments.ts index 4f66079a241..b0d4699be0a 100644 --- a/mobile/src/session/use-mobile-structured-attachments.ts +++ b/mobile/src/session/use-mobile-structured-attachments.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useRef, useState } from 'react' +import { useCallback, useRef, useState } from 'react' import type { RpcClient } from '../transport/rpc-client' import { ImageLibraryPermissionError, @@ -23,15 +23,21 @@ export function useMobileStructuredAttachments(args: { remove: (id: string) => void clear: () => void } { - const [attachments, setAttachments] = useState([]) - const [attaching, setAttaching] = useState(false) + const [state, setState] = useState<{ + sessionId: string | null + attachments: PendingNativeChatImage[] + attaching: boolean + }>({ sessionId: args.sessionId, attachments: [], attaching: false }) const sessionRef = useRef(args.sessionId) + const generationRef = useRef(0) const idCounterRef = useRef(0) const { client, getConnectionId, onError, sessionId } = args - sessionRef.current = sessionId - useEffect(() => { - setAttachments([]) - }, [sessionId]) + if (sessionRef.current !== sessionId) { + sessionRef.current = sessionId + generationRef.current += 1 + } + const attachments = state.sessionId === sessionId ? state.attachments : [] + const attaching = state.sessionId === sessionId && state.attaching const attach = useCallback( async (source: MobileImageSource): Promise => { @@ -39,28 +45,59 @@ export function useMobileStructuredAttachments(args: { return } const targetSession = sessionId + const targetGeneration = generationRef.current try { - const uploaded = await uploadMobileNativeChatImages(source, { + await uploadMobileNativeChatImages(source, { client, getConnectionId, pickImages: pickMobileImages, - onUploadStart: () => setAttaching(true) + onUploadStart: () => { + if ( + generationRef.current === targetGeneration && + sessionRef.current === targetSession + ) { + setState((current) => ({ + sessionId: targetSession, + attachments: current.sessionId === targetSession ? current.attachments : [], + attaching: true + })) + } + }, + onImageUploaded: (uploaded) => { + if ( + generationRef.current === targetGeneration && + sessionRef.current === targetSession + ) { + setState((current) => ({ + sessionId: targetSession, + attachments: appendPendingNativeChatImages( + current.sessionId === targetSession ? current.attachments : [], + [uploaded], + idCounterRef + ), + attaching: true + })) + } + } }) - if (sessionRef.current === targetSession && uploaded.length > 0) { - setAttachments((current) => - appendPendingNativeChatImages(current, uploaded, idCounterRef) + } catch (error) { + if (generationRef.current === targetGeneration && sessionRef.current === targetSession) { + onError( + error instanceof ImageLibraryPermissionError + ? 'Photo permission denied' + : error instanceof Error && error.message === 'Clipboard image is too large' + ? 'Image too large to attach' + : 'Attach failed' ) } - } catch (error) { - onError( - error instanceof ImageLibraryPermissionError - ? 'Photo permission denied' - : error instanceof Error && error.message === 'Clipboard image is too large' - ? 'Image too large to attach' - : 'Attach failed' - ) } finally { - setAttaching(false) + if (generationRef.current === targetGeneration && sessionRef.current === targetSession) { + setState((current) => ({ + sessionId: targetSession, + attachments: current.sessionId === targetSession ? current.attachments : [], + attaching: false + })) + } } }, [attaching, client, getConnectionId, onError, sessionId] @@ -70,7 +107,18 @@ export function useMobileStructuredAttachments(args: { attachments, attaching, attach, - remove: (id) => setAttachments((current) => current.filter((entry) => entry.id !== id)), - clear: () => setAttachments([]) + remove: (id) => + setState((current) => ({ + sessionId, + attachments: + current.sessionId === sessionId + ? current.attachments.filter((entry) => entry.id !== id) + : [], + attaching: current.sessionId === sessionId && current.attaching + })), + clear: () => { + generationRef.current += 1 + setState({ sessionId, attachments: [], attaching: false }) + } } } diff --git a/mobile/src/session/use-mobile-structured-composer-state.test.ts b/mobile/src/session/use-mobile-structured-composer-state.test.ts new file mode 100644 index 00000000000..1fcb71fa01a --- /dev/null +++ b/mobile/src/session/use-mobile-structured-composer-state.test.ts @@ -0,0 +1,46 @@ +import { createElement, type ReactNode } from 'react' +import { act, create, type ReactTestRenderer } from 'react-test-renderer' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { useMobileStructuredComposerState } from './use-mobile-structured-composer-state' + +describe('useMobileStructuredComposerState', () => { + let renderer: ReactTestRenderer | null = null + let sessionId = 'mobile_1' + let state: ReturnType | null = null + + function Probe(): null { + state = useMobileStructuredComposerState() + return null + } + + function Harness(): ReactNode { + return createElement(Probe, { key: sessionId }) + } + + beforeEach(() => { + globalThis.IS_REACT_ACT_ENVIRONMENT = true + sessionId = 'mobile_1' + act(() => { + renderer = create(createElement(Harness)) + }) + }) + + afterEach(() => { + act(() => renderer?.unmount()) + renderer = null + state = null + }) + + it('clears draft text and restored attachments when the session changes', () => { + act(() => { + state!.setComposerText('do not leak') + state!.setRestored([{ id: 'restored:1', path: '/host/a.png', previewUri: 'file:///a.png' }]) + }) + + sessionId = 'mobile_2' + act(() => renderer!.update(createElement(Harness))) + + expect(state!.composerText).toBe('') + expect(state!.restored).toEqual([]) + }) +}) diff --git a/mobile/src/session/use-mobile-structured-composer-state.ts b/mobile/src/session/use-mobile-structured-composer-state.ts new file mode 100644 index 00000000000..e4fed4691a4 --- /dev/null +++ b/mobile/src/session/use-mobile-structured-composer-state.ts @@ -0,0 +1,14 @@ +import { useState } from 'react' +import type { PendingNativeChatImage } from './mobile-native-chat-image-attachment' + +export function useMobileStructuredComposerState(): { + composerText: string + setComposerText: React.Dispatch> + restored: PendingNativeChatImage[] + setRestored: React.Dispatch> +} { + const [composerText, setComposerText] = useState('') + const [restored, setRestored] = useState([]) + + return { composerText, setComposerText, restored, setRestored } +} diff --git a/mobile/src/session/use-mobile-structured-session-entry.ts b/mobile/src/session/use-mobile-structured-session-entry.ts index f5b247b3243..2a6d8a927c6 100644 --- a/mobile/src/session/use-mobile-structured-session-entry.ts +++ b/mobile/src/session/use-mobile-structured-session-entry.ts @@ -42,9 +42,9 @@ export function useMobileStructuredSessionEntry(args: { }) const writes = useMobileStructuredSessionWrites({ client, + connected, sessionId, fence: session.fence, - items: session.items, submissions: session.submissions }) const sessionOptions = useMobileStructuredSessionOptions({ diff --git a/mobile/src/session/use-mobile-structured-session-mutations.ts b/mobile/src/session/use-mobile-structured-session-mutations.ts index 9ab1cbc8273..3c22f852d74 100644 --- a/mobile/src/session/use-mobile-structured-session-mutations.ts +++ b/mobile/src/session/use-mobile-structured-session-mutations.ts @@ -71,7 +71,8 @@ export function useMobileStructuredSessionMutations(args: { return null } if (!response.ok) { - throw new Error(response.error.message) + onRefusal(response.error.message) + return null } const result = response.result as AgentSessionMutationResult if (!result.ok) { diff --git a/mobile/src/session/use-mobile-structured-session-options.test.ts b/mobile/src/session/use-mobile-structured-session-options.test.ts index f3d5310f013..2d09125fd84 100644 --- a/mobile/src/session/use-mobile-structured-session-options.test.ts +++ b/mobile/src/session/use-mobile-structured-session-options.test.ts @@ -7,16 +7,18 @@ import { useMobileStructuredSessionOptions } from './use-mobile-structured-sessi describe('useMobileStructuredSessionOptions', () => { let renderer: ReactTestRenderer | null = null let controller: MobileNativeChatSessionOptionsController | null = null + let sessionId = 'mobile_1' const setOption = vi.fn<(key: string, value: string) => Promise>() function Probe(): null { - controller = useMobileStructuredSessionOptions({ sessionId: 'mobile_1', setOption }) + controller = useMobileStructuredSessionOptions({ sessionId, setOption }) return null } beforeEach(() => { globalThis.IS_REACT_ACT_ENVIRONMENT = true setOption.mockReset().mockResolvedValue(true) + sessionId = 'mobile_1' act(() => { renderer = create(createElement(Probe)) }) @@ -56,4 +58,46 @@ describe('useMobileStructuredSessionOptions', () => { valueSource: 'unknown' }) }) + + it('does not apply an in-flight option result to a replacement session', async () => { + let resolve!: (applied: boolean) => void + setOption.mockReturnValueOnce(new Promise((done) => (resolve = done))) + let oldResult!: boolean + let pending!: Promise + act(() => { + pending = controller!.setOption('model', 'gpt-5.6-sol') + }) + sessionId = 'mobile_2' + act(() => renderer!.update(createElement(Probe))) + await act(async () => { + resolve(true) + oldResult = await pending + }) + + expect(oldResult).toBe(false) + expect(controller!.pendingId).toBeNull() + expect(controller!.snapshot.find((entry) => entry.id === 'model')).toMatchObject({ + valueSource: 'unknown' + }) + }) + + it('does not restore stale pending state when returning to a prior session', async () => { + const oldRequest = new Promise(() => {}) + setOption.mockReturnValueOnce(oldRequest).mockResolvedValueOnce(true) + act(() => { + void controller!.setOption('model', 'gpt-5.6-sol') + }) + sessionId = 'mobile_2' + act(() => renderer!.update(createElement(Probe))) + sessionId = 'mobile_1' + act(() => renderer!.update(createElement(Probe))) + + let applied!: boolean + await act(async () => { + applied = await controller!.setOption('effort', 'high') + }) + + expect(applied).toBe(true) + expect(setOption).toHaveBeenNthCalledWith(2, 'effort', 'high') + }) }) diff --git a/mobile/src/session/use-mobile-structured-session-options.ts b/mobile/src/session/use-mobile-structured-session-options.ts index 04cf92eece5..bb007bca693 100644 --- a/mobile/src/session/use-mobile-structured-session-options.ts +++ b/mobile/src/session/use-mobile-structured-session-options.ts @@ -19,13 +19,18 @@ export function useMobileStructuredSessionOptions(args: { }): MobileNativeChatSessionOptionsController { const { sessionId, setOption: dispatchOption } = args const [version, setVersion] = useState(0) - const [pendingId, setPendingId] = useState(null) + const [pending, setPending] = useState<{ + sessionId: string + id: string + record: ReturnType + } | null>(null) const recordRef = useRef(createNativeChatSessionOptionRecord('codex')) const sessionRef = useRef(sessionId) if (sessionRef.current !== sessionId) { sessionRef.current = sessionId recordRef.current = createNativeChatSessionOptionRecord('codex') } + const pendingId = pending?.record === recordRef.current ? pending.id : null const snapshot = useMemo(() => { void version @@ -46,22 +51,30 @@ export function useMobileStructuredSessionOptions(args: { if (!CODEX_CATALOG || !sessionId || typeof value !== 'string' || pendingId !== null) { return false } - setPendingId(id) + const targetSessionId = sessionId + const targetRecord = recordRef.current + setPending({ sessionId: targetSessionId, id, record: targetRecord }) try { const applied = await dispatchOption(id, value) - if (!applied) { + if ( + !applied || + sessionRef.current !== targetSessionId || + recordRef.current !== targetRecord + ) { return false } const effectiveModel = resolveEffectiveNativeChatModelId( CODEX_CATALOG, CODEX_CATALOG.models, - recordRef.current + targetRecord ) - setTrackedSessionOption(recordRef.current, id, value, 'dispatched', effectiveModel) + setTrackedSessionOption(targetRecord, id, value, 'dispatched', effectiveModel) setVersion((current) => current + 1) return true } finally { - setPendingId(null) + setPending((current) => + current?.record === targetRecord && current.id === id ? null : current + ) } }, [dispatchOption, pendingId, sessionId] 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 4b3070eca7d..482b22120eb 100644 --- a/mobile/src/session/use-mobile-structured-session-writes.test.ts +++ b/mobile/src/session/use-mobile-structured-session-writes.test.ts @@ -3,6 +3,7 @@ import { act, create, type ReactTestRenderer } from 'react-test-renderer' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { RpcClient } from '../transport/rpc-client' import type { RpcSuccess } from '../transport/types' +import type { AgentJournalSubmission } from '../../../src/shared/agent-session-journal-types' import { loadMobileStructuredOutbox, saveMobileStructuredOutbox @@ -61,14 +62,17 @@ describe('useMobileStructuredSessionWrites', () => { sendRequest, getState: () => 'connected' as const } as RpcClient + let connected = true + let fence = 3 + let submissions: AgentJournalSubmission[] = [] function Probe(): null { api = useMobileStructuredSessionWrites({ client, + connected, sessionId: 'mobile_1', - fence: 3, - items: [], - submissions: [] + fence, + submissions }) return null } @@ -78,6 +82,9 @@ describe('useMobileStructuredSessionWrites', () => { vi.mocked(loadMobileStructuredOutbox).mockReset().mockResolvedValue([]) vi.mocked(saveMobileStructuredOutbox).mockReset().mockResolvedValue(undefined) sendRequest.mockReset() + connected = true + fence = 3 + submissions = [] const crypto = await import('expo-crypto') vi.mocked(crypto.randomUUID) .mockReset() @@ -121,26 +128,27 @@ describe('useMobileStructuredSessionWrites', () => { expect(sendRequest).toHaveBeenCalledTimes(2) }) - it('keeps unknown delivery visible and retries with the same client message id', async () => { - sendRequest - .mockResolvedValueOnce({ - ...accepted('unused'), - result: { - ...(accepted('unused').result as object), - value: { - ...(accepted('unused').result as { value: object }).value, - submission: { - ...(accepted('unused').result as { value: { submission: object } }).value.submission, - dispatchState: 'unknown' + it('retries a durable unknown by the same id before dispatching later messages', async () => { + sendRequest.mockImplementation(async (_method, params) => { + const id = (params as { envelope: { clientOperationId: string } }).envelope.clientOperationId + if (sendRequest.mock.calls.length === 1) { + const response = accepted(id) + return { + ...response, + result: { + ...(response.result as object), + value: { + ...(response.result as { value: object }).value, + submission: { + ...(response.result as { value: { submission: object } }).value.submission, + dispatchState: 'unknown' + } } } } - }) - .mockImplementationOnce(async (_method, params) => { - const id = (params as { envelope: { clientOperationId: string } }).envelope - .clientOperationId - return accepted(id) - }) + } + return accepted(id) + }) await act(async () => { await api!.send('possibly delivered') @@ -149,6 +157,25 @@ describe('useMobileStructuredSessionWrites', () => { }) expect(api!.outbox[0]?.state).toBe('unconfirmed') const firstId = api!.outbox[0]!.clientMessageId + submissions = [ + { + clientMessageId: firstId, + fence: 3, + payloadFingerprint: 'a'.repeat(64), + dispatchState: 'unknown', + providerItemId: null, + reason: 'reply lost', + submittedAt: 11, + resolvedAt: 12 + } + ] + + await act(async () => { + renderer!.update(createElement(Probe)) + await api!.send('later') + await Promise.resolve() + }) + expect(sendRequest).toHaveBeenCalledTimes(1) await act(async () => { await api!.retry(firstId) @@ -159,6 +186,117 @@ describe('useMobileStructuredSessionWrites', () => { const retryId = (sendRequest.mock.calls[1]![1] as { envelope: { clientOperationId: string } }) .envelope.clientOperationId expect(retryId).toBe(firstId) + await act(async () => { + await Promise.resolve() + await Promise.resolve() + }) + expect(sendRequest).toHaveBeenCalledTimes(3) + }) + + it('unblocks a refused queued head when the runtime fence advances', async () => { + sendRequest + .mockResolvedValueOnce({ + ...accepted('refused'), + result: { + ok: false, + refusal: { code: 'agent_session_fence_stale', message: 'stale fence', retryable: true } + } + }) + .mockImplementationOnce(async (_method, params) => { + const id = (params as { envelope: { clientOperationId: string } }).envelope + .clientOperationId + return accepted(id) + }) + + await act(async () => { + await api!.send('retry after reconnect') + await Promise.resolve() + await Promise.resolve() + }) + expect(sendRequest).toHaveBeenCalledTimes(1) + expect(api!.outbox[0]?.state).toBe('queued') + + fence = 4 + await act(async () => { + renderer!.update(createElement(Probe)) + await Promise.resolve() + await Promise.resolve() + }) + + expect(sendRequest).toHaveBeenCalledTimes(2) + expect( + (sendRequest.mock.calls[1]![1] as { envelope: { expectedRuntimeFence: number } }).envelope + .expectedRuntimeFence + ).toBe(4) + }) + + it('requeues an in-flight stale-fence response under the replacement fence', async () => { + const oldResponse = deferred() + sendRequest + .mockImplementationOnce(() => oldResponse.promise) + .mockImplementationOnce(async (_method, params) => { + const id = (params as { envelope: { clientOperationId: string } }).envelope + .clientOperationId + return accepted(id) + }) + + await act(async () => { + await api!.send('survive reattach') + await Promise.resolve() + }) + fence = 4 + await act(async () => { + renderer!.update(createElement(Probe)) + oldResponse.resolve({ + ...accepted('stale'), + result: { + ok: false, + refusal: { code: 'agent_session_fence_stale', message: 'stale fence', retryable: true } + } + }) + await oldResponse.promise + await Promise.resolve() + await Promise.resolve() + }) + + expect(sendRequest).toHaveBeenCalledTimes(2) + expect( + (sendRequest.mock.calls[1]![1] as { envelope: { expectedRuntimeFence: number } }).envelope + .expectedRuntimeFence + ).toBe(4) + }) + + it('retries an unresolved dispatch after a same-fence reconnect', async () => { + const oldResponse = deferred() + sendRequest + .mockImplementationOnce(() => oldResponse.promise) + .mockImplementationOnce(async (_method, params) => { + const id = (params as { envelope: { clientOperationId: string } }).envelope + .clientOperationId + return accepted(id) + }) + + await act(async () => { + await api!.send('survive reconnect') + await Promise.resolve() + }) + connected = false + act(() => renderer!.update(createElement(Probe))) + connected = true + await act(async () => { + renderer!.update(createElement(Probe)) + await Promise.resolve() + await Promise.resolve() + }) + + expect(sendRequest).toHaveBeenCalledTimes(2) + expect(api!.outbox).toEqual([]) + + await act(async () => { + oldResponse.resolve(accepted('stale')) + await oldResponse.promise + }) + expect(api!.outbox).toEqual([]) }) it('answers a durable approval with its expected revision compare-and-set', async () => { @@ -216,4 +354,21 @@ describe('useMobileStructuredSessionWrites', () => { }) ) }) + + it('turns an RPC failure response into a handled mutation error', async () => { + sendRequest.mockResolvedValue({ + id: 'set-option', + ok: false, + _meta: { runtimeId: 'runtime-1' }, + error: { code: 'forbidden', message: 'not allowed' } + }) + + let applied!: boolean + await act(async () => { + applied = await api!.setOption('model', 'gpt-5.6-sol') + }) + + expect(applied).toBe(false) + expect(api!.error).toBe('not allowed') + }) }) diff --git a/mobile/src/session/use-mobile-structured-session-writes.ts b/mobile/src/session/use-mobile-structured-session-writes.ts index ee241e557e7..5e8f51d0f9c 100644 --- a/mobile/src/session/use-mobile-structured-session-writes.ts +++ b/mobile/src/session/use-mobile-structured-session-writes.ts @@ -1,9 +1,6 @@ import * as ExpoCrypto from 'expo-crypto' import { useCallback, useEffect, useMemo, useRef, useState } from 'react' -import type { - AgentJournalRenderItem, - AgentJournalSubmission -} from '../../../src/shared/agent-session-journal-types' +import type { AgentJournalSubmission } from '../../../src/shared/agent-session-journal-types' import type { AgentSessionMutationResult, AgentSessionSendResult @@ -11,41 +8,33 @@ import type { import type { RpcClient } from '../transport/rpc-client' import type { PendingNativeChatImage } from './mobile-native-chat-image-attachment' import { + createMobileStructuredOutboxEntry, isMobileStructuredDeliveryUnknown, - mobileStructuredSendBody, + reconcileMobileStructuredOutbox, updateMobileStructuredOutboxEntry } from './mobile-structured-outbox-entry' import { createMobileStructuredOperationId, - mobileStructuredPayloadFingerprint + mobileStructuredSendRequest } from './mobile-structured-mutation-envelope' import { loadMobileStructuredOutbox, saveMobileStructuredOutbox, type MobileStructuredOutboxEntry } from './mobile-structured-outbox-store' -import { - useMobileStructuredSessionMutations, - type MobileStructuredSessionMutations -} from './use-mobile-structured-session-mutations' +import { useMobileStructuredSessionMutations } from './use-mobile-structured-session-mutations' +import type { MobileStructuredSessionWrites } from './mobile-structured-session-write-types' -export type MobileStructuredSessionWrites = MobileStructuredSessionMutations & { - outbox: MobileStructuredOutboxEntry[] - hydrated: boolean - error: string | null - send: (text: string, attachments?: readonly PendingNativeChatImage[]) => Promise - takeQueuedForEdit: (clientMessageId: string) => Promise - retry: (clientMessageId: string) => Promise -} +export type { MobileStructuredSessionWrites } export function useMobileStructuredSessionWrites(args: { client: RpcClient | null + connected: boolean sessionId: string | null fence: number | null - items: readonly AgentJournalRenderItem[] submissions: readonly AgentJournalSubmission[] }): MobileStructuredSessionWrites { - const { client, sessionId, fence, submissions } = args + const { client, connected, sessionId, fence, submissions } = args const [outbox, setOutbox] = useState([]) const [hydrated, setHydrated] = useState(false) const [error, setError] = useState(null) @@ -54,9 +43,12 @@ export function useMobileStructuredSessionWrites(args: { outboxRef.current = outbox const persistTailRef = useRef>(Promise.resolve()) const dispatchingRef = useRef(false) + const dispatchGenerationRef = useRef(0) const blockedIdRef = useRef(null) const activeSessionRef = useRef(sessionId) activeSessionRef.current = sessionId + const activeFenceRef = useRef(fence) + activeFenceRef.current = fence const mutations = useMobileStructuredSessionMutations({ client, sessionId, @@ -73,7 +65,10 @@ export function useMobileStructuredSessionWrites(args: { }, []) useEffect(() => { + dispatchGenerationRef.current += 1 + dispatchingRef.current = false setOutbox([]) + outboxRef.current = [] setHydrated(false) setError(null) blockedIdRef.current = null @@ -97,19 +92,7 @@ export function useMobileStructuredSessionWrites(args: { if (!sessionId || submissions.length === 0 || outboxRef.current.length === 0) { return } - const settled = new Map( - submissions.map((submission) => [submission.clientMessageId, submission]) - ) - const next = outboxRef.current.flatMap((entry) => { - const submission = settled.get(entry.clientMessageId) - if (submission?.dispatchState === 'accepted') { - return [] - } - if (submission?.dispatchState === 'unknown') { - return [{ ...entry, state: 'unconfirmed' as const }] - } - return [entry] - }) + const next = reconcileMobileStructuredOutbox(outboxRef.current, submissions) if ( next.length !== outboxRef.current.length || next.some((entry, index) => entry !== outboxRef.current[index]) @@ -120,6 +103,26 @@ export function useMobileStructuredSessionWrites(args: { } }, [persist, sessionId, submissions]) + useEffect(() => { + dispatchGenerationRef.current += 1 + dispatchingRef.current = false + blockedIdRef.current = null + if (sessionId) { + const current = outboxRef.current + const next = current.map((entry) => + entry.state === 'dispatching' ? { ...entry, state: 'queued' as const } : entry + ) + if (next.some((entry, index) => entry !== current[index])) { + outboxRef.current = next + setOutbox(next) + void persist(sessionId, next) + } + } + if (connected) { + setDispatchVersion((current) => current + 1) + } + }, [connected, fence, persist, sessionId]) + const replaceEntry = useCallback( async ( id: string, @@ -137,9 +140,11 @@ export function useMobileStructuredSessionWrites(args: { ) useEffect(() => { - const next = outbox.find((entry) => entry.state === 'queued') + const head = outbox[0] + const next = head?.state === 'queued' ? head : null if ( !client || + !connected || !sessionId || fence === null || !hydrated || @@ -151,7 +156,9 @@ export function useMobileStructuredSessionWrites(args: { return } dispatchingRef.current = true + const targetGeneration = dispatchGenerationRef.current const targetSessionId = sessionId + const targetFence = fence void (async () => { try { await replaceEntry(next.clientMessageId, (entry) => ({ @@ -159,23 +166,20 @@ export function useMobileStructuredSessionWrites(args: { state: 'dispatching', lastAttemptAt: Date.now() })) - const fields = { body: next.body } - const response = await client.sendRequest('agentSession.send', { - envelope: { - sessionId, - clientOperationId: next.clientMessageId, - expectedRuntimeFence: fence, - payloadFingerprint: mobileStructuredPayloadFingerprint({ - method: 'agentSession.send', - sessionId, - fields - }) - }, - ...fields - }) + const response = await client.sendRequest( + 'agentSession.send', + mobileStructuredSendRequest(next, fence) + ) if (activeSessionRef.current !== targetSessionId) { return } + if (dispatchGenerationRef.current !== targetGeneration) { + return + } + if (activeFenceRef.current !== targetFence) { + await replaceEntry(next.clientMessageId, (entry) => ({ ...entry, state: 'queued' })) + return + } if (!response.ok) { throw new Error(response.error.message) } @@ -187,7 +191,11 @@ export function useMobileStructuredSessionWrites(args: { return } if (result.value.submission.dispatchState === 'unknown') { - await replaceEntry(next.clientMessageId, (entry) => ({ ...entry, state: 'unconfirmed' })) + await replaceEntry(next.clientMessageId, (entry) => ({ + ...entry, + state: 'unconfirmed', + retryAfterUnknownSubmittedAt: null + })) return } if (result.value.submission.dispatchState === 'rejected') { @@ -202,36 +210,47 @@ export function useMobileStructuredSessionWrites(args: { if (activeSessionRef.current !== targetSessionId) { return } + if (dispatchGenerationRef.current !== targetGeneration) { + return + } + if (activeFenceRef.current !== targetFence) { + await replaceEntry(next.clientMessageId, (entry) => ({ ...entry, state: 'queued' })) + return + } if (isMobileStructuredDeliveryUnknown(caught)) { - await replaceEntry(next.clientMessageId, (entry) => ({ ...entry, state: 'unconfirmed' })) + await replaceEntry(next.clientMessageId, (entry) => ({ + ...entry, + state: 'unconfirmed', + retryAfterUnknownSubmittedAt: null + })) } else { blockedIdRef.current = next.clientMessageId setError(caught instanceof Error ? caught.message : 'Message was not sent') await replaceEntry(next.clientMessageId, (entry) => ({ ...entry, state: 'queued' })) } } finally { - dispatchingRef.current = false - setDispatchVersion((current) => current + 1) + if (dispatchGenerationRef.current === targetGeneration) { + dispatchingRef.current = false + setDispatchVersion((current) => current + 1) + } } })() - }, [client, dispatchVersion, fence, hydrated, outbox, replaceEntry, sessionId]) + }, [client, connected, dispatchVersion, fence, hydrated, outbox, replaceEntry, sessionId]) const send = useCallback( async (text: string, attachments: readonly PendingNativeChatImage[] = []): Promise => { if (!sessionId || (!text.trim() && attachments.length === 0)) { return false } - const entry: MobileStructuredOutboxEntry = { + const entry = createMobileStructuredOutboxEntry({ clientMessageId: createMobileStructuredOperationId('mobile-send', () => ExpoCrypto.randomUUID() ), sessionId, - body: mobileStructuredSendBody(text, attachments), - previewUris: attachments.map((attachment) => attachment.previewUri), - state: 'queued', - queuedAt: Date.now(), - lastAttemptAt: null - } + text, + attachments, + queuedAt: Date.now() + }) const next = [...outboxRef.current, entry] try { await persist(sessionId, next) @@ -264,9 +283,17 @@ export function useMobileStructuredSessionWrites(args: { async (clientMessageId: string): Promise => { blockedIdRef.current = null setError(null) - await replaceEntry(clientMessageId, (entry) => ({ ...entry, state: 'queued' })) + const unknown = submissions.find( + (submission) => + submission.clientMessageId === clientMessageId && submission.dispatchState === 'unknown' + ) + await replaceEntry(clientMessageId, (entry) => ({ + ...entry, + state: 'queued', + retryAfterUnknownSubmittedAt: unknown?.submittedAt ?? -1 + })) }, - [replaceEntry] + [replaceEntry, submissions] ) return useMemo( diff --git a/src/main/codex/codex-structured-journal-translation.test.ts b/src/main/codex/codex-structured-journal-translation.test.ts index a36b10ce759..d69910f0510 100644 --- a/src/main/codex/codex-structured-journal-translation.test.ts +++ b/src/main/codex/codex-structured-journal-translation.test.ts @@ -20,12 +20,13 @@ type Row = { key: string; body: AgentJournalItemBody } function recorder() { const rows: Row[] = [] + const tombstones: string[] = [] const bound: [string, string, string][] = [] let publishes = 0 const sink: StructuredAgentSessionEventSink = { appendItem: (identity: AgentJournalItemIdentity, body) => rows.push({ key: agentJournalItemKey(identity), body }), - appendTombstone: () => {}, + appendTombstone: (identity) => tombstones.push(agentJournalItemKey(identity)), publish: () => { publishes += 1 } @@ -33,6 +34,7 @@ function recorder() { return { sink, rows, + tombstones, bound, publishes: () => publishes, bindPromptItemId: (journalItemId: string, threadId: string, promptKey: string) => @@ -75,6 +77,29 @@ function translatorWith(tap = recorder(), window = manualWindow()) { } describe('codex journal translation', () => { + it('durably opens and closes the primary turn cancellation lifecycle', () => { + const tap = recorder() + const translator = createCodexJournalTranslator({ + sink: tap.sink, + primaryThreadId: () => THREAD_ID + }) + + translator.handle(TURN_STARTED) + translator.handle(notification('turn/completed', { turn: { id: TURN_ID } })) + + expect(tap.rows).toEqual([ + { + key: 'legacy:codex:session-1:turn-lifecycle%3Aturn-1', + body: { + kind: 'status', + text: 'Codex is working…', + turnLifecycle: { turnId: TURN_ID, state: 'running' } + } + } + ]) + expect(tap.tombstones).toEqual(['legacy:codex:session-1:turn-lifecycle%3Aturn-1']) + }) + it('journals a user turn and the assistant answer under durable codex keys', () => { const { translator, tap } = translatorWith() diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 2eed0a17e72..54957121da8 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -34,6 +34,7 @@ export type CodexJournalTranslatorDeps = { sink: StructuredAgentSessionEventSink /** Points an answered journal item back at the live Codex request. */ bindPromptItemId?: (journalItemId: string, threadId: string, promptKey: string) => void + primaryThreadId?: () => string | null coalesceMs?: number schedule?: AgentSessionDeltaCoalescerDeps['schedule'] } @@ -66,6 +67,33 @@ export function createCodexJournalTranslator( const latestStreamText = new Map() const checkpointLengths = new Map() + const publishTurnLifecycle = ( + sessionId: string, + threadId: string, + turnId: string, + state: 'running' | 'completed' + ): void => { + if (deps.primaryThreadId?.() !== threadId) { + return + } + const identity = { + provider: 'legacy' as const, + agent: 'codex' as const, + sessionId, + recordId: `turn-lifecycle:${turnId}` + } + if (state === 'completed') { + deps.sink.appendTombstone(identity) + } else { + deps.sink.appendItem(identity, { + kind: 'status', + text: 'Codex is working…', + turnLifecycle: { turnId, state } + }) + } + deps.sink.publish() + } + const itemKey = (threadId: string, codexItemId: string): string => `${encodeURIComponent(threadId)}:${encodeURIComponent(codexItemId)}` @@ -192,6 +220,10 @@ export function createCodexJournalTranslator( handle: (event) => { if (event.type === 'ended') { flushStreams() + for (const [threadId, turnId] of currentTurnIds) { + publishTurnLifecycle(event.sessionId, threadId, turnId, 'completed') + } + currentTurnIds.clear() return } if (event.type === 'notification' && event.method === AGENT_MESSAGE_DELTA_METHOD) { @@ -213,10 +245,15 @@ export function createCodexJournalTranslator( const turnId = readCodexTurnId(event.params) if (turnId) { currentTurnIds.set(event.threadId, turnId) + publishTurnLifecycle(event.sessionId, event.threadId, turnId, 'running') } return } if (event.method === 'turn/completed') { + const turnId = readCodexTurnId(event.params) ?? currentTurnIds.get(event.threadId) + if (turnId) { + publishTurnLifecycle(event.sessionId, event.threadId, turnId, 'completed') + } // A later item with no turn of its own belongs to no turn, not to the // one that just ended. currentTurnIds.delete(event.threadId) diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 33d4d3427a3..63344a93bda 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -69,11 +69,14 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap const launch = await this.deps.resolveLaunch({ identity: input.identity }) const attempt = createCodexAcquisitionAttempt() const acquisition = attempt.window + let primaryThreadId = + input.identity.providerHandle.kind === 'codex' ? input.identity.providerHandle.threadId : null // Registered before the spawn, because the handshake itself can emit. this.acquiring.set(sessionId, attempt) const translator = input.events ? createCodexJournalTranslator({ sink: input.events, + primaryThreadId: () => primaryThreadId, bindPromptItemId: (journalItemId, threadId, promptKey) => acquisition.prompts.bindJournalItemId(journalItemId, threadId, promptKey) }) @@ -104,6 +107,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap ) acquisition.connection = connection const opened = await openCodexThread(connection, launch, this.requestTimeoutMs) + primaryThreadId = opened.threadId const acquired: AgentSessionAcquisition = { process: await codexProcessIdentity( { ...input, pid: connection.pid }, diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 536df076de5..84ce82f85d6 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -133,6 +133,9 @@ export type AgentJournalQuestionItem = { export type AgentJournalStatusItem = { kind: 'status' text: string + /** Durable root-turn lifecycle used by clients to expose cancellation only + * while the provider can still accept it. */ + turnLifecycle?: { turnId: string; state: 'running' | 'completed' } } export type AgentJournalItemBody =