| author | |
| committer | |
| log | f0f16e4f6e8e923dcb17fc1c25d4fb1e442f80a5 |
| tree | 183b284a2439a7e88397427bbd171aef056e62f4 |
| parent | 00da6b57a43aef39162ff9e8205b5441c24298d0 |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>12 files changed, 283 insertions(+), 61 deletions(-)
apps/mobile/src/Stack.tsx+10-1| ... | ... | @@ -284,6 +284,15 @@ function workspacePathFromState(state: NavigationState): string { |
| 284 | 284 | return path.startsWith("/") ? path : `/${path}`; |
| 285 | 285 | } |
| 286 | 286 | |
| 287 | // The drain hook subscribes to the outbox, all thread shells, projects, and | |
| 288 | // connection statuses. Hosting it in a null-rendering leaf keeps those | |
| 289 | // updates from re-rendering RootStackLayout (and with it every screen) on | |
| 290 | // each enqueue, shell change, or reconnect. | |
| 291 | function ThreadOutboxDrainWorker() { | |
| 292 | useThreadOutboxDrain(); | |
| 293 | return null; | |
| 294 | } | |
| 295 | ||
| 287 | 296 | function RootStackLayout(props: { |
| 288 | 297 | readonly children: React.ReactNode; |
| 289 | 298 | readonly state: NavigationState; |
| ... | ... | @@ -292,7 +301,6 @@ function RootStackLayout(props: { |
| 292 | 301 | const { pendingShare } = useIncomingShare(); |
| 293 | 302 | const sharePresentationRef = useRef(EMPTY_INCOMING_SHARE_PRESENTATION_STATE); |
| 294 | 303 | useAgentNotificationNavigation(); |
| 295 | useThreadOutboxDrain(); | |
| 296 | 304 | // Presents the T3 Connect onboarding sheet after an in-session sign-in. |
| 297 | 305 | useConnectOnboardingNavigation(); |
| 298 | 306 | // Launcher app shortcuts: routes shortcut taps and tracks opened threads. |
| ... | ... | @@ -320,6 +328,7 @@ function RootStackLayout(props: { |
| 320 | 328 | |
| 321 | 329 | return ( |
| 322 | 330 | <HardwareKeyboardCommandProvider pathname={pathname}> |
| 331 | <ThreadOutboxDrainWorker /> | |
| 323 | 332 | <ShowcaseCaptureCoordinator pathname={pathname} /> |
| 324 | 333 | <ClerkSettingsSheetDetentProvider initiallyExpanded={false}> |
| 325 | 334 | <AdaptiveWorkspaceLayout pathname={workspacePathname}> |
apps/mobile/src/features/home/HomeScreen.tsx+21-7| ... | ... | @@ -665,6 +665,26 @@ export function HomeScreen(props: HomeScreenProps) { |
| 665 | 665 | ); |
| 666 | 666 | const v2KeyExtractor = useCallback((item: ThreadListV2ListItem) => item.key, []); |
| 667 | 667 | |
| 668 | // FlatList treats a changed extraData identity as "re-render every visible | |
| 669 | // row", so an inline object literal would invalidate all rows on every | |
| 670 | // HomeScreen render. | |
| 671 | const v2ExtraData = useMemo( | |
| 672 | () => ({ | |
| 673 | projectByKey, | |
| 674 | projectCwdByKey, | |
| 675 | projectTitleByProjectKey: v2ProjectTitleByProjectKey, | |
| 676 | serverConfigs, | |
| 677 | savedConnectionsById: props.savedConnectionsById, | |
| 678 | }), | |
| 679 | [ | |
| 680 | projectByKey, | |
| 681 | projectCwdByKey, | |
| 682 | props.savedConnectionsById, | |
| 683 | serverConfigs, | |
| 684 | v2ProjectTitleByProjectKey, | |
| 685 | ], | |
| 686 | ); | |
| 687 | ||
| 668 | 688 | const extraData = useMemo( |
| 669 | 689 | () => ({ savedConnectionsById: props.savedConnectionsById, projectCwdByKey }), |
| 670 | 690 | [props.savedConnectionsById, projectCwdByKey], |
| ... | ... | @@ -900,13 +920,7 @@ export function HomeScreen(props: HomeScreenProps) { |
| 900 | 920 | data={threadListV2Items} |
| 901 | 921 | renderItem={renderV2Item} |
| 902 | 922 | keyExtractor={v2KeyExtractor} |
| 903 | extraData={{ | |
| 904 | projectByKey, | |
| 905 | projectCwdByKey, | |
| 906 | projectTitleByProjectKey: v2ProjectTitleByProjectKey, | |
| 907 | serverConfigs, | |
| 908 | savedConnectionsById: props.savedConnectionsById, | |
| 909 | }} | |
| 923 | extraData={v2ExtraData} | |
| 910 | 924 | ListHeaderComponent={v2ListHeader} |
| 911 | 925 | ListFooterComponent={ |
| 912 | 926 | threadListV2Layout.hiddenSettledCount > 0 ? ( |
apps/mobile/src/features/shortcuts/useAppShortcuts.ts+8-1| ... | ... | @@ -54,7 +54,14 @@ function useShortcutNavigation(): void { |
| 54 | 54 | } |
| 55 | 55 | |
| 56 | 56 | function useRecentThreadShortcutSync(state: NavigationState): void { |
| 57 | const threadRef = useMemo(() => activeThreadRef(state), [state]); | |
| 57 | // Launcher shortcuts are Android-only. A null ref on iOS keeps this hook | |
| 58 | // (mounted in the root stack layout) from subscribing the root to the | |
| 59 | // active thread's shell, which would re-render every screen on each | |
| 60 | // title/status/session change. | |
| 61 | const threadRef = useMemo( | |
| 62 | () => (Platform.OS === "android" ? activeThreadRef(state) : null), | |
| 63 | [state], | |
| 64 | ); | |
| 58 | 65 | const threadShell = useThreadShell(threadRef); |
| 59 | 66 | // null until the persisted list loads; recording waits on it so the first |
| 60 | 67 | // thread opened after a cold start cannot clobber older entries. |
apps/mobile/src/features/threads/NewTaskDraftScreen.tsx+5-1| ... | ... | @@ -946,7 +946,11 @@ export function NewTaskDraftScreen(props: { |
| 946 | 946 | const promptEditor = ( |
| 947 | 947 | <ComposerEditor |
| 948 | 948 | ref={promptInputRef} |
| 949 | autoFocus={!isAndroid} | |
| 949 | // Native autoFocus fires becomeFirstResponder in didMoveToWindow, which | |
| 950 | // forces the iOS keyboard bring-up during the formSheet present | |
| 951 | // animation and stalls it. The runAfterInteractions effect above focuses | |
| 952 | // the editor once the transition settles instead. | |
| 953 | autoFocus={false} | |
| 950 | 954 | editable={!isIncomingShareTransferPending} |
| 951 | 955 | multiline |
| 952 | 956 | scrollEnabled={isExpanded} |
apps/mobile/src/features/threads/ThreadComposer.tsx+8-6| ... | ... | @@ -518,14 +518,16 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer |
| 518 | 518 | const threadKey = scopedThreadKey(props.environmentId, props.selectedThread.id); |
| 519 | 519 | if (inFlightThreadIdsRef.current.has(threadKey)) return; |
| 520 | 520 | inFlightThreadIdsRef.current.add(threadKey); |
| 521 | // Sending a prompt starts agent work: arm the lock-screen card now, while | |
| 522 | // the app is foregrounded and the activity token can be registered. | |
| 523 | armAgentAwarenessLiveActivityForLocalWork({ | |
| 524 | threadTitle: props.selectedThread.title, | |
| 525 | projectTitle: props.environmentLabel ?? "T3 Code", | |
| 526 | }); | |
| 527 | 521 | try { |
| 528 | 522 | await onSendMessage(); |
| 523 | // Sending a prompt starts agent work: arm the lock-screen card while the | |
| 524 | // app is foregrounded and the activity token can be registered. Armed | |
| 525 | // after the send so its preference read and native Activity start don't | |
| 526 | // contend with the queued-message feedback on the tap frame. | |
| 527 | armAgentAwarenessLiveActivityForLocalWork({ | |
| 528 | threadTitle: props.selectedThread.title, | |
| 529 | projectTitle: props.environmentLabel ?? "T3 Code", | |
| 530 | }); | |
| 529 | 531 | } finally { |
| 530 | 532 | inFlightThreadIdsRef.current.delete(threadKey); |
| 531 | 533 | } |
apps/mobile/src/lib/threadActivity.ts+27-13| ... | ... | @@ -957,6 +957,11 @@ function isEmptyMessage(entry: RawThreadFeedEntry): boolean { |
| 957 | 957 | |
| 958 | 958 | function groupAdjacentActivities(entries: ReadonlyArray<RawThreadFeedEntry>): ThreadFeedEntry[] { |
| 959 | 959 | const grouped: ThreadFeedEntry[] = []; |
| 960 | // Mutable backing array for the trailing group so appending an activity is | |
| 961 | // O(1) instead of re-copying the group (which made this loop quadratic on | |
| 962 | // long tool runs). The array is only mutated while it is the trailing group. | |
| 963 | let openGroupActivities: ThreadFeedActivity[] | null = null; | |
| 964 | let openGroupTurnId: TurnId | null = null; | |
| 960 | 965 | |
| 961 | 966 | for (const entry of entries) { |
| 962 | 967 | // Skip empty messages so they don't break activity grouping. |
| ... | ... | @@ -966,24 +971,23 @@ function groupAdjacentActivities(entries: ReadonlyArray<RawThreadFeedEntry>): Th |
| 966 | 971 | |
| 967 | 972 | if (entry.type !== "activity") { |
| 968 | 973 | grouped.push(entry); |
| 974 | openGroupActivities = null; | |
| 969 | 975 | continue; |
| 970 | 976 | } |
| 971 | 977 | |
| 972 | const previous = grouped.at(-1); | |
| 973 | if (previous?.type === "activity-group" && previous.turnId === entry.turnId) { | |
| 974 | grouped[grouped.length - 1] = { | |
| 975 | ...previous, | |
| 976 | activities: [...previous.activities, entry.activity], | |
| 977 | }; | |
| 978 | if (openGroupActivities !== null && openGroupTurnId === entry.turnId) { | |
| 979 | openGroupActivities.push(entry.activity); | |
| 978 | 980 | continue; |
| 979 | 981 | } |
| 980 | 982 | |
| 983 | openGroupActivities = [entry.activity]; | |
| 984 | openGroupTurnId = entry.turnId; | |
| 981 | 985 | grouped.push({ |
| 982 | 986 | type: "activity-group", |
| 983 | 987 | id: entry.id, |
| 984 | 988 | createdAt: entry.createdAt, |
| 985 | 989 | turnId: entry.turnId, |
| 986 | activities: [entry.activity], | |
| 990 | activities: openGroupActivities, | |
| 987 | 991 | }); |
| 988 | 992 | } |
| 989 | 993 | |
| ... | ... | @@ -1225,13 +1229,24 @@ function appendPresentedFeedEntry( |
| 1225 | 1229 | }); |
| 1226 | 1230 | } |
| 1227 | 1231 | |
| 1228 | export function derivePendingApprovals( | |
| 1232 | /** | |
| 1233 | * Sorts activities into lifecycle order. `derivePendingApprovals` and | |
| 1234 | * `derivePendingUserInputs` both expect this ordering; sorting once and | |
| 1235 | * passing the result to both avoids re-sorting the full activity history | |
| 1236 | * per derivation. | |
| 1237 | */ | |
| 1238 | export function sortThreadActivities( | |
| 1229 | 1239 | activities: ReadonlyArray<OrchestrationThreadActivity>, |
| 1240 | ): ReadonlyArray<OrchestrationThreadActivity> { | |
| 1241 | return Arr.sort(activities, activityOrder); | |
| 1242 | } | |
| 1243 | ||
| 1244 | export function derivePendingApprovals( | |
| 1245 | sortedActivities: ReadonlyArray<OrchestrationThreadActivity>, | |
| 1230 | 1246 | ): PendingApproval[] { |
| 1231 | 1247 | const openByRequestId = new Map<ApprovalRequestId, PendingApproval>(); |
| 1232 | const ordered = Arr.sort(activities, activityOrder); | |
| 1233 | 1248 | |
| 1234 | for (const activity of ordered) { | |
| 1249 | for (const activity of sortedActivities) { | |
| 1235 | 1250 | const payload = |
| 1236 | 1251 | activity.payload && typeof activity.payload === "object" |
| 1237 | 1252 | ? (activity.payload as Record<string, unknown>) |
| ... | ... | @@ -1273,12 +1288,11 @@ export function derivePendingApprovals( |
| 1273 | 1288 | } |
| 1274 | 1289 | |
| 1275 | 1290 | export function derivePendingUserInputs( |
| 1276 | activities: ReadonlyArray<OrchestrationThreadActivity>, | |
| 1291 | sortedActivities: ReadonlyArray<OrchestrationThreadActivity>, | |
| 1277 | 1292 | ): PendingUserInput[] { |
| 1278 | 1293 | const openByRequestId = new Map<ApprovalRequestId, PendingUserInput>(); |
| 1279 | const ordered = Arr.sort(activities, activityOrder); | |
| 1280 | 1294 | |
| 1281 | for (const activity of ordered) { | |
| 1295 | for (const activity of sortedActivities) { | |
| 1282 | 1296 | const payload = |
| 1283 | 1297 | activity.payload && typeof activity.payload === "object" |
| 1284 | 1298 | ? (activity.payload as Record<string, unknown>) |
apps/mobile/src/state/thread-outbox-manager.ts+23-6| ... | ... | @@ -88,11 +88,23 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { |
| 88 | 88 | return loadPromise; |
| 89 | 89 | }; |
| 90 | 90 | |
| 91 | const enqueue = (message: QueuedThreadMessage): Promise<void> => | |
| 92 | serialize(async () => { | |
| 91 | // The queued atom drives the composer's immediate "queued" feedback, so it | |
| 92 | // is published synchronously; the durable write happens behind it and rolls | |
| 93 | // the message back out if it fails (durability only matters for crash | |
| 94 | // recovery, not for the in-session queue). | |
| 95 | const enqueue = (message: QueuedThreadMessage): Promise<void> => { | |
| 96 | setMessages([ | |
| 97 | ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), | |
| 98 | message, | |
| 99 | ]); | |
| 100 | return serialize(async () => { | |
| 93 | 101 | try { |
| 94 | 102 | await options.storage.write(message); |
| 95 | 103 | } catch (cause) { |
| 104 | // Roll back by reference, not messageId: a retry enqueue with the same | |
| 105 | // id may have optimistically replaced this attempt while the write was | |
| 106 | // in flight, and its entry must survive this attempt's failure. | |
| 107 | setMessages(currentMessages().filter((candidate) => candidate !== message)); | |
| 96 | 108 | throw new ThreadOutboxManagerError({ |
| 97 | 109 | operation: "enqueue", |
| 98 | 110 | environmentId: message.environmentId, |
| ... | ... | @@ -101,11 +113,15 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { |
| 101 | 113 | cause, |
| 102 | 114 | }); |
| 103 | 115 | } |
| 104 | setMessages([ | |
| 105 | ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), | |
| 106 | message, | |
| 107 | ]); | |
| 108 | 116 | }); |
| 117 | }; | |
| 118 | ||
| 119 | // Resolves once all pending mutations (including any in-flight enqueue | |
| 120 | // write) have settled, reporting whether the message is still queued. The | |
| 121 | // drain awaits this before dispatching so a message whose durable write | |
| 122 | // later fails can never have been delivered first. | |
| 123 | const confirmQueued = (message: QueuedThreadMessage): Promise<boolean> => | |
| 124 | serialize(async () => currentMessages().some((candidate) => candidate === message)); | |
| 109 | 125 | |
| 110 | 126 | // Rewrites an already-queued message. A no-op when the message has been |
| 111 | 127 | // removed in the meantime (e.g. deleted or delivered), so a trailing editor |
| ... | ... | @@ -204,6 +220,7 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { |
| 204 | 220 | serialize, |
| 205 | 221 | load, |
| 206 | 222 | enqueue, |
| 223 | confirmQueued, | |
| 207 | 224 | update, |
| 208 | 225 | remove, |
| 209 | 226 | clearEnvironment, |
apps/mobile/src/state/thread-outbox.test.ts+105| ... | ... | @@ -294,6 +294,111 @@ describe("thread outbox", () => { |
| 294 | 294 | registry.dispose(); |
| 295 | 295 | }); |
| 296 | 296 | |
| 297 | it("publishes an enqueued message before the durable write resolves", async () => { | |
| 298 | const registry = AtomRegistry.make(); | |
| 299 | let releaseWrite!: () => void; | |
| 300 | const writeBlocked = new Promise<void>((resolve) => { | |
| 301 | releaseWrite = resolve; | |
| 302 | }); | |
| 303 | const manager = createThreadOutboxManager({ | |
| 304 | registry, | |
| 305 | storage: { | |
| 306 | load: async () => [], | |
| 307 | write: async () => writeBlocked, | |
| 308 | remove: async () => undefined, | |
| 309 | }, | |
| 310 | }); | |
| 311 | const message = queuedMessage({ | |
| 312 | messageId: "message-1", | |
| 313 | createdAt: "2026-06-08T10:00:01.000Z", | |
| 314 | }); | |
| 315 | ||
| 316 | const enqueueing = manager.enqueue(message); | |
| 317 | expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ | |
| 318 | "environment-1:thread-1": [message], | |
| 319 | }); | |
| 320 | ||
| 321 | releaseWrite(); | |
| 322 | await enqueueing; | |
| 323 | expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ | |
| 324 | "environment-1:thread-1": [message], | |
| 325 | }); | |
| 326 | registry.dispose(); | |
| 327 | }); | |
| 328 | ||
| 329 | it("rolls an enqueued message back out when the durable write fails", async () => { | |
| 330 | const registry = AtomRegistry.make(); | |
| 331 | const writeCause = new Error("disk full"); | |
| 332 | const manager = createThreadOutboxManager({ | |
| 333 | registry, | |
| 334 | storage: { | |
| 335 | load: async () => [], | |
| 336 | write: async () => { | |
| 337 | throw writeCause; | |
| 338 | }, | |
| 339 | remove: async () => undefined, | |
| 340 | }, | |
| 341 | }); | |
| 342 | const message = queuedMessage({ | |
| 343 | messageId: "message-1", | |
| 344 | createdAt: "2026-06-08T10:00:01.000Z", | |
| 345 | }); | |
| 346 | ||
| 347 | await expect(manager.enqueue(message)).rejects.toEqual( | |
| 348 | new ThreadOutboxManagerError({ | |
| 349 | operation: "enqueue", | |
| 350 | environmentId: message.environmentId, | |
| 351 | threadId: message.threadId, | |
| 352 | messageId: message.messageId, | |
| 353 | cause: writeCause, | |
| 354 | }), | |
| 355 | ); | |
| 356 | expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({}); | |
| 357 | registry.dispose(); | |
| 358 | }); | |
| 359 | ||
| 360 | it("keeps a same-id retry queued when the first attempt's write fails", async () => { | |
| 361 | const registry = AtomRegistry.make(); | |
| 362 | let failNextWrite = true; | |
| 363 | let releaseFirstWrite!: () => void; | |
| 364 | const firstWriteBlocked = new Promise<void>((resolve) => { | |
| 365 | releaseFirstWrite = resolve; | |
| 366 | }); | |
| 367 | const manager = createThreadOutboxManager({ | |
| 368 | registry, | |
| 369 | storage: { | |
| 370 | load: async () => [], | |
| 371 | write: async () => { | |
| 372 | if (failNextWrite) { | |
| 373 | failNextWrite = false; | |
| 374 | await firstWriteBlocked; | |
| 375 | throw new Error("disk full"); | |
| 376 | } | |
| 377 | }, | |
| 378 | remove: async () => undefined, | |
| 379 | }, | |
| 380 | }); | |
| 381 | const message = queuedMessage({ | |
| 382 | messageId: "message-1", | |
| 383 | createdAt: "2026-06-08T10:00:01.000Z", | |
| 384 | }); | |
| 385 | const retried = { ...message, text: "retried" }; | |
| 386 | ||
| 387 | const first = manager.enqueue(message); | |
| 388 | const second = manager.enqueue(retried); | |
| 389 | releaseFirstWrite(); | |
| 390 | await expect(first).rejects.toBeInstanceOf(ThreadOutboxManagerError); | |
| 391 | await second; | |
| 392 | ||
| 393 | // The failed first attempt must not roll back the retry that replaced it. | |
| 394 | expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ | |
| 395 | "environment-1:thread-1": [retried], | |
| 396 | }); | |
| 397 | await expect(manager.confirmQueued(retried)).resolves.toBe(true); | |
| 398 | await expect(manager.confirmQueued(message)).resolves.toBe(false); | |
| 399 | registry.dispose(); | |
| 400 | }); | |
| 401 | ||
| 297 | 402 | it("replaces an existing message when an enqueue retry uses the same id", async () => { |
| 298 | 403 | const registry = AtomRegistry.make(); |
| 299 | 404 | const manager = createThreadOutboxManager({ |
apps/mobile/src/state/thread-outbox.ts+5| ... | ... | @@ -20,6 +20,11 @@ export function enqueueThreadOutboxMessage(message: QueuedThreadMessage): Promis |
| 20 | 20 | return threadOutboxManager.enqueue(message); |
| 21 | 21 | } |
| 22 | 22 | |
| 23 | /** Waits for pending writes to settle; false if the message was rolled back. */ | |
| 24 | export function confirmThreadOutboxMessageQueued(message: QueuedThreadMessage): Promise<boolean> { | |
| 25 | return threadOutboxManager.confirmQueued(message); | |
| 26 | } | |
| 27 | ||
| 23 | 28 | /** Rewrite a queued message; no-op (false) if it was removed in the meantime. */ |
| 24 | 29 | export function updateThreadOutboxMessage(message: QueuedThreadMessage): Promise<boolean> { |
| 25 | 30 | return threadOutboxManager.update(message); |
apps/mobile/src/state/use-selected-thread-requests.ts+10-4| ... | ... | @@ -11,6 +11,7 @@ import { |
| 11 | 11 | derivePendingApprovals, |
| 12 | 12 | derivePendingUserInputs, |
| 13 | 13 | setPendingUserInputCustomAnswer, |
| 14 | sortThreadActivities, | |
| 14 | 15 | type PendingUserInputDraftAnswer, |
| 15 | 16 | } from "../lib/threadActivity"; |
| 16 | 17 | import { appAtomRegistry } from "./atom-registry"; |
| ... | ... | @@ -70,14 +71,19 @@ export function useSelectedThreadRequests() { |
| 70 | 71 | null, |
| 71 | 72 | ); |
| 72 | 73 | |
| 73 | const activePendingApprovals = useMemo( | |
| 74 | () => (selectedThread ? derivePendingApprovals(selectedThread.activities) : []), | |
| 74 | // Sort once; both derivations expect the same lifecycle ordering. | |
| 75 | const sortedActivities = useMemo( | |
| 76 | () => (selectedThread ? sortThreadActivities(selectedThread.activities) : []), | |
| 75 | 77 | [selectedThread], |
| 76 | 78 | ); |
| 79 | const activePendingApprovals = useMemo( | |
| 80 | () => derivePendingApprovals(sortedActivities), | |
| 81 | [sortedActivities], | |
| 82 | ); | |
| 77 | 83 | const activePendingApproval = activePendingApprovals[0] ?? null; |
| 78 | 84 | const activePendingUserInputs = useMemo( |
| 79 | () => (selectedThread ? derivePendingUserInputs(selectedThread.activities) : []), | |
| 80 | [selectedThread], | |
| 85 | () => derivePendingUserInputs(sortedActivities), | |
| 86 | [sortedActivities], | |
| 81 | 87 | ); |
| 82 | 88 | const activePendingUserInput = activePendingUserInputs[0] ?? null; |
| 83 | 89 | const activePendingUserInputDrafts = |
apps/mobile/src/state/use-thread-composer-state.ts+28-18| ... | ... | @@ -30,6 +30,7 @@ import { |
| 30 | 30 | composerDraftsAtom, |
| 31 | 31 | ensureComposerDraftsLoaded, |
| 32 | 32 | getComposerDraftSnapshot, |
| 33 | mergeComposerDraftContent, | |
| 33 | 34 | removeComposerDraftAttachment, |
| 34 | 35 | setComposerDraftText, |
| 35 | 36 | updateComposerDraftSettings, |
| ... | ... | @@ -148,27 +149,36 @@ export function useThreadComposerState() { |
| 148 | 149 | |
| 149 | 150 | const metadata = makeQueuedMessageMetadata(); |
| 150 | 151 | const messageId = MessageId.make(metadata.messageId); |
| 151 | try { | |
| 152 | await enqueueThreadOutboxMessage({ | |
| 153 | environmentId: selectedThreadShell.environmentId, | |
| 154 | threadId: selectedThreadShell.id, | |
| 155 | messageId, | |
| 156 | commandId: CommandId.make(metadata.commandId), | |
| 157 | text, | |
| 158 | attachments, | |
| 159 | modelSelection: draft.modelSelection ?? thread.modelSelection, | |
| 160 | runtimeMode: draft.runtimeMode ?? thread.runtimeMode, | |
| 161 | interactionMode: draft.interactionMode ?? thread.interactionMode, | |
| 162 | createdAt: metadata.createdAt, | |
| 163 | }); | |
| 164 | clearComposerDraftContent(threadKey); | |
| 165 | return messageId; | |
| 166 | } catch (error) { | |
| 152 | // Enqueue publishes the queued atom synchronously (the durable write | |
| 153 | // happens behind it), so clearing the draft here gives send feedback on | |
| 154 | // the tap frame instead of after file I/O. If the write fails the message | |
| 155 | // is rolled out of the queue and the content is merged back into the | |
| 156 | // draft, preserving anything typed since. | |
| 157 | const enqueuePromise = enqueueThreadOutboxMessage({ | |
| 158 | environmentId: selectedThreadShell.environmentId, | |
| 159 | threadId: selectedThreadShell.id, | |
| 160 | messageId, | |
| 161 | commandId: CommandId.make(metadata.commandId), | |
| 162 | text, | |
| 163 | attachments, | |
| 164 | modelSelection: draft.modelSelection ?? thread.modelSelection, | |
| 165 | runtimeMode: draft.runtimeMode ?? thread.runtimeMode, | |
| 166 | interactionMode: draft.interactionMode ?? thread.interactionMode, | |
| 167 | createdAt: metadata.createdAt, | |
| 168 | }); | |
| 169 | clearComposerDraftContent(threadKey); | |
| 170 | enqueuePromise.catch((error: unknown) => { | |
| 171 | // Restore text via merge (idempotent) but attachments via the uncapped | |
| 172 | // append: the merge path slots existing attachments first and truncates | |
| 173 | // at the send limit, which would silently drop this message's images if | |
| 174 | // the user attached new ones while the write was in flight. | |
| 175 | void mergeComposerDraftContent(threadKey, { text, attachments: [] }); | |
| 176 | appendComposerDraftAttachments(threadKey, attachments); | |
| 167 | 177 | setPendingConnectionError( |
| 168 | 178 | error instanceof Error ? error.message : "Failed to save the queued message.", |
| 169 | 179 | ); |
| 170 | return null; | |
| 171 | } | |
| 180 | }); | |
| 181 | return messageId; | |
| 172 | 182 | }, [selectedThreadDetail, selectedThreadShell]); |
| 173 | 183 | |
| 174 | 184 | const onChangeDraftMessage = useCallback( |
apps/mobile/src/state/use-thread-outbox-drain.ts+33-4| ... | ... | @@ -21,7 +21,11 @@ import { toUploadChatImageAttachments } from "../lib/composerImages"; |
| 21 | 21 | import { randomHex } from "../lib/uuid"; |
| 22 | 22 | import { appAtomRegistry } from "./atom-registry"; |
| 23 | 23 | import { useProjects, useThreadShells } from "./entities"; |
| 24 | import { ensureThreadOutboxLoaded, removeThreadOutboxMessage } from "./thread-outbox"; | |
| 24 | import { | |
| 25 | confirmThreadOutboxMessageQueued, | |
| 26 | ensureThreadOutboxLoaded, | |
| 27 | removeThreadOutboxMessage, | |
| 28 | } from "./thread-outbox"; | |
| 25 | 29 | import { |
| 26 | 30 | isQueuedThreadCreationSendable, |
| 27 | 31 | modelSelectionsEqual, |
| ... | ... | @@ -33,7 +37,7 @@ import { |
| 33 | 37 | type QueuedThreadMessage, |
| 34 | 38 | type ThreadOutboxCommandStage, |
| 35 | 39 | } from "./thread-outbox-model"; |
| 36 | import { threadEnvironment } from "./threads"; | |
| 40 | import { environmentThreadShells, threadEnvironment } from "./threads"; | |
| 37 | 41 | import { useAtomCommand } from "./use-atom-command"; |
| 38 | 42 | import { |
| 39 | 43 | editingQueuedMessageIdsAtom, |
| ... | ... | @@ -349,8 +353,32 @@ export function useThreadOutboxDrain(): void { |
| 349 | 353 | return false; |
| 350 | 354 | }, |
| 351 | 355 | ); |
| 352 | const delivery = | |
| 353 | deliveryAction === "remove" | |
| 356 | // Enqueues publish optimistically before their durable write settles. | |
| 357 | // Confirm the write landed (and the message wasn't rolled back) before | |
| 358 | // sending, so a failed write can never chase an already-delivered turn. | |
| 359 | const delivery = confirmThreadOutboxMessageQueued(nextQueuedMessage).then((queued) => { | |
| 360 | if (!queued) { | |
| 361 | // Rolled back by a failed write; nothing to deliver or retry. | |
| 362 | return true; | |
| 363 | } | |
| 364 | // The guards evaluated before the confirmation await are stale by now: | |
| 365 | // the thread may have gone busy, or the user may have opened this | |
| 366 | // message in the editor. Re-read both and defer to the next drain pass | |
| 367 | // (returning true skips the failure/backoff path) rather than sending | |
| 368 | // a payload the user is editing or racing an active turn. | |
| 369 | if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[nextQueuedMessage.messageId]) { | |
| 370 | return true; | |
| 371 | } | |
| 372 | const freshThread = findThread( | |
| 373 | appAtomRegistry.get(environmentThreadShells.threadShellsAtom), | |
| 374 | nextQueuedMessage, | |
| 375 | ); | |
| 376 | const freshThreadBusy = | |
| 377 | freshThread?.session?.status === "running" || freshThread?.session?.status === "starting"; | |
| 378 | if (deliveryAction === "send" && creation === undefined && freshThreadBusy) { | |
| 379 | return true; | |
| 380 | } | |
| 381 | return deliveryAction === "remove" | |
| 354 | 382 | ? removeQueuedMessage("[thread-outbox] failed to remove message for a missing thread") |
| 355 | 383 | : creation !== undefined |
| 356 | 384 | ? creationProjectCwd !== null |
| ... | ... | @@ -359,6 +387,7 @@ export function useThreadOutboxDrain(): void { |
| 359 | 387 | : thread !== undefined |
| 360 | 388 | ? sendQueuedMessage(nextQueuedMessage, thread) |
| 361 | 389 | : Promise.resolve(false); |
| 390 | }); | |
| 362 | 391 | void delivery |
| 363 | 392 | .then((sent) => { |
| 364 | 393 | if (sent) { |