diff --git a/apps/mobile/src/features/threads/ThreadComposer.tsx b/apps/mobile/src/features/threads/ThreadComposer.tsx index d015b2aa95b3..a241866ae601 100644 --- a/apps/mobile/src/features/threads/ThreadComposer.tsx +++ b/apps/mobile/src/features/threads/ThreadComposer.tsx @@ -35,9 +35,7 @@ import { } from "../../state/composer-attachment-uploads"; import Animated, { FadeIn, - FadeInDown, FadeOut, - FadeOutDown, LinearTransition, ReduceMotion, useAnimatedStyle, @@ -272,12 +270,7 @@ const ComposerConnectionStatusPill = memo(function ComposerConnectionStatusPill( }) { const isReconnecting = props.status.kind === "reconnecting"; return ( - + - + ); }); @@ -891,16 +884,6 @@ export const ThreadComposer = memo(function ThreadComposer(props: ThreadComposer - - {/* Queue count */} - {props.queueCount > 0 ? ( - - - {props.queueCount} queued message{props.queueCount === 1 ? "" : "s"} will send - automatically. - - - ) : null} diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 3392b534b634..8846e22057f7 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -37,6 +37,7 @@ import { useState, } from "react"; import { + Alert, AppState, Keyboard, Platform, @@ -67,6 +68,8 @@ import type { StatusTone } from "../../components/StatusPill"; import type { DraftComposerAttachment } from "../../lib/composerImages"; import { CHAT_CONTENT_MAX_WIDTH, type LayoutVariant } from "../../lib/layout"; import { IOS_NAV_BAR_HEIGHT } from "../../lib/layoutMetrics"; +import { editPendingThreadMessage } from "../../state/edit-pending-thread-message"; +import type { QueuedThreadMessage } from "../../state/thread-outbox-model"; import { scopedThreadKey } from "../../lib/scopedEntities"; import type { PendingApproval, @@ -127,6 +130,8 @@ export interface ThreadDetailScreenProps { readonly projectWorkspaceRoot: string | null; readonly threadCwd: string | null; readonly selectedThreadQueueCount: number; + readonly queuedMessages: ReadonlyArray; + readonly dispatchingMessageId: MessageId | null; readonly serverConfig: T3ServerConfig | null; readonly layoutVariant?: LayoutVariant; readonly usesAutomaticContentInsets?: boolean; @@ -347,6 +352,15 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread return null; })(); const showWorkingControl = floatingStatus !== null; + // Connection and working status occupy the same space. Keep the feed inset + // stable when reconnecting hands off to syncing and then to a running turn. + const showFloatingStatus = + showWorkingControl || + props.connectionStateLabel !== "connected" || + props.queuedMessages.length > 0 || + props.selectedThreadFeed.some( + (entry) => "acknowledged" in entry && entry.acknowledged === true, + ); const selectedThreadFeed = props.selectedThreadFeed; const hasCompactableConversation = selectedThreadFeed.some( @@ -486,14 +500,14 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread const userInputInsetProgress = useSharedValue(1); const userInputCardCoverage = useSharedValue(0); const floatingControlCoverage = useSharedValue( - showWorkingControl ? FLOATING_WORKING_CONTROL_COVERAGE : 0, + showFloatingStatus ? FLOATING_WORKING_CONTROL_COVERAGE : 0, ); useEffect(() => { floatingControlCoverage.value = withTiming( - showWorkingControl ? FLOATING_WORKING_CONTROL_COVERAGE : 0, + showFloatingStatus ? FLOATING_WORKING_CONTROL_COVERAGE : 0, { duration: 180, reduceMotion: ReduceMotion.System }, ); - }, [floatingControlCoverage, showWorkingControl]); + }, [floatingControlCoverage, showFloatingStatus]); // Android renders the expanded card in-flow (it cannot hit-test the iOS // overlay outside the bar's bounds), so its measured overlay height already // includes the card — the coverage extra is iOS-only. @@ -553,12 +567,12 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread useEffect(() => { const previous = previousWorkingControlStateRef.current; const threadChanged = previous.threadKey !== selectedThreadKey; - const visibilityChanged = previous.visible !== showWorkingControl; + const visibilityChanged = previous.visible !== showFloatingStatus; previousWorkingControlStateRef.current = { threadKey: selectedThreadKey, - visible: showWorkingControl, + visible: showFloatingStatus, }; - if ((!threadChanged && !visibilityChanged) || (threadChanged && !showWorkingControl)) { + if ((!threadChanged && !visibilityChanged) || (threadChanged && !showFloatingStatus)) { return; } // LegendList applies the larger inset but does not re-anchor short @@ -566,7 +580,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread // initial load. Re-pin after the finite inset transition; the callback // checks follow state again so a user who scrolled up stays put. scheduleOverlayRepin(230); - }, [scheduleOverlayRepin, selectedThreadKey, showWorkingControl]); + }, [scheduleOverlayRepin, selectedThreadKey, showFloatingStatus]); const handleToggleUserInputCollapsed = useCallback(() => { if (activeUserInputRequestId === null) { return; @@ -633,11 +647,13 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread useEffect(() => { if ( submittedMessageId === null || + anchorMessageId !== submittedMessageId || lastScrolledSubmittedMessageIdRef.current === submittedMessageId || contentPresentationKind !== "ready" || - !selectedThreadFeed.some( + (!selectedThreadFeed.some( (entry) => entry.type === "message" && entry.id === submittedMessageId, - ) + ) && + !props.queuedMessages.some((message) => message.messageId === submittedMessageId)) ) { return; } @@ -676,9 +692,11 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread }); return () => cancelAnimationFrame(frame); }, [ + anchorMessageId, submittedMessageId, freeze, contentPresentationKind, + props.queuedMessages, selectedThreadFeed, scrollMessageToEnd, selectedThreadKey, @@ -719,6 +737,22 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread selectedThreadKey, ]); + const handleEditPendingMessage = useCallback(async (message: QueuedThreadMessage) => { + try { + if ( + (await editPendingThreadMessage(message)) && + selectedThreadKeyRef.current === scopedThreadKey(message.environmentId, message.threadId) + ) { + composerEditorRef.current?.focus(); + } + } catch (error) { + Alert.alert( + "Could not edit message", + error instanceof Error ? error.message : "Please try again.", + ); + } + }, []); + const collapseComposer = useCallback(() => { composerEditorRef.current?.blur(); }, []); @@ -796,6 +830,9 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread threadId={props.selectedThread.id} workspaceRoot={props.threadCwd} feed={props.selectedThreadFeed} + queuedMessages={props.queuedMessages} + dispatchingMessageId={props.dispatchingMessageId} + onEditPendingMessage={handleEditPendingMessage} contentPresentation={props.contentPresentation} agentLabel={agentLabel} latestTurn={props.selectedThread.latestTurn} @@ -807,7 +844,7 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread contentInsetEndAdjustment={combinedContentInsetEndAdjustment} contentTopInset={0} contentBottomInset={ - estimatedOverlayHeight + (showWorkingControl ? FLOATING_WORKING_CONTROL_COVERAGE : 0) + estimatedOverlayHeight + (showFloatingStatus ? FLOATING_WORKING_CONTROL_COVERAGE : 0) } contentMaxWidth={contentMaxWidth} layoutVariant={layoutVariant} diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index f1456aabd258..408224034c9b 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -72,12 +72,7 @@ import { FilePreviewModal, type FilePreviewSource } from "../../components/FileP import { isPdfFile } from "../../lib/filePreview"; import { PresentationSource } from "../../components/NativePresentation"; import { useSafeAreaInsets } from "react-native-safe-area-context"; -import Animated, { - FadeIn, - FadeInUp, - LinearTransition, - type SharedValue, -} from "react-native-reanimated"; +import Animated, { FadeIn, LinearTransition, type SharedValue } from "react-native-reanimated"; import { useUniwindTheme } from "../../lib/useUniwindTheme"; import { IOS_NAV_BAR_HEIGHT } from "../../lib/layoutMetrics"; import { useFontFamily } from "../../lib/useFontFamily"; @@ -163,6 +158,8 @@ import { THREAD_DISCLOSURE_TRANSITION_MS, WORK_GROUP_TOGGLE_HEIGHT, } from "./thread-work-log"; +import { appendPendingThreadMessages, type PendingThreadFeedEntry } from "./pending-thread-feed"; +import type { QueuedThreadMessage } from "../../state/thread-outbox-model"; import { useMarkdownCodeHighlight } from "./markdownCodeHighlightState"; import { assetEnvironment, @@ -231,6 +228,9 @@ function isFreshTimestamp(input: string): boolean { } export interface ThreadFeedProps { + readonly queuedMessages: ReadonlyArray; + readonly dispatchingMessageId: MessageId | null; + readonly onEditPendingMessage: (message: QueuedThreadMessage) => void; readonly environmentId: EnvironmentId; readonly threadId: ThreadId; readonly workspaceRoot?: string | null; @@ -1324,8 +1324,15 @@ function useMarkdownStyles( } function renderFeedEntry( - info: { item: ThreadFeedEntry; index: number }, - props: Pick & { + info: { item: PendingThreadFeedEntry; index: number }, + props: Pick< + ThreadFeedProps, + | "environmentId" + | "onUseArtifactTemplate" + | "skills" + | "dispatchingMessageId" + | "onEditPendingMessage" + > & { readonly copiedRowId: string | null; readonly expandedWorkRows: Record; readonly workRowSizing: ReturnType; @@ -1470,12 +1477,8 @@ function renderFeedEntry( !message.streaming; if (isUser) { - const enterAnimated = isFreshTimestamp(message.createdAt); return ( - + ) : null} + {entry.pendingMessage?.attachments.map((attachment) => + attachment.type === "image" && attachment.uploadedAttachmentId ? ( + + ) : attachment.type === "image" ? ( + + ) : ( + + ), + )} {attachments.map((attachment) => { return isImageAttachment(attachment) ? ( - {timestampLabel} + {entry.pendingMessage && !entry.acknowledged ? "Pending" : timestampLabel} + {entry.pendingMessage && + !entry.acknowledged && + !entry.pendingMessage.creation && + entry.pendingMessage.messageId !== props.dispatchingMessageId ? ( + { + if (entry.pendingMessage) props.onEditPendingMessage(entry.pendingMessage); + }} + > + + + ) : null} {message.text.trim().length > 0 ? ( ) : null} - + ); } @@ -2227,6 +2268,7 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { // Keep row-local interaction props in extraData so disclosures and copy feedback repaint. const listAppearanceData = useMemo( () => ({ + dispatchingMessageId: props.dispatchingMessageId, copiedRowId, expandedWorkRows, workRowSizing, @@ -2238,6 +2280,7 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { viewportWidth, }), [ + props.dispatchingMessageId, copiedRowId, expandedWorkRows, workRowSizing, @@ -2382,14 +2425,19 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { }, [expandedWorkGroups]); const presentedFeed = useMemo( () => - deriveThreadFeedPresentation( + appendPendingThreadMessages( + deriveThreadFeedPresentation( + props.feed, + props.latestTurn, + expandedTurnIds, + expandedWorkGroupIds, + props.activeWorkStartedAt, + ), props.feed, - props.latestTurn, - expandedTurnIds, - expandedWorkGroupIds, - props.activeWorkStartedAt, + props.queuedMessages, ), [ + props.queuedMessages, expandedTurnIds, expandedWorkGroupIds, props.activeWorkStartedAt, @@ -2401,7 +2449,7 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { // content-inset override. Seed the fresh instance synchronously with the // current overlay height before the scroll integration's next reaction; // on Android the declarative contentInset floor covers this same window. - const listMountKey = `${feedThreadKey}:${props.feed.length === 0 ? "empty" : "filled"}`; + const listMountKey = `${feedThreadKey}:${presentedFeed.length === 0 ? "empty" : "filled"}`; useLayoutEffect(() => { const bottom = props.contentInsetEndAdjustment.value; if (bottom > 0) { @@ -2646,7 +2694,7 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { // Disclosures can mount existing offscreen rows as well as new work rows. // Fade those in after movement; never retain removed rows over replacements. const renderItem = useCallback( - (info: { item: ThreadFeedEntry; index: number }) => ( + (info: { item: PendingThreadFeedEntry; index: number }) => ( {renderFeedEntry(info, { environmentId: props.environmentId, + dispatchingMessageId: props.dispatchingMessageId, + onEditPendingMessage: props.onEditPendingMessage, copiedRowId, expandedWorkRows, workRowSizing, @@ -2685,6 +2735,8 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { ), [ + props.dispatchingMessageId, + props.onEditPendingMessage, copiedRowId, disclosureToggleSettling, expandedWorkRows, @@ -2716,7 +2768,7 @@ export const ThreadFeed = memo(function ThreadFeed(props: ThreadFeedProps) { ], ); - if (props.contentPresentation.kind === "unavailable") { + if (props.contentPresentation.kind === "unavailable" && props.queuedMessages.length === 0) { return ( - {props.feed.length === 0 && + {presentedFeed.length === 0 && props.activeWorkStartedAt === null && props.contentPresentation.kind === "ready" ? ( diff --git a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx index a59fa1a4450c..549bf2510381 100644 --- a/apps/mobile/src/features/threads/ThreadRouteScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadRouteScreen.tsx @@ -807,6 +807,8 @@ function ThreadRouteContent( projectWorkspaceRoot={selectedThreadProject?.workspaceRoot ?? null} threadCwd={selectedThreadCwd} selectedThreadQueueCount={composer.selectedThreadQueueCount} + queuedMessages={composer.selectedThreadQueuedMessages} + dispatchingMessageId={composer.dispatchingQueuedMessageId} layoutVariant={layout.variant} usesAutomaticContentInsets={usesNativeHeaderGlass} onOpenConnectionEditor={handleOpenConnectionEditor} diff --git a/apps/mobile/src/features/threads/pending-thread-feed.test.ts b/apps/mobile/src/features/threads/pending-thread-feed.test.ts new file mode 100644 index 000000000000..2cc0c5d88a20 --- /dev/null +++ b/apps/mobile/src/features/threads/pending-thread-feed.test.ts @@ -0,0 +1,41 @@ +import { describe, expect, it } from "vite-plus/test"; +import { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import type { QueuedThreadMessage } from "../../state/thread-outbox-model"; +import { appendPendingThreadMessages } from "./pending-thread-feed"; + +const pending = (id: string): QueuedThreadMessage => ({ + environmentId: EnvironmentId.make("env"), + threadId: ThreadId.make("thread"), + messageId: MessageId.make(id), + commandId: CommandId.make(id), + text: id, + attachments: [], + createdAt: "2026-09-06T10:00:00.000Z", +}); + +describe("pending timeline messages", () => { + it("keeps pending messages after newer agent activity in queue order", () => { + const activity = { + type: "thinking", + turnId: null, + id: "thinking", + createdAt: "2026-09-06T11:00:00.000Z", + } as const; + const entries = appendPendingThreadMessages( + [activity], + [], + [pending("first"), pending("second")], + ); + expect(entries.map((entry) => entry.id)).toEqual(["thinking", "first", "second"]); + expect(entries[1]?.pendingMessage?.text).toBe("first"); + }); + + it("reuses the message id and suppresses the pending copy when delivery appears", () => { + const queued = pending("sent"); + const optimistic = appendPendingThreadMessages([], [], [queued])[0]!; + const delivered = { ...optimistic, pendingMessage: undefined }; + expect(appendPendingThreadMessages([delivered], [delivered], [queued])).toEqual([delivered]); + // Folded messages still count as delivered even when absent from the presented rows. + expect(appendPendingThreadMessages([], [delivered], [queued])).toEqual([]); + }); +}); diff --git a/apps/mobile/src/features/threads/pending-thread-feed.ts b/apps/mobile/src/features/threads/pending-thread-feed.ts new file mode 100644 index 000000000000..84fae37fca54 --- /dev/null +++ b/apps/mobile/src/features/threads/pending-thread-feed.ts @@ -0,0 +1,39 @@ +import type { ThreadFeedEntry } from "../../lib/threadActivity"; +import type { QueuedThreadMessage } from "../../state/thread-outbox-model"; + +export type PendingThreadFeedEntry = ThreadFeedEntry & { + readonly pendingMessage?: QueuedThreadMessage; + readonly acknowledged?: boolean; +}; + +/** Append the outbox after all presented activity, until the server echoes each message. */ +export function appendPendingThreadMessages( + presentedFeed: ReadonlyArray, + feed: ReadonlyArray, + queuedMessages: ReadonlyArray, +): ReadonlyArray { + if (queuedMessages.length === 0) return presentedFeed; + const deliveredIds = new Set( + feed.flatMap((entry) => (entry.type === "message" ? [entry.message.id] : [])), + ); + return [ + ...presentedFeed, + ...queuedMessages + .filter((message) => !deliveredIds.has(message.messageId)) + .map((pendingMessage): PendingThreadFeedEntry => ({ + type: "message", + id: pendingMessage.messageId, + createdAt: pendingMessage.createdAt, + pendingMessage, + message: { + id: pendingMessage.messageId, + role: "user", + text: pendingMessage.text, + createdAt: pendingMessage.createdAt, + updatedAt: pendingMessage.createdAt, + turnId: null, + streaming: false, + }, + })), + ]; +} diff --git a/apps/mobile/src/state/acknowledged-thread-messages.ts b/apps/mobile/src/state/acknowledged-thread-messages.ts new file mode 100644 index 000000000000..0ceeb490302e --- /dev/null +++ b/apps/mobile/src/state/acknowledged-thread-messages.ts @@ -0,0 +1,27 @@ +import { Atom } from "effect/unstable/reactivity"; +import { appAtomRegistry } from "./atom-registry"; +import type { QueuedThreadMessage } from "./thread-outbox-model"; + +// A command acknowledgment can precede its message in the subscribed timeline. +// Keep the visible row until that projection arrives, independently of outbox cleanup. +export const acknowledgedThreadMessagesAtom = Atom.make>( + [], +).pipe(Atom.keepAlive); + +export function retainAcknowledgedThreadMessage(message: QueuedThreadMessage) { + const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom); + appAtomRegistry.set( + acknowledgedThreadMessagesAtom, + current.some((entry) => entry.messageId === message.messageId) + ? current + : [...current, message], + ); +} + +export function forgetAcknowledgedThreadMessage(message: QueuedThreadMessage) { + const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom); + appAtomRegistry.set( + acknowledgedThreadMessagesAtom, + current.filter((entry) => entry.messageId !== message.messageId), + ); +} diff --git a/apps/mobile/src/state/edit-pending-thread-message.test.ts b/apps/mobile/src/state/edit-pending-thread-message.test.ts new file mode 100644 index 000000000000..c9c8b4f6a47f --- /dev/null +++ b/apps/mobile/src/state/edit-pending-thread-message.test.ts @@ -0,0 +1,109 @@ +import { beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts"; +import type { QueuedThreadMessage } from "./thread-outbox-model"; + +const state = vi.hoisted(() => ({ + dispatching: null as string | null, + held: {} as Record, + draft: { text: "Existing draft", attachments: [] as { id: string }[] }, + confirm: vi.fn(async () => true), + remove: vi.fn(async () => true), + flush: vi.fn(async () => {}), +})); +vi.mock("./atom-registry", () => ({ + appAtomRegistry: { + get: (atom: string) => (atom === "dispatching" ? state.dispatching : state.held), + }, +})); +vi.mock("./use-thread-outbox", () => ({ + dispatchingQueuedMessageIdAtom: "dispatching", + editingQueuedMessageIdsAtom: "editing", + holdEditingQueuedMessage: (id: string) => { + state.held[id] = true; + }, + releaseEditingQueuedMessage: (id: string) => { + delete state.held[id]; + }, +})); +vi.mock("./thread-outbox", () => ({ + confirmThreadOutboxMessageQueued: state.confirm, + threadOutboxRevision: () => 1, +})); +vi.mock("./thread-outbox-removal", () => ({ removeThreadOutboxMessage: state.remove })); +vi.mock("./use-composer-drafts", () => ({ + waitForComposerDraftsLoaded: async () => {}, + getComposerDraftSnapshot: () => state.draft, + mergeComposerDraftContent: async (_key: string, message: QueuedThreadMessage) => { + state.draft = { + text: `${state.draft.text}\n\n${message.text}`, + attachments: [...state.draft.attachments, ...message.attachments], + }; + }, + updateComposerDraftSettings: () => {}, + flushComposerDrafts: state.flush, + undoComposerDraftMerge: async (_key: string, snapshot: typeof state.draft) => { + state.draft = snapshot; + }, +})); +import { editPendingThreadMessage } from "./edit-pending-thread-message"; + +const message: QueuedThreadMessage = { + environmentId: EnvironmentId.make("env"), + threadId: ThreadId.make("thread"), + messageId: MessageId.make("message"), + commandId: CommandId.make("command"), + text: "Queued task", + createdAt: "2026-09-06T10:00:00.000Z", + attachments: [ + { + id: "file", + type: "file", + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 10, + fileUri: "file:///notes.txt", + }, + ], +}; +beforeEach(() => { + vi.clearAllMocks(); + state.dispatching = null; + state.held = {}; + state.draft = { text: "Existing draft", attachments: [] }; + state.confirm.mockResolvedValue(true); + state.remove.mockResolvedValue(true); + state.flush.mockResolvedValue(undefined); +}); +describe("editing a pending message", () => { + it("locks delivery and persists text and attachments before removing the queued copy", async () => { + state.confirm.mockImplementationOnce(async () => { + expect(state.held[message.messageId]).toBe(true); + return true; + }); + state.remove.mockImplementationOnce(async () => { + expect(state.flush).toHaveBeenCalled(); + expect(state.draft.text).toBe("Existing draft\n\nQueued task"); + expect(state.draft.attachments).toEqual(message.attachments); + return true; + }); + expect(await editPendingThreadMessage(message)).toBe(true); + expect(state.held).toEqual({}); + }); + it("does not reclaim a message already being dispatched", async () => { + state.dispatching = message.messageId; + expect(await editPendingThreadMessage(message)).toBe(false); + expect(state.confirm).not.toHaveBeenCalled(); + expect(state.draft.text).toBe("Existing draft"); + }); + it("rolls back the draft if removing the queued message fails", async () => { + state.remove.mockRejectedValueOnce(new Error("disk error")); + await expect(editPendingThreadMessage(message)).rejects.toThrow("disk error"); + expect(state.draft).toEqual({ text: "Existing draft", attachments: [] }); + expect(state.held).toEqual({}); + }); + it("rolls back when a newer queue revision wins", async () => { + state.remove.mockResolvedValueOnce(false); + expect(await editPendingThreadMessage(message)).toBe(false); + expect(state.draft.text).toBe("Existing draft"); + }); +}); diff --git a/apps/mobile/src/state/edit-pending-thread-message.ts b/apps/mobile/src/state/edit-pending-thread-message.ts new file mode 100644 index 000000000000..e2f3a2bfcb5d --- /dev/null +++ b/apps/mobile/src/state/edit-pending-thread-message.ts @@ -0,0 +1,72 @@ +import { PROVIDER_SEND_TURN_MAX_ATTACHMENTS } from "@t3tools/contracts"; +import { scopedThreadKey } from "../lib/scopedEntities"; +import { appAtomRegistry } from "./atom-registry"; +import { + confirmThreadOutboxMessageQueued, + threadOutboxRevision, + type QueuedThreadMessage, +} from "./thread-outbox"; +import { removeThreadOutboxMessage } from "./thread-outbox-removal"; +import { + flushComposerDrafts, + getComposerDraftSnapshot, + mergeComposerDraftContent, + undoComposerDraftMerge, + updateComposerDraftSettings, + waitForComposerDraftsLoaded, +} from "./use-composer-drafts"; +import { + dispatchingQueuedMessageIdAtom, + editingQueuedMessageIdsAtom, + holdEditingQueuedMessage, + releaseEditingQueuedMessage, +} from "./use-thread-outbox"; + +/** Take delivery ownership before any await; the durable draft then takes ownership of the files. */ +export async function editPendingThreadMessage(message: QueuedThreadMessage): Promise { + if ( + message.creation || + appAtomRegistry.get(dispatchingQueuedMessageIdAtom) === message.messageId || + appAtomRegistry.get(editingQueuedMessageIdsAtom)[message.messageId] + ) { + return false; + } + holdEditingQueuedMessage(message.messageId); + const draftKey = scopedThreadKey(message.environmentId, message.threadId); + let rollback: { + snapshot: ReturnType; + merged: ReturnType; + } | null = null; + try { + if (!(await confirmThreadOutboxMessageQueued(message))) return false; + const revision = threadOutboxRevision(message.messageId); + await waitForComposerDraftsLoaded(); + const snapshot = getComposerDraftSnapshot(draftKey); + const attachmentIds = new Set(snapshot.attachments.map((attachment) => attachment.id)); + for (const attachment of message.attachments) attachmentIds.add(attachment.id); + if (attachmentIds.size > PROVIDER_SEND_TURN_MAX_ATTACHMENTS) { + throw new Error("Remove attachments from the composer before editing this message."); + } + try { + await mergeComposerDraftContent(draftKey, message); + } finally { + rollback = { snapshot, merged: getComposerDraftSnapshot(draftKey) }; + } + updateComposerDraftSettings(draftKey, { + ...(message.modelSelection ? { modelSelection: message.modelSelection } : {}), + ...(message.runtimeMode ? { runtimeMode: message.runtimeMode } : {}), + ...(message.interactionMode ? { interactionMode: message.interactionMode } : {}), + }); + rollback = { snapshot, merged: getComposerDraftSnapshot(draftKey) }; + await flushComposerDrafts(); + if (!(await removeThreadOutboxMessage(message, revision))) return false; + rollback = null; + return true; + } finally { + try { + if (rollback) await undoComposerDraftMerge(draftKey, rollback.snapshot, rollback.merged); + } finally { + releaseEditingQueuedMessage(message.messageId); + } + } +} diff --git a/apps/mobile/src/state/use-thread-composer-state.ts b/apps/mobile/src/state/use-thread-composer-state.ts index b9362ba8f172..7207b1d46a5c 100644 --- a/apps/mobile/src/state/use-thread-composer-state.ts +++ b/apps/mobile/src/state/use-thread-composer-state.ts @@ -33,6 +33,8 @@ import { import type { DraftComposerImageAttachment } from "../lib/composerImages"; import { scopedThreadKey } from "../lib/scopedEntities"; import { buildThreadFeed } from "../lib/threadActivity"; +import { acknowledgedThreadMessagesAtom } from "./acknowledged-thread-messages"; +import { appendPendingThreadMessages } from "../features/threads/pending-thread-feed"; import { appAtomRegistry } from "../state/atom-registry"; import { appendComposerDraftAttachments, @@ -102,6 +104,7 @@ export function useThreadComposerState() { const { selectedThread: selectedThreadShell, selectedEnvironmentRuntime } = useThreadSelection(); const selectedThreadDetail = useSelectedThreadDetail(); const composerDrafts = useAtomValue(composerDraftsAtom); + const acknowledgedMessages = useAtomValue(acknowledgedThreadMessagesAtom); const queuedMessagesByThreadKey = useThreadOutboxMessages(); const dispatchingQueuedMessageId = useAtomValue(dispatchingQueuedMessageIdAtom); const [feedbackSubmissionsByThreadKey, setFeedbackSubmissionsByThreadKey] = useState< @@ -138,16 +141,41 @@ export function useThreadComposerState() { ); const selectedThreadMessages = selectedThreadDetail?.messages; const selectedThreadActivities = selectedThreadDetail?.activities; - const selectedThreadFeed = useMemo( - () => + const selectedThreadFeed = useMemo(() => { + const feed = selectedThreadMessages && selectedThreadActivities ? buildThreadFeed({ messages: selectedThreadMessages, activities: selectedThreadActivities, }) - : [], - [selectedThreadActivities, selectedThreadMessages], - ); + : []; + const pendingAcknowledgments = acknowledgedMessages.filter( + (message) => + scopedThreadKey(message.environmentId, message.threadId) === selectedThreadKey && + !selectedThreadQueuedMessages.some((queued) => queued.messageId === message.messageId), + ); + if (pendingAcknowledgments.length === 0) return feed; + return appendPendingThreadMessages(feed, feed, pendingAcknowledgments).map((entry) => + entry.pendingMessage ? { ...entry, acknowledged: true } : entry, + ); + }, [ + selectedThreadActivities, + selectedThreadMessages, + selectedThreadKey, + selectedThreadQueuedMessages, + acknowledgedMessages, + ]); + useEffect(() => { + const echoedIds = new Set(selectedThreadMessages?.map((message) => message.id)); + if (acknowledgedMessages.some((message) => echoedIds.has(message.messageId))) { + appAtomRegistry.set( + acknowledgedThreadMessagesAtom, + appAtomRegistry + .get(acknowledgedThreadMessagesAtom) + .filter((message) => !echoedIds.has(message.messageId)), + ); + } + }, [acknowledgedMessages, selectedThreadMessages]); const selectedDraft = selectedThreadKey ? composerDrafts[selectedThreadKey] : null; const draftMessage = selectedDraft?.text ?? ""; @@ -558,6 +586,8 @@ export function useThreadComposerState() { dismissFeedback, selectedThreadFeed, selectedThreadQueueCount, + selectedThreadQueuedMessages, + dispatchingQueuedMessageId, activeWorkStartedAt, isCompacting, draftMessage, diff --git a/apps/mobile/src/state/use-thread-outbox-drain.test.ts b/apps/mobile/src/state/use-thread-outbox-drain.test.ts index bc295054038b..5e67fe57b840 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.test.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.test.ts @@ -1,3 +1,4 @@ +import { acknowledgedThreadMessagesAtom } from "./acknowledged-thread-messages"; import { CommandId, EnvironmentId, @@ -193,6 +194,7 @@ function remainingMessages(): ReadonlyArray { } beforeEach(() => { + appAtomRegistry.set(acknowledgedThreadMessagesAtom, []); harness.draftFile.setDocument({ schemaVersion: 1, drafts: {} }); }); @@ -431,6 +433,7 @@ describe("thread outbox drain delivery cleanup", () => { await expect(completeQueuedMessageDelivery(message, deliveryRevision)).resolves.toBe("removed"); expect(remainingMessages()).toEqual([]); + expect(appAtomRegistry.get(acknowledgedThreadMessagesAtom)).toEqual([message]); }); it("keeps a delivered message when its editor opens during storage removal", async () => { diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 9f1b6422ba53..0a197ceb8727 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -22,6 +22,10 @@ import { buildProjectThreadStartTurnInput } from "../lib/projectThreadStartTurn" import { prepareTurnAttachments, type PreparedTurnAttachments } from "../lib/attachmentUpload"; import { randomHex } from "../lib/uuid"; import { isModelSelectionUnavailable } from "../lib/modelOptions"; +import { + retainAcknowledgedThreadMessage, + forgetAcknowledgedThreadMessage, +} from "./acknowledged-thread-messages"; import { appAtomRegistry } from "./atom-registry"; import { useProjects, useServerConfigs, useThreadShells } from "./entities"; import { serverEnvironment } from "./server"; @@ -196,6 +200,7 @@ export async function completeQueuedMessageDelivery( if (appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId]) { return "edited"; } + retainAcknowledgedThreadMessage(queuedMessage); // Removal also releases the message's local attachment files. const removed = await removeThreadOutboxMessage( queuedMessage, @@ -203,6 +208,7 @@ export async function completeQueuedMessageDelivery( () => !appAtomRegistry.get(editingQueuedMessageIdsAtom)[queuedMessage.messageId], ); if (!removed) { + forgetAcknowledgedThreadMessage(queuedMessage); console.warn( "[thread-outbox] delivered message was edited before cleanup; keeping the newer message", { @@ -215,6 +221,7 @@ export async function completeQueuedMessageDelivery( } return "removed"; } catch (error) { + forgetAcknowledgedThreadMessage(queuedMessage); console.warn("[thread-outbox] failed to remove delivered queued message", { environmentId: queuedMessage.environmentId, threadId: queuedMessage.threadId,