Harden structured mobile writes before RC

This commit is contained in:
Brennan Benson
2026-08-09 19:58:58 -07:00
parent bbc6aba927
commit 8778fdedad
29 changed files with 904 additions and 184 deletions
@@ -4634,6 +4634,7 @@ export default function SessionScreen() {
</View>
) : activeStructuredTab ? (
<MobileStructuredAgentSessionView
key={activeStructuredTab.sessionId}
items={structuredSessionEntry.session.items}
status={structuredSessionEntry.session.status}
error={structuredSessionEntry.session.error}
@@ -0,0 +1,53 @@
import { createElement } from 'react'
import { TextInput } from 'react-native'
import { act, create, type ReactTestRenderer } from 'react-test-renderer'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import { MobileNativeChatQuestion } from './MobileNativeChatQuestion'
vi.mock('react-native', () => ({
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)
})
})
+36 -29
View File
@@ -7,13 +7,18 @@ import { formatQuestionAnswer, type MobileChatQuestion } from './mobile-native-c
type Props = {
question: MobileChatQuestion
onAnswer: (text: string) => Promise<boolean>
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<string[]>([])
const [freeText, setFreeText] = useState('')
const [sending, setSending] = useState(false)
@@ -124,35 +129,37 @@ export function MobileNativeChatQuestion({ question, onAnswer }: Props): React.J
</Pressable>
) : null}
<View style={styles.freeTextRow}>
<TextInput
style={styles.freeInput}
value={freeText}
onChangeText={setFreeText}
placeholder={hasOptions ? 'Or type a reply…' : 'Type your reply…'}
placeholderTextColor={colors.textMuted}
selectionColor={colors.accentBlue}
onSubmitEditing={submitFreeText}
returnKeyType="send"
multiline
/>
<Pressable
accessibilityLabel="Send reply"
style={({ pressed }) => [
styles.freeSend,
!canSendFreeText && styles.freeSendDisabled,
pressed && canSendFreeText && styles.pressed
]}
onPress={submitFreeText}
disabled={!canSendFreeText}
>
<ArrowUp
size={18}
color={canSendFreeText ? colors.bgBase : colors.textMuted}
strokeWidth={2.6}
{allowFreeText ? (
<View style={styles.freeTextRow}>
<TextInput
style={styles.freeInput}
value={freeText}
onChangeText={setFreeText}
placeholder={hasOptions ? 'Or type a reply…' : 'Type your reply…'}
placeholderTextColor={colors.textMuted}
selectionColor={colors.accentBlue}
onSubmitEditing={submitFreeText}
returnKeyType="send"
multiline
/>
</Pressable>
</View>
<Pressable
accessibilityLabel="Send reply"
style={({ pressed }) => [
styles.freeSend,
!canSendFreeText && styles.freeSendDisabled,
pressed && canSendFreeText && styles.pressed
]}
onPress={submitFreeText}
disabled={!canSendFreeText}
>
<ArrowUp
size={18}
color={canSendFreeText ? colors.bgBase : colors.textMuted}
strokeWidth={2.6}
/>
</Pressable>
</View>
) : null}
</View>
)
}
@@ -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<PendingNativeChatImage[]>([])
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
@@ -44,6 +44,7 @@ export function MobileStructuredPromptCard(props: {
return (
<MobileNativeChatQuestion
question={question}
allowFreeText={Boolean(questionBody.freeTextQuestionId)}
onAnswer={(answer) => {
const offered = questionBody.options.find((option) => option.label === answer)
const optionId =
@@ -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'])
})
})
@@ -68,12 +68,15 @@ export function shouldDownsampleMobileImage(input: {
)
}
async function downsampleMobileImage(input: {
uri: string
fileSize?: number
width?: number
height?: number
}): Promise<PreparedMobileImage | null> {
export async function prepareMobileImageForUpload(
input: {
uri: string
fileSize?: number
width?: number
height?: number
},
deletePreparedFile: (uri: string) => void = (uri) => new FsFile(uri).delete()
): Promise<PreparedMobileImage | null> {
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<PickedMobileImage> {
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<PickedMobileImage> {
const result = await launch({
type: 'image/*',
@@ -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 }] }
}
@@ -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<string, unknown> {
const fields = { body: entry.body }
return {
envelope: {
sessionId: entry.sessionId,
clientOperationId: entry.clientMessageId,
expectedRuntimeFence,
payloadFingerprint: mobileStructuredPayloadFingerprint({
method: 'agentSession.send',
sessionId: entry.sessionId,
fields
})
},
...fields
}
}
@@ -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)
}
@@ -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()
})
})
@@ -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<void> {
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
@@ -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([])
})
})
@@ -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
@@ -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<boolean>
takeQueuedForEdit: (clientMessageId: string) => Promise<MobileStructuredOutboxEntry | null>
retry: (clientMessageId: string) => Promise<void>
}
@@ -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<typeof import('./mobile-native-chat-image-attachment')>()
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<typeof useMobileStructuredAttachments> | 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')
})
})
@@ -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<PendingNativeChatImage[]>([])
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<void> => {
@@ -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 })
}
}
}
@@ -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<typeof useMobileStructuredComposerState> | 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([])
})
})
@@ -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<React.SetStateAction<string>>
restored: PendingNativeChatImage[]
setRestored: React.Dispatch<React.SetStateAction<PendingNativeChatImage[]>>
} {
const [composerText, setComposerText] = useState('')
const [restored, setRestored] = useState<PendingNativeChatImage[]>([])
return { composerText, setComposerText, restored, setRestored }
}
@@ -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({
@@ -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<TValue>
if (!result.ok) {
@@ -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<boolean>>()
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<boolean>((done) => (resolve = done)))
let oldResult!: boolean
let pending!: Promise<boolean>
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<boolean>(() => {})
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')
})
})
@@ -19,13 +19,18 @@ export function useMobileStructuredSessionOptions(args: {
}): MobileNativeChatSessionOptionsController {
const { sessionId, setOption: dispatchOption } = args
const [version, setVersion] = useState(0)
const [pendingId, setPendingId] = useState<string | null>(null)
const [pending, setPending] = useState<{
sessionId: string
id: string
record: ReturnType<typeof createNativeChatSessionOptionRecord>
} | 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]
@@ -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<RpcSuccess>()
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<RpcSuccess>()
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')
})
})
@@ -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<boolean>
takeQueuedForEdit: (clientMessageId: string) => Promise<MobileStructuredOutboxEntry | null>
retry: (clientMessageId: string) => Promise<void>
}
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<MobileStructuredOutboxEntry[]>([])
const [hydrated, setHydrated] = useState(false)
const [error, setError] = useState<string | null>(null)
@@ -54,9 +43,12 @@ export function useMobileStructuredSessionWrites(args: {
outboxRef.current = outbox
const persistTailRef = useRef<Promise<void>>(Promise.resolve())
const dispatchingRef = useRef(false)
const dispatchGenerationRef = useRef(0)
const blockedIdRef = useRef<string | null>(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<boolean> => {
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<void> => {
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(
@@ -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()
@@ -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<string, string>()
const checkpointLengths = new Map<string, number>()
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)
@@ -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 },
@@ -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 =