fix(desktop): preserve queued input and stable queue controls

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.

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<unknown> {
+		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() }
+	})
+})

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 }

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(<ComposerStatusStack queueItems={queueItems} onReorderQueueItem={onReorder} />) })
+	}
+	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() })
+	}
+})

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<ComposerQueueItem>) {
+	return renderToStaticMarkup(<ComposerStatusStack
+		queueItems={[{ id: "queue-1", text: "queued text", status: "queued", ...item }]}
+		onEditQueueItem={() => {}}
+		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(/<button[^>]*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<")
+	})
+})