diff --git a/apps/mobile/src/Stack.tsx b/apps/mobile/src/Stack.tsx index 3aedb4e4e3e..76a9399772d 100644 --- a/apps/mobile/src/Stack.tsx +++ b/apps/mobile/src/Stack.tsx @@ -284,6 +284,15 @@ function workspacePathFromState(state: NavigationState): string { return path.startsWith("/") ? path : `/${path}`; } +// The drain hook subscribes to the outbox, all thread shells, projects, and +// connection statuses. Hosting it in a null-rendering leaf keeps those +// updates from re-rendering RootStackLayout (and with it every screen) on +// each enqueue, shell change, or reconnect. +function ThreadOutboxDrainWorker() { + useThreadOutboxDrain(); + return null; +} + function RootStackLayout(props: { readonly children: React.ReactNode; readonly state: NavigationState; @@ -292,7 +301,6 @@ function RootStackLayout(props: { const { pendingShare } = useIncomingShare(); const sharePresentationRef = useRef(EMPTY_INCOMING_SHARE_PRESENTATION_STATE); useAgentNotificationNavigation(); - useThreadOutboxDrain(); // Presents the T3 Connect onboarding sheet after an in-session sign-in. useConnectOnboardingNavigation(); // Launcher app shortcuts: routes shortcut taps and tracks opened threads. @@ -320,6 +328,7 @@ function RootStackLayout(props: { return ( + diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index 0263e74eade..b96cc862b27 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -665,6 +665,26 @@ export function HomeScreen(props: HomeScreenProps) { ); const v2KeyExtractor = useCallback((item: ThreadListV2ListItem) => item.key, []); + // FlatList treats a changed extraData identity as "re-render every visible + // row", so an inline object literal would invalidate all rows on every + // HomeScreen render. + const v2ExtraData = useMemo( + () => ({ + projectByKey, + projectCwdByKey, + projectTitleByProjectKey: v2ProjectTitleByProjectKey, + serverConfigs, + savedConnectionsById: props.savedConnectionsById, + }), + [ + projectByKey, + projectCwdByKey, + props.savedConnectionsById, + serverConfigs, + v2ProjectTitleByProjectKey, + ], + ); + const extraData = useMemo( () => ({ savedConnectionsById: props.savedConnectionsById, projectCwdByKey }), [props.savedConnectionsById, projectCwdByKey], @@ -900,13 +920,7 @@ export function HomeScreen(props: HomeScreenProps) { data={threadListV2Items} renderItem={renderV2Item} keyExtractor={v2KeyExtractor} - extraData={{ - projectByKey, - projectCwdByKey, - projectTitleByProjectKey: v2ProjectTitleByProjectKey, - serverConfigs, - savedConnectionsById: props.savedConnectionsById, - }} + extraData={v2ExtraData} ListHeaderComponent={v2ListHeader} ListFooterComponent={ threadListV2Layout.hiddenSettledCount > 0 ? ( diff --git a/apps/mobile/src/features/shortcuts/useAppShortcuts.ts b/apps/mobile/src/features/shortcuts/useAppShortcuts.ts index 3070a1d54e0..30b4e169889 100644 --- a/apps/mobile/src/features/shortcuts/useAppShortcuts.ts +++ b/apps/mobile/src/features/shortcuts/useAppShortcuts.ts @@ -54,7 +54,14 @@ function useShortcutNavigation(): void { } function useRecentThreadShortcutSync(state: NavigationState): void { - const threadRef = useMemo(() => activeThreadRef(state), [state]); + // Launcher shortcuts are Android-only. A null ref on iOS keeps this hook + // (mounted in the root stack layout) from subscribing the root to the + // active thread's shell, which would re-render every screen on each + // title/status/session change. + const threadRef = useMemo( + () => (Platform.OS === "android" ? activeThreadRef(state) : null), + [state], + ); const threadShell = useThreadShell(threadRef); // null until the persisted list loads; recording waits on it so the first // thread opened after a cold start cannot clobber older entries. diff --git a/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx b/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx index e328bcef00e..f37b5559a4a 100644 --- a/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx +++ b/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx @@ -946,7 +946,11 @@ export function NewTaskDraftScreen(props: { const promptEditor = ( ): ThreadFeedEntry[] { const grouped: ThreadFeedEntry[] = []; + // Mutable backing array for the trailing group so appending an activity is + // O(1) instead of re-copying the group (which made this loop quadratic on + // long tool runs). The array is only mutated while it is the trailing group. + let openGroupActivities: ThreadFeedActivity[] | null = null; + let openGroupTurnId: TurnId | null = null; for (const entry of entries) { // Skip empty messages so they don't break activity grouping. @@ -966,24 +971,23 @@ function groupAdjacentActivities(entries: ReadonlyArray): Th if (entry.type !== "activity") { grouped.push(entry); + openGroupActivities = null; continue; } - const previous = grouped.at(-1); - if (previous?.type === "activity-group" && previous.turnId === entry.turnId) { - grouped[grouped.length - 1] = { - ...previous, - activities: [...previous.activities, entry.activity], - }; + if (openGroupActivities !== null && openGroupTurnId === entry.turnId) { + openGroupActivities.push(entry.activity); continue; } + openGroupActivities = [entry.activity]; + openGroupTurnId = entry.turnId; grouped.push({ type: "activity-group", id: entry.id, createdAt: entry.createdAt, turnId: entry.turnId, - activities: [entry.activity], + activities: openGroupActivities, }); } @@ -1225,13 +1229,24 @@ function appendPresentedFeedEntry( }); } -export function derivePendingApprovals( +/** + * Sorts activities into lifecycle order. `derivePendingApprovals` and + * `derivePendingUserInputs` both expect this ordering; sorting once and + * passing the result to both avoids re-sorting the full activity history + * per derivation. + */ +export function sortThreadActivities( activities: ReadonlyArray, +): ReadonlyArray { + return Arr.sort(activities, activityOrder); +} + +export function derivePendingApprovals( + sortedActivities: ReadonlyArray, ): PendingApproval[] { const openByRequestId = new Map(); - const ordered = Arr.sort(activities, activityOrder); - for (const activity of ordered) { + for (const activity of sortedActivities) { const payload = activity.payload && typeof activity.payload === "object" ? (activity.payload as Record) @@ -1273,12 +1288,11 @@ export function derivePendingApprovals( } export function derivePendingUserInputs( - activities: ReadonlyArray, + sortedActivities: ReadonlyArray, ): PendingUserInput[] { const openByRequestId = new Map(); - const ordered = Arr.sort(activities, activityOrder); - for (const activity of ordered) { + for (const activity of sortedActivities) { const payload = activity.payload && typeof activity.payload === "object" ? (activity.payload as Record) diff --git a/apps/mobile/src/state/thread-outbox-manager.ts b/apps/mobile/src/state/thread-outbox-manager.ts index 19f89d13c51..f6a20ccffc2 100644 --- a/apps/mobile/src/state/thread-outbox-manager.ts +++ b/apps/mobile/src/state/thread-outbox-manager.ts @@ -88,11 +88,23 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { return loadPromise; }; - const enqueue = (message: QueuedThreadMessage): Promise => - serialize(async () => { + // The queued atom drives the composer's immediate "queued" feedback, so it + // is published synchronously; the durable write happens behind it and rolls + // the message back out if it fails (durability only matters for crash + // recovery, not for the in-session queue). + const enqueue = (message: QueuedThreadMessage): Promise => { + setMessages([ + ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), + message, + ]); + return serialize(async () => { try { await options.storage.write(message); } catch (cause) { + // Roll back by reference, not messageId: a retry enqueue with the same + // id may have optimistically replaced this attempt while the write was + // in flight, and its entry must survive this attempt's failure. + setMessages(currentMessages().filter((candidate) => candidate !== message)); throw new ThreadOutboxManagerError({ operation: "enqueue", environmentId: message.environmentId, @@ -101,11 +113,15 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { cause, }); } - setMessages([ - ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), - message, - ]); }); + }; + + // Resolves once all pending mutations (including any in-flight enqueue + // write) have settled, reporting whether the message is still queued. The + // drain awaits this before dispatching so a message whose durable write + // later fails can never have been delivered first. + const confirmQueued = (message: QueuedThreadMessage): Promise => + serialize(async () => currentMessages().some((candidate) => candidate === message)); // Rewrites an already-queued message. A no-op when the message has been // removed in the meantime (e.g. deleted or delivered), so a trailing editor @@ -204,6 +220,7 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { serialize, load, enqueue, + confirmQueued, update, remove, clearEnvironment, diff --git a/apps/mobile/src/state/thread-outbox.test.ts b/apps/mobile/src/state/thread-outbox.test.ts index 6c665c432f4..89f8b26798b 100644 --- a/apps/mobile/src/state/thread-outbox.test.ts +++ b/apps/mobile/src/state/thread-outbox.test.ts @@ -294,6 +294,111 @@ describe("thread outbox", () => { registry.dispose(); }); + it("publishes an enqueued message before the durable write resolves", async () => { + const registry = AtomRegistry.make(); + let releaseWrite!: () => void; + const writeBlocked = new Promise((resolve) => { + releaseWrite = resolve; + }); + const manager = createThreadOutboxManager({ + registry, + storage: { + load: async () => [], + write: async () => writeBlocked, + remove: async () => undefined, + }, + }); + const message = queuedMessage({ + messageId: "message-1", + createdAt: "2026-06-08T10:00:01.000Z", + }); + + const enqueueing = manager.enqueue(message); + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ + "environment-1:thread-1": [message], + }); + + releaseWrite(); + await enqueueing; + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ + "environment-1:thread-1": [message], + }); + registry.dispose(); + }); + + it("rolls an enqueued message back out when the durable write fails", async () => { + const registry = AtomRegistry.make(); + const writeCause = new Error("disk full"); + const manager = createThreadOutboxManager({ + registry, + storage: { + load: async () => [], + write: async () => { + throw writeCause; + }, + remove: async () => undefined, + }, + }); + const message = queuedMessage({ + messageId: "message-1", + createdAt: "2026-06-08T10:00:01.000Z", + }); + + await expect(manager.enqueue(message)).rejects.toEqual( + new ThreadOutboxManagerError({ + operation: "enqueue", + environmentId: message.environmentId, + threadId: message.threadId, + messageId: message.messageId, + cause: writeCause, + }), + ); + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({}); + registry.dispose(); + }); + + it("keeps a same-id retry queued when the first attempt's write fails", async () => { + const registry = AtomRegistry.make(); + let failNextWrite = true; + let releaseFirstWrite!: () => void; + const firstWriteBlocked = new Promise((resolve) => { + releaseFirstWrite = resolve; + }); + const manager = createThreadOutboxManager({ + registry, + storage: { + load: async () => [], + write: async () => { + if (failNextWrite) { + failNextWrite = false; + await firstWriteBlocked; + throw new Error("disk full"); + } + }, + remove: async () => undefined, + }, + }); + const message = queuedMessage({ + messageId: "message-1", + createdAt: "2026-06-08T10:00:01.000Z", + }); + const retried = { ...message, text: "retried" }; + + const first = manager.enqueue(message); + const second = manager.enqueue(retried); + releaseFirstWrite(); + await expect(first).rejects.toBeInstanceOf(ThreadOutboxManagerError); + await second; + + // The failed first attempt must not roll back the retry that replaced it. + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ + "environment-1:thread-1": [retried], + }); + await expect(manager.confirmQueued(retried)).resolves.toBe(true); + await expect(manager.confirmQueued(message)).resolves.toBe(false); + registry.dispose(); + }); + it("replaces an existing message when an enqueue retry uses the same id", async () => { const registry = AtomRegistry.make(); const manager = createThreadOutboxManager({ diff --git a/apps/mobile/src/state/thread-outbox.ts b/apps/mobile/src/state/thread-outbox.ts index a089c49732c..59287b12e61 100644 --- a/apps/mobile/src/state/thread-outbox.ts +++ b/apps/mobile/src/state/thread-outbox.ts @@ -20,6 +20,11 @@ export function enqueueThreadOutboxMessage(message: QueuedThreadMessage): Promis return threadOutboxManager.enqueue(message); } +/** Waits for pending writes to settle; false if the message was rolled back. */ +export function confirmThreadOutboxMessageQueued(message: QueuedThreadMessage): Promise { + return threadOutboxManager.confirmQueued(message); +} + /** Rewrite a queued message; no-op (false) if it was removed in the meantime. */ export function updateThreadOutboxMessage(message: QueuedThreadMessage): Promise { return threadOutboxManager.update(message); diff --git a/apps/mobile/src/state/use-selected-thread-requests.ts b/apps/mobile/src/state/use-selected-thread-requests.ts index c9e9db12530..82ff42f247a 100644 --- a/apps/mobile/src/state/use-selected-thread-requests.ts +++ b/apps/mobile/src/state/use-selected-thread-requests.ts @@ -11,6 +11,7 @@ import { derivePendingApprovals, derivePendingUserInputs, setPendingUserInputCustomAnswer, + sortThreadActivities, type PendingUserInputDraftAnswer, } from "../lib/threadActivity"; import { appAtomRegistry } from "./atom-registry"; @@ -70,14 +71,19 @@ export function useSelectedThreadRequests() { null, ); - const activePendingApprovals = useMemo( - () => (selectedThread ? derivePendingApprovals(selectedThread.activities) : []), + // Sort once; both derivations expect the same lifecycle ordering. + const sortedActivities = useMemo( + () => (selectedThread ? sortThreadActivities(selectedThread.activities) : []), [selectedThread], ); + const activePendingApprovals = useMemo( + () => derivePendingApprovals(sortedActivities), + [sortedActivities], + ); const activePendingApproval = activePendingApprovals[0] ?? null; const activePendingUserInputs = useMemo( - () => (selectedThread ? derivePendingUserInputs(selectedThread.activities) : []), - [selectedThread], + () => derivePendingUserInputs(sortedActivities), + [sortedActivities], ); const activePendingUserInput = activePendingUserInputs[0] ?? null; const activePendingUserInputDrafts = diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index 90831f8437a..b09aadf7e6b 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -30,6 +30,7 @@ import { composerDraftsAtom, ensureComposerDraftsLoaded, getComposerDraftSnapshot, + mergeComposerDraftContent, removeComposerDraftAttachment, setComposerDraftText, updateComposerDraftSettings, @@ -148,27 +149,36 @@ export function useThreadComposerState() { const metadata = makeQueuedMessageMetadata(); const messageId = MessageId.make(metadata.messageId); - try { - await enqueueThreadOutboxMessage({ - environmentId: selectedThreadShell.environmentId, - threadId: selectedThreadShell.id, - messageId, - commandId: CommandId.make(metadata.commandId), - text, - attachments, - modelSelection: draft.modelSelection ?? thread.modelSelection, - runtimeMode: draft.runtimeMode ?? thread.runtimeMode, - interactionMode: draft.interactionMode ?? thread.interactionMode, - createdAt: metadata.createdAt, - }); - clearComposerDraftContent(threadKey); - return messageId; - } catch (error) { + // Enqueue publishes the queued atom synchronously (the durable write + // happens behind it), so clearing the draft here gives send feedback on + // the tap frame instead of after file I/O. If the write fails the message + // is rolled out of the queue and the content is merged back into the + // draft, preserving anything typed since. + const enqueuePromise = enqueueThreadOutboxMessage({ + environmentId: selectedThreadShell.environmentId, + threadId: selectedThreadShell.id, + messageId, + commandId: CommandId.make(metadata.commandId), + text, + attachments, + modelSelection: draft.modelSelection ?? thread.modelSelection, + runtimeMode: draft.runtimeMode ?? thread.runtimeMode, + interactionMode: draft.interactionMode ?? thread.interactionMode, + createdAt: metadata.createdAt, + }); + clearComposerDraftContent(threadKey); + enqueuePromise.catch((error: unknown) => { + // Restore text via merge (idempotent) but attachments via the uncapped + // append: the merge path slots existing attachments first and truncates + // at the send limit, which would silently drop this message's images if + // the user attached new ones while the write was in flight. + void mergeComposerDraftContent(threadKey, { text, attachments: [] }); + appendComposerDraftAttachments(threadKey, attachments); setPendingConnectionError( error instanceof Error ? error.message : "Failed to save the queued message.", ); - return null; - } + }); + return messageId; }, [selectedThreadDetail, selectedThreadShell]); const onChangeDraftMessage = useCallback( diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 3559fa140fe..d06a4098aab 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -21,7 +21,11 @@ import { toUploadChatImageAttachments } from "../lib/composerImages"; import { randomHex } from "../lib/uuid"; import { appAtomRegistry } from "./atom-registry"; import { useProjects, useThreadShells } from "./entities"; -import { ensureThreadOutboxLoaded, removeThreadOutboxMessage } from "./thread-outbox"; +import { + confirmThreadOutboxMessageQueued, + ensureThreadOutboxLoaded, + removeThreadOutboxMessage, +} from "./thread-outbox"; import { isQueuedThreadCreationSendable, modelSelectionsEqual, @@ -33,7 +37,7 @@ import { type QueuedThreadMessage, type ThreadOutboxCommandStage, } from "./thread-outbox-model"; -import { threadEnvironment } from "./threads"; +import { environmentThreadShells, threadEnvironment } from "./threads"; import { useAtomCommand } from "./use-atom-command"; import { editingQueuedMessageIdsAtom, @@ -349,8 +353,32 @@ export function useThreadOutboxDrain(): void { return false; }, ); - const delivery = - deliveryAction === "remove" + // Enqueues publish optimistically before their durable write settles. + // Confirm the write landed (and the message wasn't rolled back) before + // sending, so a failed write can never chase an already-delivered turn. + const delivery = confirmThreadOutboxMessageQueued(nextQueuedMessage).then((queued) => { + if (!queued) { + // Rolled back by a failed write; nothing to deliver or retry. + return true; + } + // The guards evaluated before the confirmation await are stale by now: + // the thread may have gone busy, or the user may have opened this + // message in the editor. Re-read both and defer to the next drain pass + // (returning true skips the failure/backoff path) rather than sending + // a payload the user is editing or racing an active turn. + if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[nextQueuedMessage.messageId]) { + return true; + } + const freshThread = findThread( + appAtomRegistry.get(environmentThreadShells.threadShellsAtom), + nextQueuedMessage, + ); + const freshThreadBusy = + freshThread?.session?.status === "running" || freshThread?.session?.status === "starting"; + if (deliveryAction === "send" && creation === undefined && freshThreadBusy) { + return true; + } + return deliveryAction === "remove" ? removeQueuedMessage("[thread-outbox] failed to remove message for a missing thread") : creation !== undefined ? creationProjectCwd !== null @@ -359,6 +387,7 @@ export function useThreadOutboxDrain(): void { : thread !== undefined ? sendQueuedMessage(nextQueuedMessage, thread) : Promise.resolve(false); + }); void delivery .then((sent) => { if (sent) {