From ce223ffe8e07f340d0d2a909831adc5deeb958f1 Mon Sep 17 00:00:00 2001 From: Tom Owers Date: Wed, 15 Jul 2026 11:46:42 +0100 Subject: [PATCH] feat(mobile): reorder and edit queued messages in place (port #3242) Ports the desktop queued-messages improvements to the mobile app. - Reorder queued messages with per-card move up/down controls. - Edit a queued message in place: pull it into the composer while it stays in the queue, then save it back into position (or cancel unchanged). - Drain boundary: while a message is being edited, only the messages queued before it auto-send when the turn ends; the edited message and everything after it stay queued until the edit is saved or cancelled. Generated-By: PostHog Code Task-Id: 8d0379e9-fb51-4ce7-9eb9-8f5b7a508bf0 --- apps/mobile/src/app/task/[id].tsx | 49 +++- .../tasks/composer/QueuedMessagesDock.tsx | 114 ++++++++-- .../tasks/composer/TaskChatComposer.tsx | 25 +++ .../tasks/stores/messageQueueStore.test.ts | 212 +++++++++++++++++- .../tasks/stores/messageQueueStore.ts | 118 +++++++++- .../tasks/stores/taskSessionStore.test.ts | 74 ++++++ .../features/tasks/stores/taskSessionStore.ts | 23 +- 7 files changed, 579 insertions(+), 36 deletions(-) diff --git a/apps/mobile/src/app/task/[id].tsx b/apps/mobile/src/app/task/[id].tsx index 0a7c2873ab..1375725a49 100644 --- a/apps/mobile/src/app/task/[id].tsx +++ b/apps/mobile/src/app/task/[id].tsx @@ -40,6 +40,7 @@ import { } from "@/features/tasks/hooks/useMessagingMode"; import { taskKeys } from "@/features/tasks/hooks/useTasks"; import { + type MoveDirection, type QueuedMessage, useMessageQueueStore, } from "@/features/tasks/stores/messageQueueStore"; @@ -109,6 +110,7 @@ export default function TaskDetailScreen() { getSessionForTask, setFocusedTaskId, steerQueuedMessage, + flushQueuedMessagesIfIdle, stopRun, } = useTaskSessionStore(); @@ -174,6 +176,9 @@ export default function TaskDetailScreen() { const messagingMode = useMessagingMode(taskId); const queuedCount = useQueuedCount(taskId); + const editingQueuedId = useMessageQueueStore((s) => + taskId ? s.editingByTaskId[taskId] : undefined, + ); const toggleMessagingMode = useToggleMessagingMode(taskId); const analytics = useAnalytics(); @@ -359,6 +364,18 @@ export default function TaskDetailScreen() { if (!taskId) return; Haptics.impactAsync(Haptics.ImpactFeedbackStyle.Light); + // Saving an in-place edit: overwrite the queued message and release the + // drain hold. If the turn already ended while editing, flush now — the + // turn-end drain won't fire again on its own. + const queue = useMessageQueueStore.getState(); + const editingId = queue.editingByTaskId[taskId]; + if (editingId) { + queue.update(taskId, editingId, { content: text, attachments }); + queue.clearEditing(taskId); + flushQueuedMessagesIfIdle(taskId); + return; + } + if (session?.terminalStatus) { handleSendAfterTerminal(text, attachments); return; @@ -398,6 +415,7 @@ export default function TaskDetailScreen() { messagingMode, handleSendAfterTerminal, trackPromptSent, + flushQueuedMessagesIfIdle, ], ); @@ -422,10 +440,13 @@ export default function TaskDetailScreen() { [taskId, steerQueuedMessage, trackPromptSent], ); - const handleReturnQueuedToComposer = useCallback( + // Pull a queued message into the composer for an in-place edit. It stays in + // the queue at its position (marked as the edit target); the next send saves + // it back rather than sending a new prompt. + const handleEditQueued = useCallback( (message: QueuedMessage) => { if (!taskId) return; - useMessageQueueStore.getState().remove(taskId, message.id); + useMessageQueueStore.getState().setEditing(taskId, message.id); setRestoredDraft({ text: message.content, attachments: message.attachments, @@ -434,10 +455,29 @@ export default function TaskDetailScreen() { [taskId], ); + const handleCancelEdit = useCallback(() => { + if (!taskId) return; + useMessageQueueStore.getState().clearEditing(taskId); + setRestoredDraft({ text: "", attachments: [] }); + flushQueuedMessagesIfIdle(taskId); + }, [taskId, flushQueuedMessagesIfIdle]); + + const handleMoveQueued = useCallback( + (message: QueuedMessage, direction: MoveDirection) => { + if (!taskId) return; + Haptics.selectionAsync(); + useMessageQueueStore.getState().move(taskId, message.id, direction); + }, + [taskId], + ); + const handleDiscardQueued = useCallback( (message: QueuedMessage) => { if (!taskId) return; + const wasEditing = + useMessageQueueStore.getState().editingByTaskId[taskId] === message.id; useMessageQueueStore.getState().remove(taskId, message.id); + if (wasEditing) setRestoredDraft({ text: "", attachments: [] }); }, [taskId], ); @@ -718,13 +758,16 @@ export default function TaskDetailScreen() { !session?.terminalStatus } onSteer={handleSteerQueued} - onReturnToComposer={handleReturnQueuedToComposer} + onEdit={handleEditQueued} onDiscard={handleDiscardQueued} + onMove={handleMoveQueued} /> ) : null} void; - onReturnToComposer: (message: QueuedMessage) => void; + onEdit: (message: QueuedMessage) => void; onDiscard: (message: QueuedMessage) => void; + onMove: (message: QueuedMessage, direction: MoveDirection) => void; } function previewText(message: QueuedMessage): string { @@ -33,11 +37,13 @@ export function QueuedMessagesDock({ taskId, canSteer, onSteer, - onReturnToComposer, + onEdit, onDiscard, + onMove, }: QueuedMessagesDockProps) { const themeColors = useThemeColors(); const queued = useMessageQueueStore((s) => s.queuesByTaskId[taskId]); + const editingId = useMessageQueueStore((s) => s.editingByTaskId[taskId]); const [activeId, setActiveId] = useState(null); if (!queued || queued.length === 0) return null; @@ -46,24 +52,58 @@ export function QueuedMessagesDock({ return ( <> - {queued.map((message) => ( - setActiveId(message.id)} - accessibilityRole="button" - accessibilityLabel="Queued message actions" - className="flex-row items-center gap-2 rounded-xl border border-gray-6 bg-card px-3 py-2 active:opacity-70" - > - - - {previewText(message)} - - {message.attachments.length > 0 ? ( - - ) : null} - Queued - - ))} + {queued.map((message, index) => { + const isEditing = message.id === editingId; + return ( + + setActiveId(message.id)} + accessibilityRole="button" + accessibilityLabel="Queued message actions" + className="min-w-0 flex-1 flex-row items-center gap-2 active:opacity-70" + > + + + {previewText(message)} + + {message.attachments.length > 0 ? ( + + ) : null} + + {isEditing ? "Editing" : "Queued"} + + + + } + label="Move up" + disabled={index === 0} + onPress={() => onMove(message, "up")} + /> + } + label="Move down" + disabled={index === queued.length - 1} + onPress={() => onMove(message, "down")} + /> + + + ); + })} setActiveId(null)}> @@ -93,10 +133,10 @@ export function QueuedMessagesDock({ ) : null} } - label="Edit in composer" - description="Pull it back into the composer to revise" + label="Edit in place" + description="Revise it in the composer; it keeps its place in the queue" onPress={() => { - onReturnToComposer(active); + onEdit(active); setActiveId(null); }} /> @@ -116,6 +156,34 @@ export function QueuedMessagesDock({ ); } +function ReorderButton({ + icon, + label, + disabled, + onPress, +}: { + icon: ReactNode; + label: string; + disabled: boolean; + onPress: () => void; +}) { + return ( + + {icon} + + ); +} + function ActionRow({ icon, label, diff --git a/apps/mobile/src/features/tasks/composer/TaskChatComposer.tsx b/apps/mobile/src/features/tasks/composer/TaskChatComposer.tsx index a8fafb7fca..1aa452517a 100644 --- a/apps/mobile/src/features/tasks/composer/TaskChatComposer.tsx +++ b/apps/mobile/src/features/tasks/composer/TaskChatComposer.tsx @@ -1,3 +1,4 @@ +import { Text } from "@components/text"; import * as Haptics from "expo-haptics"; import { ArrowUp, @@ -81,6 +82,9 @@ interface TaskChatComposerProps { onToggleMessagingMode: () => void; /** A queued message pulled back for editing; pass a fresh object to restore. */ restoredDraft?: { text: string; attachments: PendingAttachment[] }; + /** True while editing a queued message in place; the next send saves it. */ + editing?: boolean; + onCancelEdit?: () => void; } function modeIcon(mode: ExecutionMode, color: string, size = 14): ReactNode { @@ -167,6 +171,8 @@ export function TaskChatComposer({ queuedCount, onToggleMessagingMode, restoredDraft, + editing = false, + onCancelEdit, }: TaskChatComposerProps) { const themeColors = useThemeColors(); const [message, setMessage] = useState(() => initialMessage ?? ""); @@ -266,6 +272,25 @@ export function TaskChatComposer({ + {editing ? ( + + + + Editing queued message + + + + Cancel + + + + ) : null} { beforeEach(() => { - useMessageQueueStore.setState({ queuesByTaskId: {} }, false); + useMessageQueueStore.setState( + { queuesByTaskId: {}, editingByTaskId: {} }, + false, + ); }); it("enqueues messages in FIFO order", () => { @@ -94,6 +97,213 @@ describe("messageQueueStore", () => { }); }); +describe("move", () => { + beforeEach(() => { + useMessageQueueStore.setState( + { queuesByTaskId: {}, editingByTaskId: {} }, + false, + ); + }); + + function seed(contents: string[]) { + const { enqueue, getQueue } = useMessageQueueStore.getState(); + for (const content of contents) enqueue("t1", content, []); + return getQueue("t1"); + } + + it("moves a message down one slot", () => { + const [a] = seed(["a", "b", "c"]); + useMessageQueueStore.getState().move("t1", a.id, "down"); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["b", "a", "c"]); + }); + + it("moves a message up one slot", () => { + const queue = seed(["a", "b", "c"]); + useMessageQueueStore.getState().move("t1", queue[2].id, "up"); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["a", "c", "b"]); + }); + + it.each([ + { name: "up at the head", index: 0, direction: "up" as const }, + { name: "down at the tail", index: 2, direction: "down" as const }, + ])("is a no-op moving $name", ({ index, direction }) => { + const queue = seed(["a", "b", "c"]); + useMessageQueueStore.getState().move("t1", queue[index].id, direction); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["a", "b", "c"]); + }); + + it("ignores an unknown id", () => { + seed(["a", "b"]); + useMessageQueueStore.getState().move("t1", "nope", "up"); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["a", "b"]); + }); +}); + +describe("edit in place", () => { + beforeEach(() => { + useMessageQueueStore.setState( + { queuesByTaskId: {}, editingByTaskId: {} }, + false, + ); + }); + + it("updates content and attachments in place, keeping id and position", () => { + const { enqueue, update, getQueue } = useMessageQueueStore.getState(); + enqueue("t1", "a", []); + enqueue("t1", "b", []); + const target = getQueue("t1")[0]; + + update("t1", target.id, { content: "edited", attachments: [image("x")] }); + + const queue = useMessageQueueStore.getState().getQueue("t1"); + expect(queue.map((m) => m.id)).toEqual([target.id, getQueue("t1")[1].id]); + expect(queue[0].content).toBe("edited"); + expect(queue[0].attachments.map((att) => att.id)).toEqual(["x"]); + expect(queue[1].content).toBe("b"); + }); + + it("is a no-op when the target id is gone", () => { + const { enqueue, update, getQueue } = useMessageQueueStore.getState(); + enqueue("t1", "a", []); + update("t1", "missing", { content: "edited", attachments: [] }); + expect(getQueue("t1")[0].content).toBe("a"); + }); + + it("set/clear stores and releases the edit hold", () => { + const { enqueue, getQueue, setEditing, clearEditing } = + useMessageQueueStore.getState(); + enqueue("t1", "a", []); + const id = getQueue("t1")[0].id; + + setEditing("t1", id); + expect(useMessageQueueStore.getState().editingByTaskId.t1).toBe(id); + + clearEditing("t1"); + expect(useMessageQueueStore.getState().editingByTaskId.t1).toBeUndefined(); + }); + + it("clears the edit hold when the edited message is removed", () => { + const { enqueue, getQueue, setEditing, remove } = + useMessageQueueStore.getState(); + enqueue("t1", "a", []); + enqueue("t1", "b", []); + const id = getQueue("t1")[0].id; + setEditing("t1", id); + + remove("t1", id); + + expect(useMessageQueueStore.getState().editingByTaskId.t1).toBeUndefined(); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["b"]); + }); + + it("keeps the edit hold when a different message is removed", () => { + const { enqueue, getQueue, setEditing, remove } = + useMessageQueueStore.getState(); + enqueue("t1", "a", []); + enqueue("t1", "b", []); + const editingId = getQueue("t1")[1].id; + setEditing("t1", editingId); + + remove("t1", getQueue("t1")[0].id); + + expect(useMessageQueueStore.getState().editingByTaskId.t1).toBe(editingId); + }); +}); + +describe("drain boundary (stopAtEdited)", () => { + beforeEach(() => { + useMessageQueueStore.setState( + { queuesByTaskId: {}, editingByTaskId: {} }, + false, + ); + }); + + function seed(contents: string[]) { + const { enqueue, getQueue } = useMessageQueueStore.getState(); + for (const content of contents) enqueue("t1", content, []); + return getQueue("t1"); + } + + it("drains only the messages before the edited one", () => { + const queue = seed(["a", "b", "c"]); + useMessageQueueStore.getState().setEditing("t1", queue[1].id); + + const drained = useMessageQueueStore + .getState() + .drain("t1", { stopAtEdited: true }); + + expect(drained.map((m) => m.content)).toEqual(["a"]); + // The edited message and everything after it stay queued. + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["b", "c"]); + }); + + it("drains nothing when the head message is being edited", () => { + const queue = seed(["a", "b"]); + useMessageQueueStore.getState().setEditing("t1", queue[0].id); + + const drained = useMessageQueueStore + .getState() + .drain("t1", { stopAtEdited: true }); + + expect(drained).toEqual([]); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["a", "b"]); + }); + + it("drains the whole queue when nothing is being edited", () => { + seed(["a", "b", "c"]); + const drained = useMessageQueueStore + .getState() + .drain("t1", { stopAtEdited: true }); + expect(drained.map((m) => m.content)).toEqual(["a", "b", "c"]); + expect(useMessageQueueStore.getState().getQueue("t1")).toEqual([]); + }); + + it("ignores the boundary without stopAtEdited, even mid-edit", () => { + const queue = seed(["a", "b", "c"]); + useMessageQueueStore.getState().setEditing("t1", queue[1].id); + + const drained = useMessageQueueStore.getState().drain("t1"); + + expect(drained.map((m) => m.content)).toEqual(["a", "b", "c"]); + expect(useMessageQueueStore.getState().getQueue("t1")).toEqual([]); + }); +}); + describe("combineQueuedMessages", () => { function msg( content: string, diff --git a/apps/mobile/src/features/tasks/stores/messageQueueStore.ts b/apps/mobile/src/features/tasks/stores/messageQueueStore.ts index b4d43b4a84..f676cb326f 100644 --- a/apps/mobile/src/features/tasks/stores/messageQueueStore.ts +++ b/apps/mobile/src/features/tasks/stores/messageQueueStore.ts @@ -9,30 +9,70 @@ export interface QueuedMessage { const EMPTY: QueuedMessage[] = []; +export type MoveDirection = "up" | "down"; + let queueIdCounter = 0; function nextQueueId(): string { queueIdCounter += 1; return `queue-${queueIdCounter}`; } +// A message being edited in place acts as a drain boundary: only the messages +// queued before it may auto-send; it and everything after stay put until the +// edit is saved or cancelled. Returns the full length when nothing is being +// edited (or the edited message has already left the queue). Mirrors +// `sendableQueuePrefixLength` in @posthog/shared, reimplemented here because +// mobile's queue rows (`{id, content, attachments}`) don't match the shared +// `AgentSession` message shape. +function sendablePrefixLength( + queue: QueuedMessage[], + editingId: string | undefined, +): number { + if (!editingId) return queue.length; + const index = queue.findIndex((m) => m.id === editingId); + return index === -1 ? queue.length : index; +} + interface MessageQueueState { queuesByTaskId: Record; + /** Per-task id of the queued message currently open in the composer. */ + editingByTaskId: Record; enqueue: ( taskId: string, content: string, attachments: PendingAttachment[], ) => void; - /** Remove and return every queued message for a task, in FIFO order. */ - drain: (taskId: string) => QueuedMessage[]; + /** + * Remove and return queued messages from the head, in FIFO order. With + * `stopAtEdited`, stops at the in-place edit boundary so the edited message + * and everything after it stay queued. + */ + drain: ( + taskId: string, + options?: { stopAtEdited?: boolean }, + ) => QueuedMessage[]; /** Restore messages at the head of the queue, e.g. after a failed flush. */ prepend: (taskId: string, messages: QueuedMessage[]) => void; - /** Drop a single queued message by id. */ + /** Drop a single queued message by id. Clears the edit hold if it targeted it. */ remove: (taskId: string, messageId: string) => void; + /** Reorder a queued message one slot up or down; the send order follows. */ + move: (taskId: string, messageId: string, direction: MoveDirection) => void; + /** Replace a queued message's content/attachments in place, keeping position. */ + update: ( + taskId: string, + messageId: string, + patch: { content: string; attachments: PendingAttachment[] }, + ) => void; + /** Mark a queued message as being edited in the composer (drain boundary). */ + setEditing: (taskId: string, messageId: string) => void; + /** Release the in-place edit hold. */ + clearEditing: (taskId: string) => void; getQueue: (taskId: string) => QueuedMessage[]; } export const useMessageQueueStore = create((set, get) => ({ queuesByTaskId: {}, + editingByTaskId: {}, enqueue: (taskId, content, attachments) => set((state) => ({ queuesByTaskId: { @@ -43,14 +83,25 @@ export const useMessageQueueStore = create((set, get) => ({ ], }, })), - drain: (taskId) => { + drain: (taskId, options) => { const queued = get().queuesByTaskId[taskId] ?? EMPTY; if (queued.length === 0) return EMPTY; + const cutoff = options?.stopAtEdited + ? sendablePrefixLength(queued, get().editingByTaskId[taskId]) + : queued.length; + if (cutoff === 0) return EMPTY; + const drained = queued.slice(0, cutoff); + const rest = queued.slice(cutoff); set((state) => { - const { [taskId]: _drained, ...rest } = state.queuesByTaskId; - return { queuesByTaskId: rest }; + if (rest.length === 0) { + const { [taskId]: _drained, ...others } = state.queuesByTaskId; + return { queuesByTaskId: others }; + } + return { + queuesByTaskId: { ...state.queuesByTaskId, [taskId]: rest }, + }; }); - return queued; + return drained; }, prepend: (taskId, messages) => set((state) => ({ @@ -65,17 +116,68 @@ export const useMessageQueueStore = create((set, get) => ({ if (!queue) return state; const next = queue.filter((m) => m.id !== messageId); if (next.length === queue.length) return state; + const editingByTaskId = + state.editingByTaskId[taskId] === messageId + ? omit(state.editingByTaskId, taskId) + : state.editingByTaskId; if (next.length === 0) { const { [taskId]: _emptied, ...rest } = state.queuesByTaskId; - return { queuesByTaskId: rest }; + return { queuesByTaskId: rest, editingByTaskId }; } + return { + queuesByTaskId: { ...state.queuesByTaskId, [taskId]: next }, + editingByTaskId, + }; + }), + move: (taskId, messageId, direction) => + set((state) => { + const queue = state.queuesByTaskId[taskId]; + if (!queue) return state; + const from = queue.findIndex((m) => m.id === messageId); + if (from === -1) return state; + const to = direction === "up" ? from - 1 : from + 1; + if (to < 0 || to >= queue.length) return state; + const next = [...queue]; + const [moved] = next.splice(from, 1); + next.splice(to, 0, moved); return { queuesByTaskId: { ...state.queuesByTaskId, [taskId]: next }, }; }), + update: (taskId, messageId, patch) => + set((state) => { + const queue = state.queuesByTaskId[taskId]; + if (!queue?.some((m) => m.id === messageId)) return state; + const next = queue.map((m) => + m.id === messageId + ? { ...m, content: patch.content, attachments: patch.attachments } + : m, + ); + return { + queuesByTaskId: { ...state.queuesByTaskId, [taskId]: next }, + }; + }), + setEditing: (taskId, messageId) => + set((state) => ({ + editingByTaskId: { ...state.editingByTaskId, [taskId]: messageId }, + })), + clearEditing: (taskId) => + set((state) => + taskId in state.editingByTaskId + ? { editingByTaskId: omit(state.editingByTaskId, taskId) } + : state, + ), getQueue: (taskId) => get().queuesByTaskId[taskId] ?? EMPTY, })); +function omit>( + record: T, + key: string, +): Record { + const { [key]: _removed, ...rest } = record; + return rest; +} + /** * Combine buffered messages into a single prompt, preserving the order they * were typed: texts join with a blank line, attachments concatenate. diff --git a/apps/mobile/src/features/tasks/stores/taskSessionStore.test.ts b/apps/mobile/src/features/tasks/stores/taskSessionStore.test.ts index db60b6b667..ddc5da1459 100644 --- a/apps/mobile/src/features/tasks/stores/taskSessionStore.test.ts +++ b/apps/mobile/src/features/tasks/stores/taskSessionStore.test.ts @@ -141,6 +141,80 @@ describe("steerQueuedMessage", () => { }); }); +describe("flushQueuedMessagesIfIdle", () => { + beforeEach(() => { + useMessageQueueStore.setState( + { queuesByTaskId: {}, editingByTaskId: {} }, + false, + ); + useTaskSessionStore.setState({ sessions: {} }); + }); + + it("sends the queue when the agent is idle", async () => { + seedSession({ isPromptPending: false }); + const sendInterrupting = vi.fn(() => Promise.resolve()); + useTaskSessionStore.setState({ sendInterrupting }); + useMessageQueueStore.getState().enqueue("t1", "a", []); + useMessageQueueStore.getState().enqueue("t1", "b", []); + + useTaskSessionStore.getState().flushQueuedMessagesIfIdle("t1"); + await vi.waitFor(() => expect(sendInterrupting).toHaveBeenCalled()); + + expect(sendInterrupting).toHaveBeenCalledWith("t1", "a\n\nb", []); + expect(useMessageQueueStore.getState().getQueue("t1")).toEqual([]); + }); + + it("sends only the messages before the one being edited", async () => { + seedSession({ isPromptPending: false }); + const sendInterrupting = vi.fn(() => Promise.resolve()); + useTaskSessionStore.setState({ sendInterrupting }); + useMessageQueueStore.getState().enqueue("t1", "a", []); + useMessageQueueStore.getState().enqueue("t1", "b", []); + useMessageQueueStore.getState().enqueue("t1", "c", []); + const edited = useMessageQueueStore.getState().getQueue("t1")[1]; + useMessageQueueStore.getState().setEditing("t1", edited.id); + + useTaskSessionStore.getState().flushQueuedMessagesIfIdle("t1"); + await vi.waitFor(() => expect(sendInterrupting).toHaveBeenCalled()); + + expect(sendInterrupting).toHaveBeenCalledWith("t1", "a", []); + expect( + useMessageQueueStore + .getState() + .getQueue("t1") + .map((m) => m.content), + ).toEqual(["b", "c"]); + }); + + it.each([ + { name: "a turn is running", overrides: { isPromptPending: true } }, + { name: "the run is terminal", overrides: { terminalStatus: "completed" } }, + { name: "the agent is compacting", overrides: { isCompacting: true } }, + ] as const)("no-ops when $name", async ({ overrides }) => { + seedSession({ isPromptPending: false, ...overrides }); + const sendInterrupting = vi.fn(() => Promise.resolve()); + useTaskSessionStore.setState({ sendInterrupting }); + useMessageQueueStore.getState().enqueue("t1", "a", []); + + useTaskSessionStore.getState().flushQueuedMessagesIfIdle("t1"); + await Promise.resolve(); + + expect(sendInterrupting).not.toHaveBeenCalled(); + expect(useMessageQueueStore.getState().getQueue("t1")).toHaveLength(1); + }); + + it("no-ops when the queue is empty", async () => { + seedSession({ isPromptPending: false }); + const sendInterrupting = vi.fn(() => Promise.resolve()); + useTaskSessionStore.setState({ sendInterrupting }); + + useTaskSessionStore.getState().flushQueuedMessagesIfIdle("t1"); + await Promise.resolve(); + + expect(sendInterrupting).not.toHaveBeenCalled(); + }); +}); + describe("_resumeCloudRun", () => { const mockGetTask = vi.mocked(getTask); const mockRunTaskInCloud = vi.mocked(runTaskInCloud); diff --git a/apps/mobile/src/features/tasks/stores/taskSessionStore.ts b/apps/mobile/src/features/tasks/stores/taskSessionStore.ts index 01b4606282..fdf91dfe9d 100644 --- a/apps/mobile/src/features/tasks/stores/taskSessionStore.ts +++ b/apps/mobile/src/features/tasks/stores/taskSessionStore.ts @@ -360,6 +360,10 @@ interface TaskSessionStore { attachments?: PendingAttachment[], ) => Promise; flushQueuedMessages: (taskId: string) => Promise; + /** Flush the queue only if the agent is idle. Used after an in-place edit is + * saved or cancelled: the turn may have ended while the user was editing, so + * nothing else would trigger the turn-end drain. A no-op mid-turn. */ + flushQueuedMessagesIfIdle: (taskId: string) => void; /** Drop one queued message and resend it now as a steer (interrupt + resend). */ steerQueuedMessage: (taskId: string, messageId: string) => Promise; setConfigOption: ( @@ -864,7 +868,9 @@ export const useTaskSessionStore = create((set, get) => ({ if (flushingTasks.has(taskId)) return; flushingTasks.add(taskId); try { - const drained = useMessageQueueStore.getState().drain(taskId); + const drained = useMessageQueueStore + .getState() + .drain(taskId, { stopAtEdited: true }); if (drained.length === 0) return; const { text, attachments } = combineQueuedMessages(drained); @@ -882,6 +888,21 @@ export const useTaskSessionStore = create((set, get) => ({ } }, + flushQueuedMessagesIfIdle: (taskId: string) => { + const session = get().getSessionForTask(taskId); + if ( + session?.status === "connected" && + !session.isPromptPending && + !session.terminalStatus && + !session.isCompacting && + useMessageQueueStore.getState().getQueue(taskId).length > 0 + ) { + get() + .flushQueuedMessages(taskId) + .catch((err) => log.warn("Queue flush failed", err)); + } + }, + steerQueuedMessage: async (taskId: string, messageId: string) => { const session = get().getSessionForTask(taskId); // Steering only makes sense against a live turn. Mid-compaction it would