diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index 6bc60598c11..8dd4af33e04 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -103,7 +103,6 @@ export interface ThreadComposerProps { readonly threadSyncPhase?: "loading" | "syncing" | null; readonly selectedThread: OrchestrationThreadShell; readonly serverConfig: T3ServerConfig | null; - readonly queueCount: number; readonly activeThreadBusy: boolean; readonly environmentId: EnvironmentId; readonly projectCwd: string | null; @@ -314,10 +313,7 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer props.selectedThread.session?.status === "running" || props.selectedThread.session?.status === "starting"; - const sendLabel = - props.connectionState !== "connected" || props.activeThreadBusy || props.queueCount > 0 - ? "Queue" - : "Send"; + const sendLabel = "Send"; const currentModelSelection = props.selectedThread.modelSelection; const currentRuntimeMode = props.selectedThread.runtimeMode; const currentInteractionMode = props.selectedThread.interactionMode ?? "default"; @@ -991,16 +987,6 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer ) : null} - - {/* Queue count */} - {props.queueCount > 0 ? ( - - - {props.queueCount} queued message{props.queueCount === 1 ? "" : "s"} will send - automatically. - - - ) : null} void; readonly onStopThread: () => void; readonly onSendMessage: () => Promise; + readonly onSteerQueuedMessage: (messageId: MessageId) => Promise; + readonly onRemoveQueuedMessage: ( + messageId: MessageId, + source: "local" | "server", + ) => Promise; readonly onStartNewThread: () => void; readonly onReconnectEnvironment: () => void; readonly onUpdateThreadModelSelection: (modelSelection: ModelSelection) => void; @@ -247,10 +251,11 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread }, [freeze, selectedThreadKey]); useEffect(() => { + // Anchor as soon as the target row exists in the feed — including local + // outbox "Sending" bubbles painted before thread detail has finished loading. if ( anchorMessageId === null || lastScrolledAnchorMessageIdRef.current === anchorMessageId || - contentPresentationKind !== "ready" || !selectedThreadFeed.some((entry) => entry.type === "message" && entry.id === anchorMessageId) ) { return; @@ -289,14 +294,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread }); }); return () => cancelAnimationFrame(frame); - }, [ - anchorMessageId, - freeze, - contentPresentationKind, - selectedThreadFeed, - scrollMessageToEnd, - selectedThreadKey, - ]); + }, [anchorMessageId, freeze, selectedThreadFeed, scrollMessageToEnd, selectedThreadKey]); const handleSendMessage = useCallback(async () => { const targetThreadKey = selectedThreadKey; @@ -382,6 +380,8 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread hasMoreOlder={props.hasMoreOlderActivities} loadingOlder={props.loadingOlderActivities} onLoadOlder={props.onLoadOlderActivities} + onSteerQueuedMessage={props.onSteerQueuedMessage} + onRemoveQueuedMessage={props.onRemoveQueuedMessage} /> ) : ( @@ -439,7 +439,6 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread threadSyncPhase={threadSyncPhase} selectedThread={props.selectedThread} serverConfig={props.serverConfig} - queueCount={props.selectedThreadQueueCount} activeThreadBusy={props.activeThreadBusy} environmentId={props.environmentId} projectCwd={props.projectWorkspaceRoot} diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 7e991e0aaae..065d4b91208 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -1,7 +1,7 @@ import * as Haptics from "expo-haptics"; import { KeyboardAwareLegendList } from "@legendapp/list/keyboard"; import { type LegendListRef } from "@legendapp/list/react-native"; -import type { EnvironmentId, MessageId, ThreadId, TurnId } from "@t3tools/contracts"; +import { MessageId, type EnvironmentId, type ThreadId, type TurnId } from "@t3tools/contracts"; import { CHAT_LIST_ANCHOR_OFFSET, resolveChatListAnchoredEndSpace } from "@t3tools/shared/chatList"; import { formatElapsed } from "@t3tools/shared/orchestrationTiming"; import { SymbolView } from "../../components/AppSymbol"; @@ -86,6 +86,7 @@ import { useAppearanceCodeSurface } from "../settings/appearance/useAppearanceCo import { markdownFileIconSource } from "@t3tools/mobile-markdown-text/file-icons"; import { resolveMarkdownLinkPresentation } from "@t3tools/mobile-markdown-text/links"; import { + deriveQueuedMessageControls, deriveThreadFeedPresentation, type ThreadFeedEntry, type ThreadFeedLatestTurn, @@ -146,6 +147,11 @@ export interface ThreadFeedProps { readonly hasMoreOlder?: boolean; readonly loadingOlder?: boolean; readonly onLoadOlder?: () => void; + readonly onSteerQueuedMessage: (messageId: MessageId) => Promise; + readonly onRemoveQueuedMessage: ( + messageId: MessageId, + source: "local" | "server", + ) => Promise; } function MessageAttachmentImage(props: { @@ -801,7 +807,10 @@ function useMarkdownStyles(onLinkPress: (href: string) => void): MarkdownStyleSe function renderFeedEntry( info: { item: ThreadFeedEntry; index: number }, - props: Pick & { + props: Pick< + ThreadFeedProps, + "environmentId" | "skills" | "onSteerQueuedMessage" | "onRemoveQueuedMessage" + > & { readonly copiedRowId: string | null; readonly expandedWorkRows: Record; readonly terminalAssistantMessageIds: ReadonlySet; @@ -867,6 +876,10 @@ function renderFeedEntry( const styles = isUser ? markdownStyles.user : markdownStyles.assistant; const timestampLabel = formatMessageTime(isUser ? message.createdAt : message.updatedAt); const attachments = message.attachments ?? []; + const previewAttachments = entry.previewAttachments ?? []; + const deliveryState = entry.deliveryState; + const queueSource = entry.queueSource; + const queueControls = deriveQueuedMessageControls(deliveryState, queueSource); const hasReviewCommentContext = message.text.includes(" ); })} + {previewAttachments.map((attachment) => ( + + ))} + {deliveryState === "sending" ? ( + <> + + + Sending + + + ) : deliveryState === "waiting" ? ( + + Waiting for connection + + ) : deliveryState === "queued" ? ( + + Queued on server + + ) : null} + {queueControls.canSteer ? ( + void props.onSteerQueuedMessage(MessageId.make(message.id))} + className="min-h-8 justify-center rounded-full px-2" + > + Send now + + ) : null} + {queueControls.canRemove && queueSource ? ( + + void props.onRemoveQueuedMessage(MessageId.make(message.id), queueSource) + } + className="min-h-8 justify-center rounded-full px-2" + > + + {queueSource === "local" ? "Discard" : "Remove"} + + + ) : null} {timestampLabel} @@ -1722,6 +1790,8 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { onToggleTurnFold, onPressImage, onMarkdownLinkPress, + onSteerQueuedMessage: props.onSteerQueuedMessage, + onRemoveQueuedMessage: props.onRemoveQueuedMessage, iconSubtleColor, userBubbleColor, markdownStyles, @@ -1743,6 +1813,8 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { userBubbleMaxWidth, onCopyWorkRow, onMarkdownLinkPress, + props.onSteerQueuedMessage, + props.onRemoveQueuedMessage, onPressImage, onToggleTurnFold, onToggleWorkGroup, diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index ca42e97dab8..1c1b6d903b9 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -793,7 +793,6 @@ function ThreadRouteContent( environmentId={selectedThread.environmentId} projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null} threadCwd={selectedThreadCwd} - selectedThreadQueueCount={composer.selectedThreadQueueCount} layoutVariant={layout.variant} usesAutomaticContentInsets={usesNativeHeaderGlass} onOpenConnectionEditor={handleOpenConnectionEditor} @@ -804,6 +803,8 @@ function ThreadRouteContent( serverConfig={serverConfig} onStopThread={handleStopThread} onSendMessage={composer.onSendMessage} + onSteerQueuedMessage={composer.onSteerQueuedMessage} + onRemoveQueuedMessage={composer.onRemoveQueuedMessage} onStartNewThread={handleStartNewThread} onReconnectEnvironment={handleReconnectEnvironment} onUpdateThreadModelSelection={composer.onUpdateModelSelection} diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index c191cd108f7..36c7f8d14df 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -13,11 +13,35 @@ import { import { buildThreadFeed, + deriveQueuedMessageControls, deriveThreadFeedPresentation, type ThreadFeedActivity, type ThreadFeedEntry, } from "./threadActivity"; +describe("deriveQueuedMessageControls", () => { + it("allows steering or removing server-queued messages", () => { + expect(deriveQueuedMessageControls("queued", "server")).toEqual({ + canSteer: true, + canRemove: true, + }); + }); + + it("allows discarding an offline local-outbox message", () => { + expect(deriveQueuedMessageControls("waiting", "local")).toEqual({ + canSteer: false, + canRemove: true, + }); + }); + + it("does not claim an in-flight local send can still be cancelled", () => { + expect(deriveQueuedMessageControls("sending", "local")).toEqual({ + canSteer: false, + canRemove: false, + }); + }); +}); + function makeActivity( input: Partial & Pick, diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 9be3b6c1a60..9487e94391b 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -18,6 +18,8 @@ import { deriveResolvedUserInputTranscripts } from "@t3tools/shared/userInputTra import * as Arr from "effect/Array"; import * as Order from "effect/Order"; +import type { DraftComposerImageAttachment } from "./composerImages"; + export interface PendingApproval { readonly requestId: ApprovalRequestId; readonly requestKind: "command" | "file-read" | "file-change"; @@ -94,6 +96,9 @@ type RawThreadFeedEntry = readonly id: string; readonly createdAt: string; readonly message: OrchestrationThread["messages"][number]; + readonly deliveryState?: "waiting" | "sending" | "queued"; + readonly queueSource?: "local" | "server"; + readonly previewAttachments?: ReadonlyArray; } | { readonly type: "activity"; @@ -136,6 +141,18 @@ export type ThreadFeedEntry = readonly expanded: boolean; }; +export function deriveQueuedMessageControls( + deliveryState: "waiting" | "sending" | "queued" | undefined, + queueSource: "local" | "server" | undefined, +): { readonly canSteer: boolean; readonly canRemove: boolean } { + return { + canSteer: deliveryState === "queued" && queueSource === "server", + canRemove: + (deliveryState === "queued" && queueSource === "server") || + (deliveryState === "waiting" && queueSource === "local"), + }; +} + export type ThreadFeedLatestTurn = Pick< OrchestrationLatestTurn, "turnId" | "state" | "startedAt" | "completedAt" diff --git a/apps/mobile/src/state/thread-outbox-manager.ts b/apps/mobile/src/state/thread-outbox-manager.ts index 19f89d13c51..bfa25549bdd 100644 --- a/apps/mobile/src/state/thread-outbox-manager.ts +++ b/apps/mobile/src/state/thread-outbox-manager.ts @@ -88,11 +88,24 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { return loadPromise; }; - const enqueue = (message: QueuedThreadMessage): Promise => - serialize(async () => { + const enqueue = (message: QueuedThreadMessage): Promise => { + // Paint the optimistic bubble immediately. Disk durability trails the + // in-memory queue so send UX never waits on filesystem latency. + setMessages([ + ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), + message, + ]); + return serialize(async () => { + // Dropped while the write was queued (e.g. user discarded / delivered). + if (!currentMessages().some((candidate) => candidate.messageId === message.messageId)) { + return; + } try { await options.storage.write(message); } catch (cause) { + setMessages( + currentMessages().filter((candidate) => candidate.messageId !== message.messageId), + ); throw new ThreadOutboxManagerError({ operation: "enqueue", environmentId: message.environmentId, @@ -101,11 +114,8 @@ export function createThreadOutboxManager(options: ThreadOutboxManagerOptions) { cause, }); } - setMessages([ - ...currentMessages().filter((candidate) => candidate.messageId !== message.messageId), - 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 diff --git a/apps/mobile/src/state/thread-outbox-model.ts b/apps/mobile/src/state/thread-outbox-model.ts index 3ba61be3872..57720970bb2 100644 --- a/apps/mobile/src/state/thread-outbox-model.ts +++ b/apps/mobile/src/state/thread-outbox-model.ts @@ -153,7 +153,6 @@ export function resolveThreadOutboxDeliveryAction(input: { readonly threadExists: boolean; readonly shellStatus: EnvironmentShellStatus; readonly environmentConnected: boolean; - readonly threadBusy: boolean; }): ThreadOutboxDeliveryAction { if (input.isCreation) { // A pending task creates its thread on delivery. If the thread already @@ -169,7 +168,10 @@ export function resolveThreadOutboxDeliveryAction(input: { if (!input.threadExists) { return input.shellStatus === "live" ? "remove" : "wait"; } - return input.environmentConnected && !input.threadBusy ? "send" : "wait"; + // Once connected, hand ownership to the server immediately. The server + // persists follow-ups that arrive during an active turn; this local outbox + // is only the offline/transport safety boundary. + return input.environmentConnected ? "send" : "wait"; } /** diff --git a/apps/mobile/src/state/thread-outbox.test.ts b/apps/mobile/src/state/thread-outbox.test.ts index 6c665c432f4..1782a71b369 100644 --- a/apps/mobile/src/state/thread-outbox.test.ts +++ b/apps/mobile/src/state/thread-outbox.test.ts @@ -247,6 +247,69 @@ describe("thread outbox", () => { registry.dispose(); }); + it("surfaces enqueue in memory before the durable write resolves", async () => { + const registry = AtomRegistry.make(); + let releaseWrite!: () => void; + const writeGate = new Promise((resolve) => { + releaseWrite = resolve; + }); + const storage: ThreadOutboxStorage = { + load: async () => [], + write: async () => { + await writeGate; + }, + remove: async () => undefined, + }; + const manager = createThreadOutboxManager({ registry, storage }); + const message = queuedMessage({ + messageId: "message-1", + createdAt: "2026-06-08T10:00:01.000Z", + }); + + const enqueuePromise = manager.enqueue(message); + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ + "environment-1:thread-1": [message], + }); + + releaseWrite(); + await enqueuePromise; + registry.dispose(); + }); + + it("rolls back the in-memory queue when the durable write fails", async () => { + const registry = AtomRegistry.make(); + const writeCause = new Error("write failed"); + const storage: ThreadOutboxStorage = { + load: async () => [], + write: async () => { + throw writeCause; + }, + remove: async () => undefined, + }; + const manager = createThreadOutboxManager({ registry, storage }); + const message = queuedMessage({ + messageId: "message-1", + createdAt: "2026-06-08T10:00:01.000Z", + }); + + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({}); + const enqueuePromise = manager.enqueue(message); + expect(registry.get(manager.queuedMessagesByThreadKeyAtom)).toEqual({ + "environment-1:thread-1": [message], + }); + await expect(enqueuePromise).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 atom state aligned with durable writes and removals", async () => { const registry = AtomRegistry.make(); const stored = new Map(); @@ -352,14 +415,13 @@ describe("thread outbox", () => { registry.dispose(); }); - it("only removes a missing-thread message after shell synchronization is live", () => { + it("keeps offline messages local and hands connected messages to the server", () => { expect( resolveThreadOutboxDeliveryAction({ isCreation: false, threadExists: false, shellStatus: "synchronizing", environmentConnected: true, - threadBusy: false, }), ).toBe("wait"); expect( @@ -368,16 +430,22 @@ describe("thread outbox", () => { threadExists: false, shellStatus: "live", environmentConnected: true, - threadBusy: false, }), ).toBe("remove"); + expect( + resolveThreadOutboxDeliveryAction({ + isCreation: false, + threadExists: true, + shellStatus: "live", + environmentConnected: false, + }), + ).toBe("wait"); expect( resolveThreadOutboxDeliveryAction({ isCreation: false, threadExists: true, shellStatus: "live", environmentConnected: true, - threadBusy: false, }), ).toBe("send"); }); @@ -389,7 +457,6 @@ describe("thread outbox", () => { threadExists: false, shellStatus: "cached", environmentConnected: false, - threadBusy: false, }), ).toBe("wait"); // Connected but not yet synchronized: a previously delivered creation may @@ -400,7 +467,6 @@ describe("thread outbox", () => { threadExists: false, shellStatus: "synchronizing", environmentConnected: true, - threadBusy: false, }), ).toBe("wait"); expect( @@ -409,7 +475,6 @@ describe("thread outbox", () => { threadExists: false, shellStatus: "live", environmentConnected: true, - threadBusy: false, }), ).toBe("send"); expect( @@ -418,7 +483,6 @@ describe("thread outbox", () => { threadExists: true, shellStatus: "live", environmentConnected: true, - threadBusy: true, }), ).toBe("remove"); }); diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index be83fe3ea31..e8eb8d85adc 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -41,12 +41,16 @@ import { updateComposerDraftSettings, useComposerDraft, } from "./use-composer-drafts"; -import { setPendingConnectionError } from "../state/use-remote-environment-registry"; +import { + setPendingConnectionError, + useRemoteConnectionStatus, +} from "../state/use-remote-environment-registry"; import { orchestrationEnvironment } from "../state/orchestration"; import { useSelectedThreadDetail } from "../state/use-thread-detail"; import { useThreadSelection } from "../state/use-thread-selection"; import { useAtomCommand } from "./use-atom-command"; -import { enqueueThreadOutboxMessage } from "./thread-outbox"; +import { threadEnvironment } from "./threads"; +import { enqueueThreadOutboxMessage, removeThreadOutboxMessage } from "./thread-outbox"; import { useThreadOutboxMessages } from "./use-thread-outbox"; const EMPTY_ACTIVITIES: ReadonlyArray = []; @@ -87,6 +91,7 @@ export function useThreadComposerState() { const selectedThreadDetail = useSelectedThreadDetail(); const composerDrafts = useAtomValue(composerDraftsAtom); const queuedMessagesByThreadKey = useThreadOutboxMessages(); + const { connectedEnvironments } = useRemoteConnectionStatus(); useEffect(() => { ensureComposerDraftsLoaded(); @@ -106,6 +111,12 @@ export function useThreadComposerState() { const loadThreadActivities = useAtomCommand(orchestrationEnvironment.loadThreadActivities, { reportFailure: false, }); + const steerQueuedMessage = useAtomCommand(threadEnvironment.steerQueuedMessage, { + label: "steer queued message", + }); + const removeServerQueuedMessage = useAtomCommand(threadEnvironment.removeQueuedMessage, { + label: "remove queued message", + }); const selectedEnvironmentIdForActivities = selectedThreadShell?.environmentId ?? null; const selectedThreadIdForActivities = selectedThreadShell?.id ?? null; const loadOlderActivitiesPage = useCallback( @@ -143,18 +154,91 @@ export function useThreadComposerState() { loadPage: loadOlderActivitiesPage, }); - const selectedThreadFeed = useMemo( - () => - selectedThreadDetail - ? buildThreadFeed({ ...selectedThreadDetail, activities: mergedActivities }) - : [], - [selectedThreadDetail, mergedActivities], - ); + const selectedThreadFeed = useMemo(() => { + // Local outbox rows must still paint while detail is hydrating — otherwise + // send during "Loading messages…" produces no bubble until the snapshot lands. + const feed = selectedThreadDetail + ? buildThreadFeed({ ...selectedThreadDetail, activities: mergedActivities }) + : []; + const timelineMessageIds = new Set( + selectedThreadDetail?.messages.map((message) => message.id) ?? [], + ); + const optimisticByMessageId = new Map< + MessageId, + (typeof feed)[number] & { readonly type: "message" } + >(); + + for (const message of selectedThreadDetail?.queuedMessages ?? []) { + if (timelineMessageIds.has(message.messageId)) { + continue; + } + optimisticByMessageId.set(message.messageId, { + type: "message", + id: message.messageId, + createdAt: message.queuedAt, + deliveryState: "queued", + queueSource: "server", + message: { + id: message.messageId, + role: "user", + text: message.text, + attachments: message.attachments, + turnId: null, + streaming: false, + createdAt: message.queuedAt, + updatedAt: message.queuedAt, + }, + }); + } + + // A local outbox entry wins over the matching server projection until the + // command acknowledgement removes it. That makes the bubble transition + // from "Sending" to "Queued" without rendering twice. + for (const message of selectedThreadQueuedMessages) { + if (timelineMessageIds.has(message.messageId)) { + continue; + } + optimisticByMessageId.set(message.messageId, { + type: "message", + id: message.messageId, + createdAt: message.createdAt, + queueSource: "local", + deliveryState: connectedEnvironments.some( + (environment) => + environment.environmentId === message.environmentId && + environment.connectionState === "connected", + ) + ? "sending" + : "waiting", + previewAttachments: message.attachments, + message: { + id: message.messageId, + role: "user", + text: message.text, + attachments: [], + turnId: null, + streaming: false, + createdAt: message.createdAt, + updatedAt: message.createdAt, + }, + }); + } + + if (optimisticByMessageId.size === 0) { + return feed; + } + + return [ + ...feed, + ...Array.from(optimisticByMessageId.values()).sort((left, right) => + left.createdAt.localeCompare(right.createdAt), + ), + ]; + }, [connectedEnvironments, selectedThreadDetail, selectedThreadQueuedMessages, mergedActivities]); const selectedDraft = selectedThreadKey ? composerDrafts[selectedThreadKey] : null; const draftMessage = selectedDraft?.text ?? ""; const draftAttachments = selectedDraft?.attachments ?? []; - const selectedThreadQueueCount = selectedThreadQueuedMessages.length; const selectedThread = selectedThreadDetail ?? selectedThreadShell; const modelSelection = selectedDraft?.modelSelection ?? selectedThread?.modelSelection ?? null; const runtimeMode = selectedDraft?.runtimeMode ?? selectedThread?.runtimeMode ?? null; @@ -205,29 +289,69 @@ 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 updates the in-memory outbox synchronously so the feed can paint + // "Sending" before disk I/O finishes. Clear the draft in the same turn and + // return immediately — durability trails off the critical path. + void 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, + }).catch((error) => { + // Memory outbox already rolled back on write failure; restore the draft. + setComposerDraftText(threadKey, draft.text); + if (draft.attachments.length > 0) { + appendComposerDraftAttachments(threadKey, draft.attachments); + } setPendingConnectionError( error instanceof Error ? error.message : "Failed to save the queued message.", ); - return null; - } + }); + clearComposerDraftContent(threadKey); + return messageId; }, [selectedThreadDetail, selectedThreadShell]); + const onSteerQueuedMessage = useCallback( + async (messageId: MessageId) => { + if (!selectedThreadShell) { + return; + } + await steerQueuedMessage({ + environmentId: selectedThreadShell.environmentId, + input: { threadId: selectedThreadShell.id, messageId }, + }); + }, + [selectedThreadShell, steerQueuedMessage], + ); + + const onRemoveQueuedMessage = useCallback( + async (messageId: MessageId, source: "local" | "server") => { + if (!selectedThreadShell) { + return; + } + if (source === "local") { + const message = selectedThreadQueuedMessages.find( + (candidate) => candidate.messageId === messageId, + ); + if (message) { + await removeThreadOutboxMessage(message); + } + return; + } + await removeServerQueuedMessage({ + environmentId: selectedThreadShell.environmentId, + input: { threadId: selectedThreadShell.id, messageId }, + }); + }, + [removeServerQueuedMessage, selectedThreadQueuedMessages, selectedThreadShell], + ); + const onChangeDraftMessage = useCallback( (value: string) => { if (!selectedThreadShell) { @@ -348,7 +472,6 @@ export function useThreadComposerState() { return { selectedThreadFeed, - selectedThreadQueueCount, activeWorkStartedAt, draftMessage, draftAttachments, @@ -369,6 +492,8 @@ export function useThreadComposerState() { onNativePasteImages, onRemoveDraftImage, onSendMessage, + onSteerQueuedMessage, + onRemoveQueuedMessage, onUpdateModelSelection, onUpdateRuntimeMode, onUpdateInteractionMode, diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 3559fa140fe..9ebf3808f94 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -310,7 +310,6 @@ export function useThreadOutboxDrain(): void { threadExists: thread !== undefined, shellStatus, environmentConnected: environment?.connectionState === "connected", - threadBusy: thread?.session?.status === "running" || thread?.session?.status === "starting", }); if (deliveryAction === "wait") { continue; diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 49ae73007d2..e1156b9cd32 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -717,6 +717,7 @@ describe("hasServerAcknowledgedLocalDispatch", () => { phase: "running", latestTurn: runningTurn, latestUserMessageId: localDispatch.latestUserMessageId, + projectedMessageIds: new Set(), session: runningSession, hasPendingApproval: false, hasPendingUserInput: false, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 3f1566345e2..dfa8dd059be 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -4844,34 +4844,44 @@ function ChatViewContent(props: ChatViewProps) { preparingWorktree: Boolean(baseBranchForWorktree), messageId: messageIdForSend, }); + let turnStartSucceeded = false; - const composerImagesSnapshot = [...composerImages]; - const composerTerminalContextsSnapshot = [...sendableComposerTerminalContexts]; - const composerElementContextsSnapshot = [...composerElementContexts]; - const composerPreviewAnnotationsSnapshot = [...composerPreviewAnnotations]; - const composerReviewCommentsSnapshot: ReviewCommentContext[] = [...composerReviewComments]; - const messageTextWithContexts = appendElementContextsToPrompt( - appendTerminalContextsToPrompt(promptForSend, composerTerminalContextsSnapshot), - composerElementContextsSnapshot, - ); - const messageTextWithPreviewAnnotations = composerPreviewAnnotationsSnapshot.reduce( - (text, annotation) => appendPreviewAnnotationPrompt(text, annotation), - messageTextWithContexts, - ); - const messageTextForSend = appendReviewCommentsToPrompt( - messageTextWithPreviewAnnotations, - composerReviewCommentsSnapshot, - ); - const messageCreatedAt = new Date().toISOString(); - const outgoingMessageText = formatOutgoingPrompt({ - provider: ctxSelectedProvider, - model: ctxSelectedModel, - models: ctxSelectedProviderModels, - effort: ctxSelectedPromptEffort, - text: messageTextForSend || IMAGE_ONLY_BOOTSTRAP_PROMPT, - }); - const turnAttachmentsPromise = Promise.all( - composerImagesSnapshot.map(async (image) => ({ + try { + const composerImagesSnapshot = [...composerImages]; + const composerTerminalContextsSnapshot = [...sendableComposerTerminalContexts]; + const composerElementContextsSnapshot = [...composerElementContexts]; + const composerPreviewAnnotationsSnapshot = [...composerPreviewAnnotations]; + const composerReviewCommentsSnapshot: ReviewCommentContext[] = [...composerReviewComments]; + const messageTextWithContexts = appendElementContextsToPrompt( + appendTerminalContextsToPrompt(promptForSend, composerTerminalContextsSnapshot), + composerElementContextsSnapshot, + ); + const messageTextWithPreviewAnnotations = composerPreviewAnnotationsSnapshot.reduce( + (text, annotation) => appendPreviewAnnotationPrompt(text, annotation), + messageTextWithContexts, + ); + const messageTextForSend = appendReviewCommentsToPrompt( + messageTextWithPreviewAnnotations, + composerReviewCommentsSnapshot, + ); + const messageCreatedAt = new Date().toISOString(); + const outgoingMessageText = formatOutgoingPrompt({ + provider: ctxSelectedProvider, + model: ctxSelectedModel, + models: ctxSelectedProviderModels, + effort: ctxSelectedPromptEffort, + text: messageTextForSend || IMAGE_ONLY_BOOTSTRAP_PROMPT, + }); + const turnAttachmentsPromise = Promise.all( + composerImagesSnapshot.map(async (image) => ({ + type: "image" as const, + name: image.name, + mimeType: image.mimeType, + sizeBytes: image.sizeBytes, + dataUrl: await readFileAsDataUrl(image.file), + })), + ); + const optimisticAttachments = composerImagesSnapshot.map((image) => ({ type: "image" as const, id: image.id, name: image.name, @@ -4992,51 +5002,43 @@ function ChatViewContent(props: ChatViewProps) { failure = turnAttachmentsResult; } - const turnAttachmentsResult = await settlePromise(() => turnAttachmentsPromise); - if (failure === null && turnAttachmentsResult._tag === "Failure") { - failure = turnAttachmentsResult; - } - - let turnStartSucceeded = false; - if (failure === null && turnAttachmentsResult._tag === "Success") { - const bootstrap = - isLocalDraftThread || baseBranchForWorktree - ? { - ...(isLocalDraftThread - ? { - createThread: { - projectId: activeProject.id, - title, - modelSelection: threadCreateModelSelection, - runtimeMode, - interactionMode, - branch: activeThreadBranch, - worktreePath: activeThread.worktreePath, - createdAt: activeThread.createdAt, - }, - } - : {}), - ...(baseBranchForWorktree - ? { - prepareWorktree: { - projectCwd: activeProject.workspaceRoot, - baseBranch: baseBranchForWorktree, - ...(reuseBaseBranch - ? { reuseBaseBranch: true } - : { - branch: buildTemporaryWorktreeBranchName(randomHex), - ...(startFromOrigin ? { startFromOrigin: true } : {}), - }), - }, - runSetupScript: true, - } - : {}), - } - : undefined; - beginLocalDispatch({ preparingWorktree: false, messageId: messageIdForSend }); - const startResult = await startThreadTurn({ - environmentId, - input: { + if (failure === null && turnAttachmentsResult._tag === "Success") { + const bootstrap = + isLocalDraftThread || baseBranchForWorktree + ? { + ...(isLocalDraftThread + ? { + createThread: { + projectId: activeProject.id, + title, + modelSelection: threadCreateModelSelection, + runtimeMode, + interactionMode, + branch: activeThreadBranch, + worktreePath: activeThread.worktreePath, + createdAt: activeThread.createdAt, + }, + } + : {}), + ...(baseBranchForWorktree + ? { + prepareWorktree: { + projectCwd: activeProject.workspaceRoot, + baseBranch: baseBranchForWorktree, + ...(reuseBaseBranch + ? { reuseBaseBranch: true } + : { + branch: buildTemporaryWorktreeBranchName(randomHex), + ...(startFromOrigin ? { startFromOrigin: true } : {}), + }), + }, + runSetupScript: true, + } + : {}), + } + : undefined; + const queuedTurnInput = { + commandId: newCommandId(), threadId: threadIdForSend, message: { messageId: messageIdForSend,