From 8ebd7fb0e0f4f737a516a3a8cae9c01b30ee91ee Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Sun, 8 Nov 2026 23:34:07 +0800 Subject: [PATCH] fix(desktop): preserve queued input and stable queue controls --- apps/desktop/docs/development.md | 23 +++ .../src/v2/client-queue-steering.test.ts | 147 ++++++++++++++ .../packages/devo-ai-sdk/src/v2/client.ts | 29 +-- .../chat/composer-queue-drag.test.tsx | 42 ++++ .../chat/composer-queue-row.test.tsx | 33 +++ .../components/chat/composer-status-stack.tsx | 55 +++-- .../hooks/use-composer-queue.test.tsx | 190 ++++++++++++++++++ .../src/renderer/hooks/use-composer-queue.ts | 84 +++++--- 8 files changed, 533 insertions(+), 70 deletions(-) create mode 100644 apps/desktop/packages/devo-ai-sdk/src/v2/client-queue-steering.test.ts create mode 100644 apps/desktop/src/renderer/components/chat/composer-queue-drag.test.tsx create mode 100644 apps/desktop/src/renderer/components/chat/composer-queue-row.test.tsx create mode 100644 apps/desktop/src/renderer/hooks/use-composer-queue.test.tsx diff --git a/apps/desktop/docs/development.md b/apps/desktop/docs/development.md index f85349c8..625425f1 100644 --- a/apps/desktop/docs/development.md +++ b/apps/desktop/docs/development.md @@ -39,3 +39,26 @@ a 500 ms graceful window before forced termination. Background commands are stopped across connections even when their session has no active model turn. Confirmed deletions leave the sidebar immediately. Failed deletions retain the folder and its connection so removal can be retried. + +## Queued follow-up controls + +Dragging queued messages tracks stable queue IDs and resolves Native's one-based +positions at drop time, so draining a previous message cannot move the wrong +entry. Steering +submits the complete queued input, including attachments, skills, and mentions, +rather than its display preview. Text-only editing is unavailable for messages +with non-text context so editing cannot silently discard it. + +Failed queue actions remain visible beside the message and can be retried. If +steering succeeded but removing the original queue entry failed, the control +changes to **Retry removal** and retries cleanup without steering the message +again. Pending operations and cleanup retries are scoped to their originating +session and survive navigation between conversations while the composer stays +mounted. This client-side state does not survive unmounting the composer or +restarting the app. + +Steering and removing the original queued input are still two separate requests, +not an atomic server-side queue-to-steering operation. A server queue drain +between them can still consume the original message. Exactly-once transfer +requires a future Native operation that claims the original queue entry and +deduplicates retries, including when a steering response is lost. diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client-queue-steering.test.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client-queue-steering.test.ts new file mode 100644 index 00000000..d29771c4 --- /dev/null +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client-queue-steering.test.ts @@ -0,0 +1,147 @@ +import { describe, expect, test } from "bun:test" +import { createDevoClient, type DevoNativeTransport, type DevoNativeTransportEvent } from "./client" + +const nativeSession = { + id: "session-1", + version: 1, + cwd: "/repo", + title: "Native session", + parent: null, + createdAt: "2026-08-22T00:00:00Z", + lastActivityAt: "2026-08-22T00:00:00Z", + status: "idle", + flags: [], + archived: false, + ephemeral: false, + model: { provider: "test", model: "test-model" }, + settings: { permissionProfile: "default" }, + preview: "", + queuedCount: 0, + usage: { + total: { + inputTokens: 0, + outputTokens: 0, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + reasoningTokens: 0, + totalTokens: 0, + callCount: 0, + meteredCallCount: 0, + failedCallCount: 0, + cancelledCallCount: 0, + }, + byPurpose: [], + updatedAt: "2026-08-22T00:00:00Z", + }, +} + +const nativeTurnInProgress = { + id: "turn-1", + sessionId: nativeSession.id, + sequence: 1, + kind: "regular", + status: "inProgress", + model: nativeSession.model, + startedAt: "2026-08-24T00:00:00Z", +} + + +const queuedInput = [ + { type: "text", text: "Review these references" }, + { type: "image", uri: "data:image/png;base64,aGVsbG8=", mimeType: "image/png", detail: "high" }, + { type: "localImage", path: "/repo/screenshot.png", detail: "auto" }, + { type: "audio", uri: "data:audio/wav;base64,aGVsbG8=", mimeType: "audio/wav" }, + { type: "skill", name: "review" }, + { type: "mention", uri: "file:///repo/notes.md" }, +] +const entry = { + queueItemId: "queue-1", + position: 1, + preview: "Review these references", + enqueuedAt: "2026-08-24T00:00:00Z", + input: queuedInput, +} +class QueueTransport implements DevoNativeTransport { + requests: Array<{ method: string; params: unknown }> = [] + listener?: (event: DevoNativeTransportEvent) => void + async request(method: string, params?: unknown): Promise { + this.requests.push({ method, params }) + switch (method) { + case "initialize": return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + case "session/new": return { session: nativeSession } + case "subscription/create": return { subscriptionId: "sub-1", cursors: [] } + case "subscription/ack": return { serverTimeMs: 1 } + case "session/queue/list": return { entries: [entry] } + case "turn/steer": return { outcome: "injected", itemId: "steered-1" } + default: throw new Error(`Unexpected method: ${method}`) + } + } + async respond() {} + subscribe(listener: (event: DevoNativeTransportEvent) => void) { + this.listener = listener + return () => { this.listener = undefined } + } + connected() { return true } +} +async function setup() { + const transport = new QueueTransport() + const client = createDevoClient({ directory: "/repo", transport }) + await client.session.create() + transport.listener?.({ type: "notification", method: "turn/started", params: { turn: nativeTurnInProgress } }) + return { client, transport } +} + +describe("Native queued steering", () => { + test("queue list and steer preserve every Native input field", async () => { + const { client, transport } = await setup() + try { + expect((await client.session.queue.list({ sessionId: nativeSession.id })).data.entries).toEqual([entry]) + await client.session.steer({ sessionId: nativeSession.id, queueItemId: entry.queueItemId }) + expect(transport.requests.filter(({ method }) => method === "turn/steer")).toEqual([{ + method: "turn/steer", + params: { + sessionId: nativeSession.id, + expectedTurnId: nativeTurnInProgress.id, + input: queuedInput, + idempotencyKey: expect.any(String), + }, + }]) + } finally { client.dispose() } + }) + + test("steering uses the latest queue notification instead of an earlier preview", async () => { + const { client, transport } = await setup() + try { + await client.session.queue.list({ sessionId: nativeSession.id }) + const updated = { ...entry, input: [{ type: "skill", name: "updated-skill" }] } + transport.listener?.({ type: "notification", method: "queue/updated", params: { + sessionId: nativeSession.id, change: "updated", queueItemId: entry.queueItemId, queue: [updated], + } }) + await client.session.steer({ sessionId: nativeSession.id, queueItemId: entry.queueItemId }) + expect(transport.requests.find(({ method }) => method === "turn/steer")?.params).toEqual({ + sessionId: nativeSession.id, expectedTurnId: nativeTurnInProgress.id, + input: updated.input, idempotencyKey: expect.any(String), + }) + } finally { client.dispose() } + }) + + test("a missing queue entry fails before sending or deleting anything", async () => { + const { client, transport } = await setup() + try { + await expect(client.session.steer({ sessionId: nativeSession.id, queueItemId: "missing" })) + .rejects.toThrow("Queued message is no longer available to steer") + expect(transport.requests.filter(({ method }) => method === "turn/steer" || method === "session/queue/remove")).toEqual([]) + } finally { client.dispose() } + }) + + test("existing direct-text steering keeps its request shape", async () => { + const { client, transport } = await setup() + try { + await client.session.steer({ sessionId: nativeSession.id, parts: [{ type: "text", text: "Direct steer" }] }) + expect(transport.requests.find(({ method }) => method === "turn/steer")?.params).toEqual({ + sessionId: nativeSession.id, expectedTurnId: nativeTurnInProgress.id, + input: [{ type: "text", text: "Direct steer" }], idempotencyKey: expect.any(String), + }) + } finally { client.dispose() } + }) +}) diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts index a64b77d7..da73e9d5 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts @@ -608,7 +608,7 @@ export type QueueWireEntry = { position: number preview: string enqueuedAt?: string - input?: Array<{ type: string; text?: string }> + input?: UserInput[] } function parseQueueWireEntries(value: unknown): QueueWireEntry[] { @@ -621,15 +621,10 @@ function parseQueueWireEntries(value: unknown): QueueWireEntry[] { position: Number(entry.position ?? 0), preview: String(entry.preview ?? ""), enqueuedAt: typeof entry.enqueuedAt === "string" ? entry.enqueuedAt : undefined, - input: Array.isArray(entry.input) - ? entry.input.map((part) => { - const record = objectRecord(part) - return { - type: String(record?.type ?? "text"), - text: typeof record?.text === "string" ? record.text : undefined, - } - }) - : undefined, + // Native results are schema-validated before reaching this parser. + // Retain every input field so steering cannot discard attachments, + // skill names, or file mentions from the queued message. + input: Array.isArray(entry.input) ? entry.input as UserInput[] : undefined, })) .filter((entry) => entry.queueItemId.length > 0) .sort((left, right) => left.position - right.position) @@ -967,15 +962,25 @@ class NativeClient { return { data: null } }, }, - steer: async (params: { sessionId: string; parts: PromptPartInput[] }) => { + steer: async (params: + | { sessionId: string; parts: PromptPartInput[] } + | { sessionId: string; queueItemId: string } + ) => { const expectedTurnId = this.activeTurnIds.get(params.sessionId) if (!expectedTurnId) { throw new Error("No active turn to steer") } + const input = "queueItemId" in params + ? this.queueEntriesForSession(params.sessionId) + .find((entry) => entry.queueItemId === params.queueItemId)?.input + : userInputsFromPromptParts(params.parts) + if (!input?.length) { + throw new Error("Queued message is no longer available to steer") + } const result = await this.requestCanonical("turn/steer", { sessionId: params.sessionId, expectedTurnId, - input: userInputsFromPromptParts(params.parts), + input, idempotencyKey: crypto.randomUUID(), }) return { data: result } diff --git a/apps/desktop/src/renderer/components/chat/composer-queue-drag.test.tsx b/apps/desktop/src/renderer/components/chat/composer-queue-drag.test.tsx new file mode 100644 index 00000000..0c719bcb --- /dev/null +++ b/apps/desktop/src/renderer/components/chat/composer-queue-drag.test.tsx @@ -0,0 +1,42 @@ +import { afterAll, expect, mock, test } from "bun:test" +import { Window } from "happy-dom" +import { act } from "react" +import { createRoot } from "react-dom/client" +import { ComposerStatusStack, type ComposerQueueItem } from "./composer-status-stack" + +const dom = new Window({ url: "http://localhost" }) +const original = Object.getOwnPropertyDescriptors(globalThis) +Object.assign(globalThis, { + window: dom, document: dom.document, navigator: dom.navigator, + IS_REACT_ACT_ENVIRONMENT: true, +}) +afterAll(() => { + for (const key of ["window", "document", "navigator", "IS_REACT_ACT_ENVIRONMENT"]) { + if (original[key]) Object.defineProperty(globalThis, key, original[key]) + else Reflect.deleteProperty(globalThis, key) + } + void dom.happyDOM.close() +}) + +test("drag retains message identity when the queue changes before drop", async () => { + const container = dom.document.createElement("div") + const root = createRoot(container as unknown as HTMLElement) + const onReorder = mock((_from: string, _to: string) => {}) + const items: ComposerQueueItem[] = ["A", "B", "C", "D"].map((id) => ({ id, text: id, status: "queued" })) + const render = async (queueItems: ComposerQueueItem[]) => { + await act(async () => { root.render() }) + } + try { + await render(items) + const start = new dom.Event("dragstart", { bubbles: true }) + Object.assign(start, { dataTransfer: { effectAllowed: "", setData: mock(() => {}) } }) + await act(async () => { container.querySelectorAll('[draggable="true"]')[1].dispatchEvent(start) }) + await render(items.slice(1)) + await act(async () => { + container.querySelectorAll('[draggable="true"]')[2].dispatchEvent(new dom.Event("drop", { bubbles: true })) + }) + expect(onReorder.mock.calls).toEqual([["B", "D"]]) + } finally { + await act(async () => { root.unmount() }) + } +}) diff --git a/apps/desktop/src/renderer/components/chat/composer-queue-row.test.tsx b/apps/desktop/src/renderer/components/chat/composer-queue-row.test.tsx new file mode 100644 index 00000000..59043cd7 --- /dev/null +++ b/apps/desktop/src/renderer/components/chat/composer-queue-row.test.tsx @@ -0,0 +1,33 @@ +import { describe, expect, test } from "bun:test" +import { renderToStaticMarkup } from "react-dom/server" +import { ComposerStatusStack, type ComposerQueueItem } from "./composer-status-stack" + +function render(item: Partial) { + return renderToStaticMarkup( {}} + onRemoveQueueItem={() => {}} + onSteerQueueItem={() => {}} + />) +} + +describe("composer queue row", () => { + test("announces action errors while keeping queued actions available for retry", () => { + const html = render({ error: "Connection unavailable" }) + expect(html).toContain('role="alert"') + expect(html).toContain("Connection unavailable") + expect(html).not.toContain('disabled=""') + }) + + test("prevents text editing from dropping skills or other non-text context", () => { + const html = render({ canEdit: false }) + const editButton = html.match(/]*aria-label="Edit queued message"[^>]*>/)?.[0] + expect(editButton).toContain('disabled=""') + }) + + test("labels a successful steer with failed cleanup as removal-only retry", () => { + const html = render({ steered: true, canEdit: false, error: "Remove unavailable" }) + expect(html).toContain("Retry removal") + expect(html).not.toContain(">Steer<") + }) +}) diff --git a/apps/desktop/src/renderer/components/chat/composer-status-stack.tsx b/apps/desktop/src/renderer/components/chat/composer-status-stack.tsx index c9336f00..db3494ad 100644 --- a/apps/desktop/src/renderer/components/chat/composer-status-stack.tsx +++ b/apps/desktop/src/renderer/components/chat/composer-status-stack.tsx @@ -33,6 +33,8 @@ export interface ComposerQueueItem { fileCount?: number createdAtMs?: number error?: string + canEdit?: boolean + steered?: boolean } interface ComposerStatusStackProps { @@ -47,7 +49,7 @@ interface ComposerStatusStackProps { onSteerQueueItem?: (item: ComposerQueueItem) => void onEditQueueItem?: (item: ComposerQueueItem) => void onRemoveQueueItem?: (item: ComposerQueueItem) => void - onReorderQueueItem?: (fromIndex: number, toIndex: number) => void + onReorderQueueItem?: (fromId: string, toId: string) => void onQueueDragStart?: (itemId: string) => void onQueueDragEnd?: () => void } @@ -135,21 +137,19 @@ function queueItemBusy(status: ComposerQueueItemStatus): boolean { interface QueueItemRowProps { item: ComposerQueueItem - index: number dragging?: boolean dragOver?: boolean onSteer?: (item: ComposerQueueItem) => void onEdit?: (item: ComposerQueueItem) => void onRemove?: (item: ComposerQueueItem) => void - onDragStart?: (index: number) => void - onDragEnter?: (index: number) => void + onDragStart?: (itemId: string) => void + onDragEnter?: (itemId: string) => void onDragEnd?: () => void - onDrop?: (index: number) => void + onDrop?: (itemId: string) => void } function QueueItemRow({ item, - index, dragging, dragOver, onSteer, @@ -162,7 +162,7 @@ function QueueItemRow({ }: QueueItemRowProps) { const busy = queueItemBusy(item.status) const canAct = item.status === "queued" - const canEdit = !busy && (item.fileCount ?? 0) === 0 + const canEdit = !busy && item.canEdit !== false && (item.fileCount ?? 0) === 0 const preview = queueRenderPreview(item.text) const handleDragStart = (event: DragEvent) => { @@ -172,19 +172,19 @@ function QueueItemRow({ } event.dataTransfer.effectAllowed = "move" event.dataTransfer.setData("text/plain", item.id) - onDragStart?.(index) + onDragStart?.(item.id) } const handleDragOver = (event: DragEvent) => { if (!canAct) return event.preventDefault() event.dataTransfer.dropEffect = "move" - onDragEnter?.(index) + onDragEnter?.(item.id) } const handleDrop = (event: DragEvent) => { event.preventDefault() - onDrop?.(index) + onDrop?.(item.id) } return ( @@ -214,8 +214,8 @@ function QueueItemRow({ {item.fileCount} file{item.fileCount === 1 ? "" : "s"} ) : null} - {item.status === "error" && item.error ? ( - {item.error} + {item.error ? ( + {item.error} ) : null}
@@ -230,7 +230,7 @@ function QueueItemRow({ ) : ( )} - Steer + {item.steered ? "Retry removal" : "Steer"} void onEdit?: (item: ComposerQueueItem) => void onRemove?: (item: ComposerQueueItem) => void - onReorder?: (fromIndex: number, toIndex: number) => void + onReorder?: (fromId: string, toId: string) => void onDragStart?: (itemId: string) => void onDragEnd?: () => void } @@ -278,37 +278,36 @@ function ComposerQueueList({ onDragStart, onDragEnd, }: ComposerQueueListProps) { - const [dragIndex, setDragIndex] = useState(null) - const [hoverIndex, setHoverIndex] = useState(null) + const [dragItemId, setDragItemId] = useState(null) + const [hoverItemId, setHoverItemId] = useState(null) if (items.length === 0) return null return ( <> - {items.map((item, index) => ( + {items.map((item) => ( { - setDragIndex(nextIndex) + onDragStart={(itemId) => { + setDragItemId(itemId) onDragStart?.(item.id) }} - onDragEnter={setHoverIndex} + onDragEnter={setHoverItemId} onDragEnd={() => { - setDragIndex(null) - setHoverIndex(null) + setDragItemId(null) + setHoverItemId(null) onDragEnd?.() }} - onDrop={(toIndex) => { - if (dragIndex !== null) onReorder?.(dragIndex, toIndex) - setDragIndex(null) - setHoverIndex(null) + onDrop={(targetId) => { + if (dragItemId !== null) onReorder?.(dragItemId, targetId) + setDragItemId(null) + setHoverItemId(null) onDragEnd?.() }} /> diff --git a/apps/desktop/src/renderer/hooks/use-composer-queue.test.tsx b/apps/desktop/src/renderer/hooks/use-composer-queue.test.tsx new file mode 100644 index 00000000..35845d06 --- /dev/null +++ b/apps/desktop/src/renderer/hooks/use-composer-queue.test.tsx @@ -0,0 +1,190 @@ +import { afterAll, afterEach, beforeEach, describe, expect, mock, test } from "bun:test" +import { Window } from "happy-dom" +import { createStore, Provider } from "jotai" +import { act } from "react" +import { createRoot, type Root } from "react-dom/client" +import { sessionQueueFamily } from "../atoms/queue" +import type { QueueWireEntry } from "../lib/queue-helpers" + +const dom = new Window({ url: "http://localhost" }) +const original = Object.getOwnPropertyDescriptors(globalThis) +Object.assign(globalThis, { + window: dom, + document: dom.document, + navigator: dom.navigator, + IS_REACT_ACT_ENVIRONMENT: true, +}) +const list = mock(async (_params: unknown) => ({ data: { entries: [] } })) +const update = mock(async (_params: unknown) => ({})) +const remove = mock(async (_params: unknown) => ({})) +const steer = mock(async (_params: unknown) => ({})) +mock.module("../services/connection-manager", () => ({ + getProjectClient: () => ({ session: { queue: { list, update, remove }, steer } }), +})) +const { useComposerQueue } = await import("./use-composer-queue") +let queue: ReturnType +let root: Root +let store: ReturnType +function Harness({ sessionId }: { sessionId: string }) { + queue = useComposerQueue(sessionId, "/repo") + return null +} +const entries: QueueWireEntry[] = ["first", "second", "third"].map((text, index) => ({ + queueItemId: `queue-${index + 1}`, + position: index + 1, + preview: text, + input: [{ type: "text", text }], +})) +async function mount(input = entries) { + store = createStore() + store.set(sessionQueueFamily("session-1"), input) + store.set(sessionQueueFamily("session-2"), input) + root = createRoot(dom.document.createElement("div") as unknown as HTMLElement) + await navigate("session-1") +} +async function navigate(sessionId: string) { + await act(async () => { + root.render() + }) +} +beforeEach(() => { + for (const fn of [list, update, remove, steer]) fn.mockClear() + update.mockImplementation(async () => ({})) + remove.mockImplementation(async () => ({})) + steer.mockImplementation(async () => ({})) +}) +afterEach(async () => { await act(async () => { root?.unmount() }) }) +afterAll(() => { + mock.restore() + for (const key of ["window", "document", "navigator", "IS_REACT_ACT_ENVIRONMENT"]) { + if (original[key]) Object.defineProperty(globalThis, key, original[key]) + else Reflect.deleteProperty(globalThis, key) + } + void dom.happyDOM.close() +}) + +describe("composer queue actions", () => { + test("reorders upward and downward using one-based Native positions", async () => { + await mount() + await act(async () => { await queue.reorderQueueItem("queue-1", "queue-2") }) + await act(async () => { await queue.reorderQueueItem("queue-1", "queue-3") }) + await act(async () => { await queue.reorderQueueItem("queue-3", "queue-1") }) + expect(update.mock.calls).toEqual([ + [{ sessionId: "session-1", queueItemId: "queue-1", position: 2 }], + [{ sessionId: "session-1", queueItemId: "queue-1", position: 3 }], + [{ sessionId: "session-1", queueItemId: "queue-3", position: 1 }], + ]) + }) + + test("steers by queue identity rather than the lossy text preview", async () => { + await mount([{ ...entries[0], input: [{ type: "image" }, { type: "skill" }] }]) + await act(async () => { await queue.steerQueueItem(queue.queueItems[0]) }) + expect([steer.mock.calls, remove.mock.calls]).toEqual([ + [[{ sessionId: "session-1", queueItemId: "queue-1" }]], + [[{ sessionId: "session-1", queueItemId: "queue-1" }]], + ]) + }) + + test("reorders the original message if an earlier entry drains during dragging", async () => { + await mount() + await act(async () => { store.set(sessionQueueFamily("session-1"), entries.slice(1)) }) + await act(async () => { await queue.reorderQueueItem("queue-2", "queue-3") }) + expect(update.mock.calls).toEqual([[{ + sessionId: "session-1", queueItemId: "queue-2", position: 2, + }]]) + }) + + test("does not reorder a replacement when the dragged message has already drained", async () => { + await mount() + await act(async () => { store.set(sessionQueueFamily("session-1"), entries.slice(1)) }) + await act(async () => { await queue.reorderQueueItem("queue-1", "queue-3") }) + expect(update.mock.calls).toEqual([]) + }) + + test("keeps non-text inputs in the queue when text-only editing is requested", async () => { + await mount([{ ...entries[0], input: [{ type: "skill" }] }]) + let result: string | null | undefined + await act(async () => { result = await queue.editQueueItem(queue.queueItems[0]) }) + expect([queue.queueItems[0].canEdit, result, remove.mock.calls]).toEqual([false, null, []]) + }) + + test("does not inject twice on repeated clicks or removal retry", async () => { + await mount() + let finishSteer!: () => void + steer.mockImplementationOnce(() => new Promise((resolve) => { + finishSteer = () => resolve({}) + })) + remove.mockImplementationOnce(async () => { throw new Error("Remove unavailable") }) + const item = queue.queueItems[0] + await act(async () => { + const first = queue.steerQueueItem(item) + await queue.steerQueueItem(item) + finishSteer() + await first + }) + expect([queue.queueItems[0].error, queue.queueItems[0].steered, queue.queueItems[0].canEdit]).toEqual([ + "Remove unavailable", true, false, + ]) + await act(async () => { await queue.steerQueueItem(queue.queueItems[0]) }) + expect([steer.mock.calls.length, remove.mock.calls.length, queue.queueItems[0].error]).toEqual([1, 2, undefined]) + }) + + test("retains the original queued message when steering fails", async () => { + await mount() + steer.mockImplementationOnce(async () => { throw new Error("Turn finished") }) + await act(async () => { await queue.steerQueueItem(queue.queueItems[0]) }) + expect([queue.queueItems[0].error, queue.queueItems[0].status, remove.mock.calls]).toEqual([ + "Turn finished", "queued", [], + ]) + }) + + test("retains removal-only retry state when navigating away and back", async () => { + await mount() + remove.mockImplementationOnce(async () => { throw new Error("Remove unavailable") }) + await act(async () => { await queue.steerQueueItem(queue.queueItems[0]) }) + await navigate("session-2") + expect([queue.queueItems[0].steered, queue.queueItems[0].error]).toEqual([false, undefined]) + await navigate("session-1") + expect([queue.queueItems[0].steered, queue.queueItems[0].error]).toEqual([true, "Remove unavailable"]) + await act(async () => { await queue.steerQueueItem(queue.queueItems[0]) }) + expect([steer.mock.calls.length, remove.mock.calls.length]).toEqual([1, 2]) + }) + + test("finishes pending actions in their original session after navigation", async () => { + await mount() + let finishSteer!: () => void + steer.mockImplementationOnce(() => new Promise((resolve) => { + finishSteer = () => resolve({}) + })) + remove.mockImplementationOnce(async () => { throw new Error("Remove unavailable") }) + let pending!: Promise + await act(async () => { pending = queue.steerQueueItem(queue.queueItems[0]) }) + await navigate("session-2") + expect(queue.queueItems[0].status).toBe("queued") + await act(async () => { finishSteer(); await pending }) + expect([queue.queueItems[0].steered, queue.queueItems[0].error]).toEqual([false, undefined]) + await navigate("session-1") + expect([queue.queueItems[0].steered, queue.queueItems[0].error]).toEqual([true, "Remove unavailable"]) + await act(async () => { await queue.steerQueueItem(queue.queueItems[0]) }) + expect([steer.mock.calls.length, remove.mock.calls.length]).toEqual([1, 2]) + }) + + test("reports removal failure without rejecting a fire-and-forget click or restoring a draft", async () => { + await mount() + remove.mockImplementation(async () => { throw new Error("Offline") }) + let removed: boolean | undefined + let text: string | null | undefined + await act(async () => { removed = await queue.removeQueueItem(queue.queueItems[0]) }) + await act(async () => { text = await queue.editQueueItem(queue.queueItems[0]) }) + expect([removed, text, queue.queueItems[0].error]).toEqual([false, null, "Offline"]) + }) + + test("reports failed reordering and refreshes authoritative order", async () => { + await mount() + update.mockImplementationOnce(async () => { throw new Error("Reorder unavailable") }) + await act(async () => { await queue.reorderQueueItem("queue-1", "queue-3") }) + expect([queue.queueItems[0].error, queue.queueItems[0].status, list.mock.calls.length]).toEqual([ + "Reorder unavailable", "queued", 2, + ]) + }) +}) diff --git a/apps/desktop/src/renderer/hooks/use-composer-queue.ts b/apps/desktop/src/renderer/hooks/use-composer-queue.ts index 6ab93500..61d8e80f 100644 --- a/apps/desktop/src/renderer/hooks/use-composer-queue.ts +++ b/apps/desktop/src/renderer/hooks/use-composer-queue.ts @@ -1,5 +1,5 @@ import { useAtomValue } from "jotai" -import { useCallback, useEffect, useMemo, useState } from "react" +import { useCallback, useEffect, useMemo, useRef, useState } from "react" import { sessionActiveTurnFamily, sessionQueueFamily } from "../atoms/queue" import { createLogger } from "../lib/logger" import { @@ -21,6 +21,7 @@ function wireEntryToComposerItem( entry: QueueWireEntry, action: QueueAction | null, error?: string, + steered?: boolean, ): ComposerQueueItem { return { id: entry.queueItemId, @@ -30,6 +31,8 @@ function wireEntryToComposerItem( fileCount: countQueueFileParts(entry) || undefined, createdAtMs: entry.enqueuedAt ? Date.parse(entry.enqueuedAt) : undefined, error, + canEdit: !steered && !entry.input?.some((part) => part.type !== "text"), + steered, } } @@ -39,6 +42,9 @@ export function useComposerQueue(sessionId: string, directory: string | null) { const [pendingActions, setPendingActions] = useState>(new Map()) const [actionErrors, setActionErrors] = useState>(new Map()) const [draggingId, setDraggingId] = useState(null) + const pendingIds = useRef(new Set()) + const steeredIds = useRef(new Set()) + const queueScope = JSON.stringify([directory, sessionId]) const refreshQueue = useCallback(async () => { if (!directory) return @@ -56,50 +62,58 @@ export function useComposerQueue(sessionId: string, directory: string | null) { }, [refreshQueue]) useEffect(() => { - setPendingActions(new Map()) - setActionErrors(new Map()) + setDraggingId(null) }, [sessionId]) const queueItems = useMemo(() => { return entries.map((entry) => wireEntryToComposerItem( entry, - pendingActions.get(entry.queueItemId) ?? null, - actionErrors.get(entry.queueItemId), + pendingActions.get(queueScope + entry.queueItemId) ?? null, + actionErrors.get(queueScope + entry.queueItemId), + steeredIds.current.has(queueScope + entry.queueItemId), ), ) - }, [actionErrors, entries, pendingActions]) + }, [actionErrors, entries, pendingActions, queueScope]) + // The composer is reused across sessions. Keep action state keyed by the + // original session so navigation cannot reset in-flight locks or retries. const setPending = useCallback((queueItemId: string, action: QueueAction | null) => { + const key = queueScope + queueItemId + if (action) pendingIds.current.add(key) + else pendingIds.current.delete(key) setPendingActions((current) => { const next = new Map(current) - if (action) next.set(queueItemId, action) - else next.delete(queueItemId) + if (action) next.set(key, action) + else next.delete(key) return next }) - }, []) + }, [queueScope]) const setItemError = useCallback((queueItemId: string, message: string | null) => { + const key = queueScope + queueItemId setActionErrors((current) => { const next = new Map(current) - if (message) next.set(queueItemId, message) - else next.delete(queueItemId) + if (message) next.set(key, message) + else next.delete(key) return next }) - }, []) + }, [queueScope]) const steerQueueItem = useCallback( async (item: ComposerQueueItem) => { + if (pendingIds.current.has(queueScope + item.id)) return if (!directory || !clientSupportsQueue(directory)) return const client = getProjectClient(directory) if (!client?.session?.steer || !client?.session?.queue?.remove) return setPending(item.id, "steering") setItemError(item.id, null) try { - await client.session.steer({ - sessionId: sessionId, - parts: [{ type: "text", text: item.text }], - }) + if (!steeredIds.current.has(queueScope + item.id)) { + await client.session.steer({ sessionId, queueItemId: item.id }) + // A retry after removal fails must not inject the same message twice. + steeredIds.current.add(queueScope + item.id) + } await client.session.queue.remove({ sessionId: sessionId, queueItemId: item.id, @@ -112,11 +126,12 @@ export function useComposerQueue(sessionId: string, directory: string | null) { setPending(item.id, null) } }, - [directory, sessionId, setItemError, setPending], + [directory, sessionId, queueScope, setItemError, setPending], ) const removeQueueItem = useCallback( async (item: ComposerQueueItem) => { + if (pendingIds.current.has(queueScope + item.id)) return if (!directory || !clientSupportsQueue(directory)) return const client = getProjectClient(directory) if (!client?.session?.queue?.remove) return @@ -127,46 +142,55 @@ export function useComposerQueue(sessionId: string, directory: string | null) { sessionId: sessionId, queueItemId: item.id, }) + return true } catch (err) { const message = err instanceof Error ? err.message : "Remove failed" setItemError(item.id, message) log.error("queue.remove failed", { sessionId, queueItemId: item.id }, err) - throw err + return false } finally { setPending(item.id, null) } }, - [directory, sessionId, setItemError, setPending], + [directory, sessionId, queueScope, setItemError, setPending], ) const editQueueItem = useCallback( async (item: ComposerQueueItem) => { - if ((item.fileCount ?? 0) > 0) return null - await removeQueueItem(item) - return item.text + if (pendingIds.current.has(queueScope + item.id) || steeredIds.current.has(queueScope + item.id)) return null + if (item.canEdit === false || (item.fileCount ?? 0) > 0) return null + const removed = await removeQueueItem(item) + return removed ? item.text : null }, - [removeQueueItem], + [queueScope, removeQueueItem], ) const reorderQueueItem = useCallback( - async (fromIndex: number, toIndex: number) => { - if (!directory || fromIndex === toIndex) return + async (queueItemId: string, targetQueueItemId: string) => { + if (!directory || queueItemId === targetQueueItemId) return const client = getProjectClient(directory) if (!client?.session?.queue?.update) return - const item = entries[fromIndex] - if (!item) return + const item = entries.find((entry) => entry.queueItemId === queueItemId) + const toIndex = entries.findIndex((entry) => entry.queueItemId === targetQueueItemId) + if (!item || toIndex < 0 || pendingIds.current.has(queueScope + item.queueItemId)) return + setPending(item.queueItemId, "submitting") + setItemError(item.queueItemId, null) try { await client.session.queue.update({ sessionId: sessionId, queueItemId: item.queueItemId, - position: toIndex, + // Native queue positions are one-based; drag indices are zero-based. + position: toIndex + 1, }) } catch (err) { - log.error("queue.reorder failed", { sessionId, fromIndex, toIndex }, err) + setItemError(item.queueItemId, err instanceof Error ? err.message : "Reorder failed") + log.error("queue.reorder failed", { sessionId, queueItemId, targetQueueItemId }, err) void refreshQueue() + } finally { + setPending(item.queueItemId, null) } }, - [directory, entries, refreshQueue, sessionId], + [directory, entries, refreshQueue, sessionId, queueScope, setItemError, setPending], ) return {