diff --git a/CHANGELOG.md b/CHANGELOG.md index a6ad7865..2f120d8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ All notable changes to **omp-web** (`@kahme247/ompweb`) are documented in this f ### Fixes & Improvements +- Promote queued follow-ups through OMP's `promote_queued_message` command before relabeling them as steering; unsupported runtimes leave the queue unchanged. +- Cancel queued messages in OMP before removing their chips or recalling them for editing. Failed or timed-out cancellation leaves the message visible; successful recall survives composer remounts without losing newer typing. Requires native `remove_queued_message` support. +- Preserve question-dialog answers and selections when an SSE reconnect replays the same pending request or answer submission fails. - Keep composer controls on one line, with equally sized Send, Stop, and Queue buttons and model names truncating before short effort labels. - Align the + button and primary action with matching composer insets. - Clearly dim Attach files while the agent is running; queued messages remain text-only. diff --git a/README.ja.md b/README.ja.md index dafee4bd..36891497 100644 --- a/README.ja.md +++ b/README.ja.md @@ -29,6 +29,10 @@ - [omp](https://github.com/can1357/oh-my-pi) がインストールされ、`PATH` に含まれていること（または `OMP_WEB_OMP_BIN` で指定） - Node.js `>= 22.19.0` +キュー内のフォローアップに対する **Steer** 操作には、`promote_queued_message` RPC コマンドをサポートする omp ランタイムが必要です（omp 18.1.16 では未対応）。古いランタイムではエラーを表示し、メッセージはフォローアップとしてキューに残ります。ompweb が重複するステアリングメッセージを送信することはありません。 + +キューの **Delete** と **Edit** には、さらに `remove_queued_message` が必要です。OMP がキャンセルを確認してから表示を削除し、編集時はキャンセル成功後にテキストを入力欄へ戻します。未対応のランタイムや、すでにキューに存在しないメッセージでは、キューを変更せず通知を表示します。 + ## クイックスタート **インストールせずに直接実行:** @@ -58,7 +62,7 @@ ompweb --no-open # ブラウザ自動起動を無効 ## 主な機能 - **リアルタイムチャット**: ローカルの `omp` エージェントとストリーミング対話。 -- **キュー削除の確認**: キュー内のフォローアップやステアメッセージをパネルから削除する前に、内容を表示して確認します。OMP 内部ですでにキューに入ったメッセージの配信は取り消しません。 +- **キュー削除の確認**: OMP にキュー登録されたフォローアップやステアメッセージをキャンセルする前に、内容を表示して確認します。ネイティブの `remove_queued_message` 対応が必要です。配信済みのメッセージは取り消せません。 - **セッション管理**: プロジェクトごとに履歴を一覧表示、分岐やフォークにも対応。 - **下書きの復元**: 未送信のテキストを会話または新規セッションのワークスペースごとに保存し、ブラウザストレージが利用可能な場合は、同じタブでの「戻る」「進む」や再読み込み後に復元します（最大 50 件）。画像と添付ファイルはメモリ内にのみ保持されます。 - **ライブタスク＆サブエージェント**: Todo リストと稼働中サブエージェントの進捗を折りたたみパネルでリアルタイム表示。 diff --git a/README.md b/README.md index 429e233a..4ea81688 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,10 @@ A clean, modern web UI for the [oh-my-pi (omp)](https://github.com/can1357/oh-my - [omp](https://github.com/can1357/oh-my-pi) installed and available on your `PATH` (or specified via `OMP_WEB_OMP_BIN`) - Node.js `>= 22.19.0` +The **Steer** action on a queued follow-up requires an omp runtime with the `promote_queued_message` RPC command (not available in omp 18.1.16). Older runtimes report an error and leave the message queued as a follow-up; ompweb does not send a duplicate steering message. + +Queue **Delete** and **Edit** additionally require `remove_queued_message`. Deletion is confirmed by OMP before the chip disappears; editing recalls text only after cancellation succeeds. Unsupported runtimes or messages that are no longer pending leave the queue unchanged and display a notice. + ## Quick Start **Run directly without installing:** @@ -176,7 +180,7 @@ host (KDE Plasma, and most Wayland/X11 desktops). ## Features - **Interactive Chat**: Real-time streaming conversation with your local `omp` agent — tool calls, thinking levels, token counts, cost, context gauge, queue controls, and interrupt & retry. -- **Queue Deletion Confirmation**: Preview and confirm before removing queued follow-ups or steered messages from the queue panel. This does not cancel delivery already queued inside OMP. +- **Queue Deletion Confirmation**: Preview and confirm before cancelling queued follow-ups or steered messages in OMP. Requires native `remove_queued_message` support; already-delivered messages cannot be recalled. - **Session Management**: Browse past conversations by project, fork sessions, branch within a session, archive/restore, import session files, and deep-link via URL. - **Draft Recovery**: Unsent text stays scoped to its conversation or new-session workspace and is restored after Back/Forward navigation or reload in the same tab when browser storage is available (up to 50 drafts). Images and file attachments remain in memory only. - **Live Plans & Subagents**: Collapsible panels pinned above the composer track live todo phases and running subagents (status, tool, retries, tokens/cost, nested tasks) with transcript dialogs and history recovery. diff --git a/README.zh-CN.md b/README.zh-CN.md index f1867a46..6689593c 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -29,6 +29,10 @@ - 已安装 [omp](https://github.com/can1357/oh-my-pi) 且在 `PATH` 中（或通过 `OMP_WEB_OMP_BIN` 指定路径） - Node.js `>= 22.19.0` +将排队的后续消息转为 **Steer** 需要 omp 运行时支持 `promote_queued_message` RPC 命令（omp 18.1.16 尚不支持）。旧版运行时会返回错误，消息仍作为后续消息排队；ompweb 不会重复发送一条引导消息。 + +队列中的 **Delete** 和 **Edit** 还需要支持 `remove_queued_message`。只有 OMP 确认取消后才会移除消息；编辑时也必须先取消成功，再将文本恢复到输入框。运行时不支持该命令或消息已不在队列中时，界面保留队列并显示提示。 + ## 快速开始 **免安装直接运行：** @@ -58,7 +62,7 @@ ompweb --no-open # 不自动打开浏览器 ## 功能特性 - **实时对话**：与本地 `omp` 智能体进行低延迟流式交互。 -- **队列删除确认**：从队列面板移除后续消息或引导消息前，先预览内容并确认。此操作不会取消 OMP 内部已排队消息的发送。 +- **队列删除确认**：取消 OMP 中排队的后续消息或引导消息前，先预览内容并确认。需要原生支持 `remove_queued_message`；已发送的消息无法撤回。 - **会话管理**：按项目浏览历史会话，支持会话分叉与分支回溯。 - **草稿恢复**：未发送的文本按会话或新会话的工作区分别保存；浏览器存储可用时，在同一标签页后退、前进或重新加载后可恢复（最多 50 份草稿）。图片和文件附件仅保留在内存中。 - **实时任务与子智能体**：可折叠面板实时展示任务清单（todo）与子智能体进度，并支持查看完整转录。 diff --git a/components/ChatInput-banners.tsx b/components/ChatInput-banners.tsx index cd276d09..039a1b36 100644 --- a/components/ChatInput-banners.tsx +++ b/components/ChatInput-banners.tsx @@ -11,11 +11,13 @@ export function QueuedActionButton({ onClick, title, accent = false, + disabled = false, children, }: { onClick: () => void; title: string; accent?: boolean; + disabled?: boolean; children: React.ReactNode; }) { return ( @@ -23,6 +25,7 @@ export function QueuedActionButton({ type="button" onClick={onClick} title={title} + disabled={disabled} style={{ flexShrink: 0, padding: "4px 8px", minHeight: 24, @@ -30,7 +33,8 @@ export function QueuedActionButton({ borderRadius: 6, background: "transparent", color: accent ? "var(--accent)" : "var(--text-dim)", - cursor: "pointer", + cursor: disabled ? "wait" : "pointer", + opacity: disabled ? 0.5 : 1, fontSize: 11, fontWeight: accent ? 600 : 400, transition: "background var(--dur-fast) var(--ease-out-warm), color var(--dur-fast) var(--ease-out-warm)", diff --git a/components/ChatInput.navigation.test.mjs b/components/ChatInput.navigation.test.mjs index c9832493..9cac7a99 100644 --- a/components/ChatInput.navigation.test.mjs +++ b/components/ChatInput.navigation.test.mjs @@ -103,3 +103,114 @@ test("attachment-only drafts stay protected across live draft-key changes and re delete globalThis.document; } }); + +for (const navigation of ["draft-key change", "unmount"]) { + test(`queued Edit recovers the old draft after ${navigation} while cancellation is pending`, async () => { + installBrowser(); + const oldKey = `edit-old-${navigation}`; + const newKey = `edit-new-${navigation}`; + const ref = React.createRef(); + let release; + const cancellation = new Promise((resolve) => { release = resolve; }); + const queuedMessages = { steering: [], followUp: ["queued question"] }; + const render = (draftKey) => React.createElement(ChatInput, { + ref, draftKey, queuedMessages, isStreaming: true, + onSend() {}, onAbort() {}, onRemoveQueuedMessage: () => cancellation, + }); + let renderer; + const type = (value) => renderer.root.findByType("textarea").props.onChange({ target: { value, selectionStart: value.length } }); + try { + await act(() => { renderer = TestRenderer.create(render(oldKey)); }); + await act(() => type("original draft")); + await act(async () => { + ref.current.addFiles([new File(["keep this attachment"], "notes.txt", { type: "text/plain" })]); + }); + const originalFiles = getDraft(oldKey).files; + await act(() => renderer.root.findAllByType("button").find((button) => button.props.children === "Edit").props.onClick()); + await act(() => type("draft updated while waiting")); + assert.equal(getDraft(oldKey).value, "draft updated while waiting", "Edit must wait for cancellation acknowledgement"); + + if (navigation === "unmount") { + await act(() => renderer.unmount()); + await act(() => { renderer = TestRenderer.create(render(newKey)); }); + } else { + await act(() => renderer.update(render(newKey))); + } + await act(() => type("new session draft")); + await act(async () => { + release(true); + await cancellation; + }); + assert.equal(renderer.root.findByType("textarea").props.value, "new session draft"); + assert.equal(getDraft(newKey).value, "new session draft"); + assert.equal(getDraft(oldKey).value, "queued question\n\ndraft updated while waiting"); + assert.deepEqual(getDraft(oldKey).files, originalFiles); + + if (navigation === "unmount") { + await act(() => renderer.unmount()); + await act(() => { renderer = TestRenderer.create(render(oldKey)); }); + } else { + await act(() => renderer.update(render(oldKey))); + } + assert.equal(renderer.root.findByType("textarea").props.value, "queued question\n\ndraft updated while waiting"); + assert.equal(getDraft(newKey).value, "new session draft", "returning to the old chat preserves the new chat's draft"); + } finally { + await act(() => renderer?.unmount()); + clearDraft(oldKey); + clearDraft(newKey); + delete globalThis.window; + delete globalThis.document; + } + }); +} + +for (const navigation of ["original composer", "same-key remount"]) { + test(`queued Edit restores ${navigation} without losing batched typing or duplicating recalled text`, async () => { + installBrowser(); + const draftKey = `edit-recovery-${navigation}`; + const ref = React.createRef(); + let release; + const cancellation = new Promise((resolve) => { release = resolve; }); + const render = () => React.createElement(ChatInput, { + ref, draftKey, queuedMessages: { steering: [], followUp: ["queued question"] }, isStreaming: true, + onSend() {}, onAbort() {}, onRemoveQueuedMessage: () => cancellation, + }); + let renderer; + const textarea = () => renderer.root.findByType("textarea"); + const type = (value) => textarea().props.onChange({ target: { value, selectionStart: value.length } }); + try { + await act(() => { renderer = TestRenderer.create(render()); }); + await act(() => type("original draft")); + await act(async () => { + ref.current.addFiles([new File(["keep this attachment"], "notes.txt", { type: "text/plain" })]); + }); + const originalFiles = getDraft(draftKey).files; + await act(() => renderer.root.findAllByType("button").find((button) => button.props.children === "Edit").props.onClick()); + if (navigation === "same-key remount") { + await act(() => renderer.unmount()); + await act(() => { renderer = TestRenderer.create(render()); }); + } + await act(async () => { + type("typed while waiting"); + ref.current.insertText("and queued insertion"); + assert.equal(getDraft(draftKey).value, "original draft", "pending React updates have not been persisted yet"); + release(true); + await cancellation; + }); + const recovered = "queued question\n\ntyped while waiting and queued insertion"; + assert.equal(textarea().props.value, recovered); + assert.equal(getDraft(draftKey).value, recovered); + assert.deepEqual(getDraft(draftKey).files, originalFiles); + + await act(() => type(`${textarea().props.value}\nnext user edit`)); + assert.equal(textarea().props.value, `${recovered}\nnext user edit`); + assert.equal(getDraft(draftKey).value, `${recovered}\nnext user edit`); + assert.deepEqual(getDraft(draftKey).files, originalFiles); + } finally { + await act(() => renderer?.unmount()); + clearDraft(draftKey); + delete globalThis.window; + delete globalThis.document; + } + }); +} diff --git a/components/ChatInput.tsx b/components/ChatInput.tsx index 60c198c2..a807a844 100644 --- a/components/ChatInput.tsx +++ b/components/ChatInput.tsx @@ -11,7 +11,7 @@ import { useDictation } from "@/hooks/useDictation"; import type { GenerationSpeedInfo, SessionStatsInfo } from "@/lib/pi-types"; import { formatCompactNumber, formatPercent } from "@/lib/format"; import { ContextDetailPanel } from "./ComposerPanels"; -import { clearDraft, getDraft, setDraft } from "@/lib/draft-store"; +import { clearDraft, getDraft, recoverDraftText, setDraft, subscribeDraftRecovery } from "@/lib/draft-store"; import { expandWebSlashCommand } from "@/lib/web-slash-commands"; import type { AttachedImage, AttachedTextFile } from "./ChatInput-draft-attachments"; import { @@ -123,10 +123,10 @@ interface Props { modelCapacity?: { contextWindow?: number; maxTokens?: number } | null; /** Generation speed shown in the context ring popover. */ generationSpeed?: GenerationSpeedInfo | null; - /** Remove one queued message from the queue panel (Edit/Delete/Steer). */ - onRemoveQueuedMessage?: (text: string) => void; - /** Relabel the first queued follow-up as a steering message. */ - onPromoteQueuedToSteer?: (text: string) => void; + /** Cancel one pending message in omp before removing it from the queue panel. */ + onRemoveQueuedMessage?: (text: string, queue: keyof QueuedMessages) => Promise; + /** Promote the first matching native follow-up into steering. */ + onPromoteQueuedToSteer?: (text: string) => void | Promise; slashCommands?: SlashCommandInfo[]; slashCommandsLoading?: boolean; onLoadSlashCommands?: () => Promise | SlashCommandInfo[]; @@ -272,6 +272,7 @@ export const ChatInput = memo(forwardRef(function ChatIn const [value, setValue] = useState(() => (draftKey ? getDraft(draftKey)?.value ?? "" : "")); const [queuedDeleteTarget, setQueuedDeleteTarget] = useState<{ text: string; + kind: "follow-up" | "steer"; draftKey: string | undefined; queue: Props["queuedMessages"]; } | null>(null); @@ -630,6 +631,23 @@ export const ChatInput = memo(forwardRef(function ChatIn setAttachedTextFiles(draftFilesToAttachedFiles(draft?.files)); }, [draftKey]); + useLayoutEffect(() => subscribeDraftRecovery((key, text) => { + if (draftKeyRef.current !== key) return; + // Merge with pending edits rather than replacing them with a store snapshot. + setValue((current) => current ? `${text}\n\n${current}` : text); + setAtQuery(null); + setHistoryMenuOpen(false); + requestAnimationFrame(() => { + if (draftKeyRef.current !== key) return; + const ta = textareaRef.current; + if (!ta) return; + ta.focus(); + ta.setSelectionRange(text.length, text.length); + ta.style.height = "auto"; + ta.style.height = Math.min(ta.scrollHeight, 200) + "px"; + }); + }), []); + useEffect(() => { const ta = textareaRef.current; if (!ta) return; @@ -1019,9 +1037,7 @@ export const ChatInput = memo(forwardRef(function ChatIn && Boolean(onFollowUp); // ── Queued follow-up bar ──────────────────────────────────────────────── - // omp reports only a queued count over RPC; the texts are tracked in a - // client-side mirror, so Edit/Delete/Steer act on that mirror through the - // session hook's helpers. + // Texts are mirrored locally; mutations wait for native acknowledgement. const queuedEntries = [ ...(queuedMessages?.followUp ?? []).map((text) => ({ kind: "follow-up" as const, text })), ...(queuedMessages?.steering ?? []).map((text) => ({ kind: "steer" as const, text })), @@ -1030,49 +1046,59 @@ export const ChatInput = memo(forwardRef(function ChatIn const queuedCount = queuedEntries.length; const [queueExpanded, setQueueExpanded] = useState(false); + const [queueActionPending, setQueueActionPending] = useState(false); + const queueActionPendingRef = useRef(false); // Invalidate confirmation if delivery or navigation changes the queue. const activeDeleteTarget = queuedDeleteTarget?.draftKey === draftKey && queuedDeleteTarget?.queue === queuedMessages ? queuedDeleteTarget : null; - const handleItemEdit = useCallback((text: string) => { - onRemoveQueuedMessage?.(text); - setValue(text); - setAtQuery(null); - setHistoryMenuOpen(false); - requestAnimationFrame(() => { - const ta = textareaRef.current; - if (!ta) return; - ta.focus(); - ta.setSelectionRange(text.length, text.length); - ta.style.height = "auto"; - ta.style.height = Math.min(ta.scrollHeight, 200) + "px"; - }); - }, [onRemoveQueuedMessage]); + const handleQueueAction = useCallback(async ( + entry: { kind: "follow-up" | "steer"; text: string }, + action: "edit" | "delete" | "steer", + ) => { + if (queueActionPendingRef.current) return; + queueActionPendingRef.current = true; + setQueueActionPending(true); + const key = draftKeyRef.current; + try { + if (action === "steer") { + if (entry.kind === "follow-up") await onPromoteQueuedToSteer?.(entry.text); + return; + } + const removed = await onRemoveQueuedMessage?.( + entry.text, entry.kind === "steer" ? "steering" : "followUp", + ); + setQueuedDeleteTarget(null); + if (!removed || action !== "edit") return; + // Recover through the store even if a new composer now owns this key. + recoverDraftText(key, entry.text); + } catch (error) { + setQueuedDeleteTarget(null); + toast.error(error instanceof Error ? error.message : String(error)); + } finally { + queueActionPendingRef.current = false; + setQueueActionPending(false); + } + }, [onRemoveQueuedMessage, onPromoteQueuedToSteer]); - const handleItemDelete = useCallback((text: string) => { - setQueuedDeleteTarget({ text, draftKey, queue: queuedMessages }); + const handleItemDelete = useCallback((entry: { kind: "follow-up" | "steer"; text: string }) => { + setQueuedDeleteTarget({ ...entry, draftKey, queue: queuedMessages }); }, [draftKey, queuedMessages]); - const handleItemSteer = useCallback((entry: { kind: "follow-up" | "steer"; text: string }) => { - if (entry.kind === "follow-up") { - onPromoteQueuedToSteer?.(entry.text); - } - }, [onPromoteQueuedToSteer]); - const handleQueuedEdit = useCallback(() => { if (!firstQueued) return; - handleItemEdit(firstQueued.text); - }, [firstQueued, handleItemEdit]); + void handleQueueAction(firstQueued, "edit"); + }, [firstQueued, handleQueueAction]); const handleQueuedDelete = useCallback(() => { if (!firstQueued) return; - handleItemDelete(firstQueued.text); + handleItemDelete(firstQueued); }, [firstQueued, handleItemDelete]); const handleQueuedSteer = useCallback(() => { if (!firstQueued) return; - handleItemSteer(firstQueued); - }, [firstQueued, handleItemSteer]); + void handleQueueAction(firstQueued, "steer"); + }, [firstQueued, handleQueueAction]); const getNextSlashIndex = useCallback((direction: "up" | "down" | "left" | "right") => { const lastIndex = filteredSlashCommands.length - 1; @@ -1473,10 +1499,10 @@ export const ChatInput = memo(forwardRef(function ChatIn confirmLabel={t("chatInput.queuedDelete")} cancelLabel={t("chatInput.cancel")} danger + busy={queueActionPending} onConfirm={() => { if (!activeDeleteTarget) return; - setQueuedDeleteTarget(null); - onRemoveQueuedMessage?.(activeDeleteTarget.text); + void handleQueueAction(activeDeleteTarget, "delete"); }} /> {/* Hidden file input */} @@ -2042,14 +2068,14 @@ export const ChatInput = memo(forwardRef(function ChatIn > {firstQueued?.text} - + {t("chatInput.queuedEdit")} - + {t("chatInput.queuedDelete")} {firstQueued?.kind === "follow-up" && ( - + {t("chatInput.queuedSteerAction")} )} @@ -2184,14 +2210,14 @@ export const ChatInput = memo(forwardRef(function ChatIn > {entry.text} - handleItemEdit(entry.text)} title={t("chatInput.queuedEditTitle")}> + void handleQueueAction(entry, "edit")} title={t("chatInput.queuedEditTitle")}> {t("chatInput.queuedEdit")} - handleItemDelete(entry.text)} title={t("chatInput.queuedDeleteTitle")}> + handleItemDelete(entry)} title={t("chatInput.queuedDeleteTitle")}> {t("chatInput.queuedDelete")} {entry.kind === "follow-up" && ( - handleItemSteer(entry)} title={t("chatInput.queuedSteerTitle")} accent> + void handleQueueAction(entry, "steer")} title={t("chatInput.queuedSteerTitle")} accent> {t("chatInput.queuedSteerAction")} )} diff --git a/components/ExtensionDialog.test.mjs b/components/ExtensionDialog.test.mjs new file mode 100644 index 00000000..632d0cf8 --- /dev/null +++ b/components/ExtensionDialog.test.mjs @@ -0,0 +1,79 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import React, { act } from "react"; +import TestRenderer from "react-test-renderer"; +import { createJiti } from "jiti"; + +const jiti = createJiti(import.meta.url, { jsx: { runtime: "automatic" }, tsconfigPaths: true }); +const { ExtensionDialog } = await jiti.import("./ExtensionDialog.tsx"); +globalThis.IS_REACT_ACT_ENVIRONMENT = true; + +test("Ask editor preserves an answer on request replay and initializes the next request", async () => { + const request = { + type: "extension_ui_request", + id: "answer-1", + method: "editor", + title: "How should this work?", + }; + const responses = []; + const onRespond = (current, response) => responses.push({ id: current.id, ...response }); + const render = (current) => React.createElement(ExtensionDialog, { request: current, onRespond, attached: true }); + const answer = "Keep my answer\nincluding this second line."; + let renderer; + try { + await act(() => { renderer = TestRenderer.create(render(request)); }); + await act(() => renderer.root.findByType("textarea").props.onChange({ target: { value: answer } })); + + // Reconnecting SSE reparses the pending request into a fresh object. + await act(() => renderer.update(render(JSON.parse(JSON.stringify(request))))); + assert.equal(renderer.root.findByType("textarea").props.value, answer); + await act(() => renderer.root.findByType("textarea").props.onKeyDown({ key: "Enter", ctrlKey: true })); + assert.deepEqual(responses, [{ id: "answer-1", value: answer }]); + + await act(() => renderer.update(render({ ...request, id: "answer-2" }))); + assert.equal(renderer.root.findByType("textarea").props.value, ""); + + const nextRequest = { ...request, id: "answer-3", prefill: "Suggested answer" }; + await act(() => renderer.update(render(nextRequest))); + assert.equal(renderer.root.findByType("textarea").props.value, "Suggested answer"); + await act(() => renderer.root.findByType("textarea").props.onChange({ target: { value: "Edited suggestion" } })); + await act(() => renderer.update(render(JSON.parse(JSON.stringify(nextRequest))))); + assert.equal(renderer.root.findByType("textarea").props.value, "Edited suggestion"); + + await act(() => renderer.update(render({ ...request, id: "answer-4" }))); + assert.equal(renderer.root.findByType("textarea").props.value, ""); + } finally { + await act(() => renderer?.unmount()); + } +}); + +test("question selection survives request replay but is cleared for the next question", async () => { + const request = { + type: "extension_ui_request", + id: "choice-1", + method: "select", + title: "Which approach?", + options: ["First", "Second"], + }; + const responses = []; + const onRespond = (current, response) => responses.push({ id: current.id, ...response }); + const render = (current) => React.createElement(ExtensionDialog, { request: current, onRespond, attached: true }); + let renderer; + const choice = () => renderer.root.findAllByType("button").find((button) => button.props.children === "Second"); + const next = () => renderer.root.findAllByType("button").find((button) => button.props.disabled !== undefined); + try { + await act(() => { renderer = TestRenderer.create(render(request)); }); + await act(() => choice().props.onClick()); + await act(() => renderer.update(render(JSON.parse(JSON.stringify(request))))); + assert.equal(choice().props["aria-pressed"], true); + assert.equal(next().props.disabled, false); + await act(() => next().props.onClick()); + assert.deepEqual(responses, [{ id: "choice-1", value: "Second" }]); + + await act(() => renderer.update(render({ ...request, id: "choice-2" }))); + assert.equal(choice().props["aria-pressed"], false); + assert.equal(next().props.disabled, true); + } finally { + await act(() => renderer?.unmount()); + } +}); diff --git a/components/ExtensionDialog.tsx b/components/ExtensionDialog.tsx index 18f32c47..88a90504 100644 --- a/components/ExtensionDialog.tsx +++ b/components/ExtensionDialog.tsx @@ -1,6 +1,6 @@ "use client"; -import { useEffect, useState } from "react"; +import { useEffect, useRef, useState } from "react"; import type { ExtensionUiRequest } from "@/lib/types"; import { useI18n } from "@/lib/i18n"; import { useModalDialog } from "@/hooks/useModalDialog"; @@ -38,8 +38,12 @@ export function ExtensionDialog({ const { t } = useI18n(); const [value, setValue] = useState(request.method === "editor" ? request.prefill ?? "" : ""); const [selectedOption, setSelectedOption] = useState(null); + const requestIdRef = useRef(request.id); useEffect(() => { + // SSE reconnects replay the same request as a fresh object, not a new question. + if (requestIdRef.current === request.id) return; + requestIdRef.current = request.id; setValue(request.method === "editor" ? request.prefill ?? "" : ""); setSelectedOption(null); }, [request]); diff --git a/hooks/useAgentSession-queue.ts b/hooks/useAgentSession-queue.ts index cc40142b..e098ac2a 100644 --- a/hooks/useAgentSession-queue.ts +++ b/hooks/useAgentSession-queue.ts @@ -1,5 +1,4 @@ -// Queued-prompt tracking and sessionStorage persistence helpers -// extracted from useAgentSession (pure logic only — no hook state). +// Queued-prompt tracking, persistence, and same-document queue notifications. export interface QueuedMessages { steering: string[]; @@ -8,6 +7,10 @@ export interface QueuedMessages { export const EMPTY_QUEUE: QueuedMessages = { steering: [], followUp: [] }; +// In-flight RPCs outlive their initiating hook. Remounted delivery handlers must +// mark the same occurrence consumed before its acknowledgement can relabel it. +export const pendingQueuedPromotions = new Map>(); + // omp reports only queuedMessageCount over RPC; the queued texts live in React // state and would vanish on reload. Mirror them into sessionStorage (per // session, best-effort, size-bounded) so a reload can restore the queue panel. @@ -66,3 +69,23 @@ export function clearPersistedQueue(sessionId: string | null): void { // ignore storage errors } } + +type QueueListener = (sessionId: string, queue: QueuedMessages) => void; +const queueListeners = new Set(); + +/** Publish the already-applied snapshot, not a removal to repeat per listener. */ +export function publishQueueChange(sessionId: string, queue: QueuedMessages): void { + persistQueue(sessionId, queue); + for (const listener of [...queueListeners]) { + try { + listener(sessionId, queue); + } catch { + // A failing subscriber must not stop the others. + } + } +} + +export function subscribeQueueChanges(listener: QueueListener): () => void { + queueListeners.add(listener); + return () => { queueListeners.delete(listener); }; +} diff --git a/hooks/useAgentSession.rpc.test.mjs b/hooks/useAgentSession.rpc.test.mjs index 5aa0287c..e47b5deb 100644 --- a/hooks/useAgentSession.rpc.test.mjs +++ b/hooks/useAgentSession.rpc.test.mjs @@ -25,10 +25,11 @@ globalThis.localStorage = { removeItem: (k) => kvStore.delete(k), clear: () => kvStore.clear(), }; +const queueStore = new Map(); globalThis.sessionStorage = { - getItem: () => null, - setItem: () => {}, - removeItem: () => {}, + getItem: (key) => queueStore.get(key) ?? null, + setItem: (key, value) => queueStore.set(key, String(value)), + removeItem: (key) => queueStore.delete(key), }; // Listeners are capturable so tests can fire visibilitychange/online. function makeEventTarget() { @@ -129,7 +130,7 @@ function jsonResponse(status, value) { const world = { esInstances: [], calls: [], - holds: [], // { match(method, url), produce: () => Promise<{ status, value }> } + holds: [], // { match(method, url, body), produce: () => Promise<{ status, value }> } sessions: new Map(), // sid -> { leafId, messages, entryIds } agents: new Map(), // sid -> { running, state } subagentSnapshots: new Map(), // sid -> SubagentSnapshotLike[] @@ -138,10 +139,11 @@ const world = { async function fetchStub(url, init = {}) { const method = (init.method ?? "GET").toUpperCase(); const u = String(url); - world.calls.push({ method, url: u, body: typeof init.body === "string" ? safeParse(init.body) : null }); + const body = typeof init.body === "string" ? safeParse(init.body) : null; + world.calls.push({ method, url: u, body }); for (let i = 0; i < world.holds.length; i++) { - if (world.holds[i].match(method, u)) { + if (world.holds[i].match(method, u, body)) { const h = world.holds.splice(i, 1)[0]; const { status = 200, value } = await h.produce(); return jsonResponse(status, value); @@ -174,7 +176,6 @@ async function fetchStub(url, init = {}) { return jsonResponse(200, { running: a.running, state: a.state }); } if (method === "POST") { - const body = typeof init.body === "string" ? safeParse(init.body) : null; if (body?.type === "get_subagents") { return jsonResponse(200, { success: true, data: { subagents: world.subagentSnapshots.get(sid) ?? [] } }); } @@ -221,7 +222,7 @@ function sessionInfo(sid) { }; } -async function mountSession(sid, onAgentEnd) { +async function mountSession(sid, onAgentEnd, strictMode = false) { let latest = null; function Chat({ session }) { latest = useAgentSession({ session, newSessionCwd: null, ...(onAgentEnd ? { onAgentEnd } : {}) }); @@ -229,7 +230,8 @@ async function mountSession(sid, onAgentEnd) { } let renderer; await act(async () => { - renderer = TestRenderer.create(React.createElement(Chat, { session: sessionInfo(sid) })); + const chat = React.createElement(Chat, { session: sessionInfo(sid) }); + renderer = TestRenderer.create(strictMode ? React.createElement(React.StrictMode, null, chat) : chat); }); await settle(); // hydration: loadSession + /state + models + subagents activeRenderers.add(renderer); @@ -269,6 +271,7 @@ function resetWorld() { world.sessions.clear(); world.agents.clear(); world.subagentSnapshots.clear(); + queueStore.clear(); } function primeSession(sid, messages) { @@ -290,8 +293,8 @@ const assistantMsg = (id, text) => ({ }); /** Mount + hydrate, then send a prompt and open the stream. Returns the ES. */ -async function startRun(t, sid, message) { - const w = await mountSession(sid); +async function startRun(t, sid, message, strictMode = false) { + const w = await mountSession(sid, undefined, strictMode); if (t) t.after(unmountAll); assert.equal(w.latest.loading, false, "hydration must complete"); assert.equal(w.latest.agentRunning, false); @@ -316,6 +319,526 @@ async function startRun(t, sid, message) { // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- +for (const queue of ["steering", "followUp"]) { + test(`queued cancellation waits for acknowledgement and removes one ${queue} duplicate`, async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("cancellation", [userMsg("u0", "q")]); + const w = await mountSession("cancellation", undefined, true); + await act(async () => { + await w.latest.handleSteer("target"); + await w.latest.handleFollowUp("target"); + await w.latest.handleFollowUp("target"); + }); + const before = { steering: ["target"], followUp: ["target", "target"] }; + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "remove_queued_message", + produce: () => acknowledgement, + }); + let cancellation; + await act(async () => { cancellation = w.latest.removeQueuedMessage("target", queue); }); + assert.deepEqual(w.latest.queuedMessages, before); + const sent = world.calls.filter((call) => call.body?.type === "remove_queued_message"); + assert.deepEqual(sent.map((call) => call.body), [{ type: "remove_queued_message", message: "target", queue }]); + await act(async () => { + release({ value: { success: true, data: { removed: true } } }); + assert.equal(await cancellation, true); + }); + assert.deepEqual(w.latest.queuedMessages, { + ...before, [queue]: before[queue].slice(1), + }); + }); +} + +test("queued cancellation preserves pending messages on refusal and unsupported runtimes", async (t) => { + for (const response of [ + { value: { success: true, data: { removed: false } } }, + { status: 400, value: { error: "Unknown RPC command: remove_queued_message" } }, + ]) { + await t.test(response.status ? "unsupported" : "not pending", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("cancel-failure", [userMsg("u0", "q")]); + const w = await mountSession("cancel-failure"); + await act(async () => { await w.latest.handleFollowUp("target"); }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "remove_queued_message", + produce: async () => response, + }); + await act(async () => { assert.equal(await w.latest.removeQueuedMessage("target", "followUp"), false); }); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.equal(w.latest.notices.length, 1); + }); + } +}); + +test("successful cancellation after unmount still reports success for draft recovery", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("cancel-old", [userMsg("u0", "q")]); + primeSession("cancel-new", [userMsg("u1", "other")]); + const old = await mountSession("cancel-old"); + await act(async () => { await old.latest.handleFollowUp("target"); }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "remove_queued_message", + produce: () => acknowledgement, + }); + let cancellation; + await act(async () => { cancellation = old.latest.removeQueuedMessage("target", "followUp"); }); + await act(async () => { old.renderer.unmount(); }); + activeRenderers.delete(old.renderer); + const current = await mountSession("cancel-new"); + await act(async () => { await current.latest.handleFollowUp("target"); }); + await act(async () => { + release({ value: { success: true, data: { removed: true } } }); + assert.equal(await cancellation, true, "the caller must recover text already removed by native"); + }); + assert.deepEqual(current.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.deepEqual(current.latest.notices, []); +}); + +for (const remaining of ["other", "target"]) { + test(`cancellation acknowledgement updates a remounted session without removing its ${remaining} remainder`, async (t) => { + t.after(unmountAll); + resetWorld(); + const sid = `cancel-remount-${remaining}`; + primeSession(sid, [userMsg("u0", "q")]); + primeSession("cancel-unrelated", [userMsg("u1", "other session")]); + const old = await mountSession(sid); + await act(async () => { + await old.latest.handleFollowUp("target"); + await old.latest.handleFollowUp(remaining); + }); + world.agents.set(sid, { running: true, state: { queuedMessageCount: 2, isStreaming: true } }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, url, body) => method === "POST" && url === `/api/agent/${sid}` && body?.type === "remove_queued_message", + produce: () => acknowledgement, + }); + let cancellation; + await act(async () => { cancellation = old.latest.removeQueuedMessage("target", "followUp"); }); + await act(() => old.renderer.unmount()); + activeRenderers.delete(old.renderer); + + const current = await mountSession(sid, undefined, true); + await act(() => lastEs().open()); + const observer = await mountSession(sid); + await act(() => lastEs().open()); + const unrelated = await mountSession("cancel-unrelated"); + await act(async () => { await unrelated.latest.handleFollowUp("target"); }); + assert.deepEqual(current.latest.queuedMessages, { steering: [], followUp: ["target", remaining] }); + assert.deepEqual(observer.latest.queuedMessages, current.latest.queuedMessages); + await act(async () => { + // Native still has one pending item, so count-only reconciliation cannot + // discover which text was cancelled. + world.agents.set(sid, { running: true, state: { queuedMessageCount: 1, isStreaming: true } }); + release({ value: { success: true, data: { removed: true } } }); + assert.equal(await cancellation, true); + }); + const after = { steering: [], followUp: [remaining] }; + assert.deepEqual(current.latest.queuedMessages, after); + assert.deepEqual(observer.latest.queuedMessages, after, "subscribers adopt one removal, not one each"); + assert.deepEqual(JSON.parse(queueStore.get(`omp-queue-${sid}`)), after); + assert.deepEqual(unrelated.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.deepEqual(JSON.parse(queueStore.get("omp-queue-cancel-unrelated")), unrelated.latest.queuedMessages); + }); +} + +test("an answered dialog handoff cannot clear the next unanswered request", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("dialog-handoff", [userMsg("u0", "q")]); + const { w, es } = await startRun(null, "dialog-handoff", "ask me"); + const first = { type: "extension_ui_request", id: "first", method: "editor", title: "First question" }; + const next = { ...first, id: "next", title: "Next question" }; + await act(() => es.emit(first)); + await settle(300); + assert.equal(w.latest.extensionDialog?.id, first.id, "unanswered requests have no clear timer"); + + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "extension_ui_response", + produce: () => acknowledgement, + }); + let response; + await act(async () => { response = w.latest.respondToExtensionUi(first, { value: "My answer" }); }); + await act(() => es.emit(next)); + await act(async () => { + release({ value: { success: true, data: {} } }); + await response; + }); + await settle(300); + assert.equal(w.latest.extensionDialog?.id, next.id, "a delayed response timer only clears its own request"); +}); + +for (const failure of ["HTTP rejection", "connection failure"]) { + test(`a failed question response preserves the pending question for retry: ${failure}`, async (t) => { + t.after(unmountAll); + resetWorld(); + const sid = `answer-failure-${failure}`; + primeSession(sid, [userMsg("u0", "q")]); + const { w, es } = await startRun(null, sid, "ask"); + const request = { type: "extension_ui_request", id: "retry-answer", method: "editor", title: "Answer" }; + await act(() => es.emit(request)); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "extension_ui_response", + produce: async () => { + if (failure === "connection failure") throw new Error("connection lost"); + return { status: 500, value: { error: "answer delivery failed" } }; + }, + }); + await act(async () => { await w.latest.respondToExtensionUi(request, { value: "my retained answer" }); }); + await settle(300); + assert.equal(w.latest.extensionDialog?.id, request.id, "failure must not unmount and erase the answer"); + assert.equal(w.latest.notices.at(-1)?.type, "error"); + await act(() => es.emit({ ...request })); + assert.equal(w.latest.extensionDialog?.id, request.id); + await act(async () => { await w.latest.respondToExtensionUi(request, { value: "my retained answer" }); }); + await settle(300); + assert.equal(w.latest.extensionDialog, null, "a successful retry closes the answered question"); + }); +} + +test("a stale local queue action reports that its target is unavailable", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("missing-local-target", [userMsg("u0", "q")]); + const w = await mountSession("missing-local-target"); + await act(async () => { + assert.equal(await w.latest.removeQueuedMessage("already delivered", "followUp"), false); + }); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: [] }); + assert.equal(w.latest.notices.at(-1)?.type, "warning"); +}); + + +test("queued promotion waits for native acknowledgement and its promoted occurrence can be cancelled", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("promotion", [userMsg("u0", "q")]); + const w = await mountSession("promotion", undefined, true); + await act(async () => { + await w.latest.handleSteer("existing steer"); + await w.latest.handleFollowUp("other"); + await w.latest.handleFollowUp("target"); + await w.latest.handleFollowUp("target"); + }); + const before = { steering: ["existing steer"], followUp: ["other", "target", "target"] }; + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, url, body) => method === "POST" && url === "/api/agent/promotion" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + const callsBefore = world.calls.length; + let promotion; + await act(async () => { promotion = w.latest.promoteQueuedToSteer("target"); }); + assert.deepEqual(w.latest.queuedMessages, before, "pending RPC must not relabel the chip"); + assert.deepEqual(world.calls.slice(callsBefore), [{ + method: "POST", url: "/api/agent/promotion", body: { type: "promote_queued_message", message: "target" }, + }], "promotion must not enqueue a separate steer or follow-up"); + await act(async () => { + release({ value: { success: true, data: { promoted: true } } }); + await promotion; + }); + assert.deepEqual(w.latest.queuedMessages, { + steering: ["existing steer", "target"], followUp: ["other", "target"], + }); + assert.deepEqual(w.latest.notices, []); + + let releaseCancellation; + const cancellationAcknowledgement = new Promise((resolve) => { releaseCancellation = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "remove_queued_message", + produce: () => cancellationAcknowledgement, + }); + const callsBeforeCancellation = world.calls.length; + let cancellation; + await act(async () => { cancellation = w.latest.removeQueuedMessage("target", "steering"); }); + assert.deepEqual(w.latest.queuedMessages, { + steering: ["existing steer", "target"], followUp: ["other", "target"], + }); + assert.deepEqual(world.calls.slice(callsBeforeCancellation), [{ + method: "POST", url: "/api/agent/promotion", + body: { type: "remove_queued_message", message: "target", queue: "steering" }, + }]); + await act(async () => { + releaseCancellation({ value: { success: true, data: { removed: true } } }); + assert.equal(await cancellation, true); + }); + assert.deepEqual(w.latest.queuedMessages, { + steering: ["existing steer"], followUp: ["other", "target"], + }, "Steer then Delete must leave the same-text follow-up queued"); +}); + +test("promotion acknowledgement updates remounted observers and Delete cancels steering only", async (t) => { + t.after(unmountAll); + resetWorld(); + const sid = "promotion-remount"; + primeSession(sid, [userMsg("u0", "q")]); + primeSession("promotion-unrelated", [userMsg("u1", "other session")]); + const old = await mountSession(sid); + await act(async () => { + await old.latest.handleFollowUp("target"); + await old.latest.handleFollowUp("target"); + }); + world.agents.set(sid, { running: true, state: { queuedMessageCount: 2, isStreaming: true } }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, url, body) => method === "POST" && url === `/api/agent/${sid}` && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let promotion; + await act(async () => { promotion = old.latest.promoteQueuedToSteer("target"); }); + await act(() => old.renderer.unmount()); + activeRenderers.delete(old.renderer); + + const current = await mountSession(sid, undefined, true); + await act(() => lastEs().open()); + const observer = await mountSession(sid); + await act(() => lastEs().open()); + const unrelated = await mountSession("promotion-unrelated"); + await act(async () => { await unrelated.latest.handleFollowUp("target"); }); + assert.deepEqual(current.latest.queuedMessages, { steering: [], followUp: ["target", "target"] }); + await act(async () => { + release({ value: { success: true, data: { promoted: true } } }); + await promotion; + }); + const promoted = { steering: ["target"], followUp: ["target"] }; + assert.deepEqual(current.latest.queuedMessages, promoted); + assert.deepEqual(observer.latest.queuedMessages, promoted, "observers adopt one promotion, not one each"); + assert.deepEqual(JSON.parse(queueStore.get(`omp-queue-${sid}`)), promoted); + assert.deepEqual(unrelated.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.deepEqual(JSON.parse(queueStore.get("omp-queue-promotion-unrelated")), unrelated.latest.queuedMessages); + assert.deepEqual(unrelated.latest.notices, []); + + world.holds.push({ + match: (method, url, body) => method === "POST" && url === `/api/agent/${sid}` && body?.type === "remove_queued_message" && body?.queue === "steering", + produce: async () => ({ value: { success: true, data: { removed: true } } }), + }); + await act(async () => { + assert.equal(await current.latest.removeQueuedMessage("target", "steering"), true); + }); + assert.deepEqual(world.calls.filter((call) => call.body?.type === "remove_queued_message").map((call) => call.body), [ + { type: "remove_queued_message", message: "target", queue: "steering" }, + ]); + const remaining = { steering: [], followUp: ["target"] }; + assert.deepEqual(current.latest.queuedMessages, remaining); + assert.deepEqual(observer.latest.queuedMessages, remaining); + assert.deepEqual(JSON.parse(queueStore.get(`omp-queue-${sid}`)), remaining); +}); + +test("remounted delivery before promotion acknowledgement preserves the next duplicate and pending guard", async (t) => { + t.after(unmountAll); + resetWorld(); + const sid = "promotion-remount-delivery"; + primeSession(sid, [userMsg("u0", "q")]); + const old = await mountSession(sid); + await act(async () => { + await old.latest.handleFollowUp("target"); + await old.latest.handleFollowUp("target"); + }); + world.agents.set(sid, { running: true, state: { queuedMessageCount: 2, isStreaming: true } }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let promotion; + await act(async () => { promotion = old.latest.promoteQueuedToSteer("target"); }); + await act(() => old.renderer.unmount()); + activeRenderers.delete(old.renderer); + const current = await mountSession(sid, undefined, true); + const es = lastEs(); + await act(() => es.open()); + await act(async () => { + await current.latest.promoteQueuedToSteer("target"); + assert.equal(await current.latest.removeQueuedMessage("target", "followUp"), false); + es.emit({ type: "message_end", message: userMsg("delivered", "target") }); + release({ value: { success: true, data: { promoted: true } } }); + await promotion; + }); + const remaining = { steering: [], followUp: ["target"] }; + assert.deepEqual(current.latest.queuedMessages, remaining); + assert.deepEqual(JSON.parse(queueStore.get(`omp-queue-${sid}`)), remaining); + assert.equal(current.latest.messages.filter((message) => message.role === "user" && message.content === "target").length, 1); + assert.equal(world.calls.filter((call) => call.body?.type === "promote_queued_message").length, 1); + assert.equal(world.calls.some((call) => call.body?.type === "remove_queued_message"), false); + assert.deepEqual(current.latest.notices, []); +}); + +test("queued promotion preserves the follow-up and reports native refusal or rejection", async (t) => { + for (const outcome of [ + { name: "not-found", response: { value: { success: true, data: { promoted: false } } }, noticeType: "warning" }, + { name: "unsupported", response: { status: 400, value: { error: "Unknown RPC command: promote_queued_message" } }, noticeType: "error" }, + ]) { + await t.test(outcome.name, async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession(outcome.name, [userMsg("u0", "q")]); + const w = await mountSession(outcome.name); + await act(async () => { await w.latest.handleFollowUp("target"); }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let promotion; + await act(async () => { promotion = w.latest.promoteQueuedToSteer("target"); }); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: ["target"] }); + await act(async () => { + release(outcome.response); + await promotion; + }); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.equal(w.latest.notices.length, 1); + assert.equal(w.latest.notices[0].type, outcome.noticeType); + if (outcome.response.value.error) { + assert.equal(w.latest.notices[0].message, outcome.response.value.error); + } + + // A settled failure releases the overlap guard so a deliberate retry works. + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: async () => ({ value: { success: true, data: { promoted: true } } }), + }); + await act(async () => { await w.latest.promoteQueuedToSteer("target"); }); + assert.deepEqual(w.latest.queuedMessages, { steering: ["target"], followUp: [] }); + }); + } +}); + +test("overlapping same-text promotion clicks send one command and promote one occurrence", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("double-promotion", [userMsg("u0", "q")]); + const w = await mountSession("double-promotion"); + await act(async () => { + await w.latest.handleFollowUp("target"); + await w.latest.handleFollowUp("target"); + }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let first; + let second; + await act(async () => { + first = w.latest.promoteQueuedToSteer("target"); + second = w.latest.promoteQueuedToSteer("target"); + }); + assert.equal(world.calls.filter((call) => call.body?.type === "promote_queued_message").length, 1); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: ["target", "target"] }); + await act(async () => { + release({ value: { success: true, data: { promoted: true } } }); + await Promise.all([first, second]); + }); + assert.deepEqual(w.latest.queuedMessages, { steering: ["target"], followUp: ["target"] }); +}); + +test("delivery before promotion acknowledgement does not relabel the next duplicate or resurrect the delivered chip", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("delivered-promotion", [userMsg("u0", "q")]); + const { w, es } = await startRun(t, "delivered-promotion", "run", true); + await act(async () => { + es.emit({ type: "agent_start" }); + await w.latest.handleFollowUp("target"); + await w.latest.handleFollowUp("target"); + }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let promotion; + await act(async () => { promotion = w.latest.promoteQueuedToSteer("target"); }); + await act(async () => { es.emit({ type: "message_end", message: userMsg("delivered", "target") }); }); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.equal(w.latest.messages.filter((message) => message.role === "user" && message.content === "target").length, 1); + await act(async () => { + release({ value: { success: true, data: { promoted: true } } }); + await promotion; + }); + assert.deepEqual(w.latest.queuedMessages, { steering: [], followUp: ["target"] }); +}); + +test("cancellation cannot race a pending promotion and remove the next duplicate", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("removed-promotion", [userMsg("u0", "q")]); + const w = await mountSession("removed-promotion", undefined, true); + await act(async () => { + await w.latest.handleFollowUp("target"); + await w.latest.handleFollowUp("target"); + }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let promotion; + await act(async () => { promotion = w.latest.promoteQueuedToSteer("target"); }); + await act(async () => { + assert.equal(await w.latest.removeQueuedMessage("target", "followUp"), false); + release({ value: { success: true, data: { promoted: true } } }); + await promotion; + }); + assert.deepEqual(w.latest.queuedMessages, { steering: ["target"], followUp: ["target"] }); + assert.equal(world.calls.some((call) => call.body?.type === "remove_queued_message"), false); + assert.deepEqual(w.latest.notices, []); +}); + +test("promotion acknowledgements after navigation cannot change the newly mounted session", async (t) => { + for (const response of [ + { value: { success: true, data: { promoted: true } } }, + { status: 400, value: { error: "Native promotion failed" } }, + ]) { + await t.test(response.status ? "error" : "success", async (t) => { + t.after(unmountAll); + resetWorld(); + primeSession("old-promotion", [userMsg("u0", "old")]); + primeSession("new-promotion", [userMsg("u1", "new")]); + const old = await mountSession("old-promotion"); + await act(async () => { await old.latest.handleFollowUp("target"); }); + let release; + const acknowledgement = new Promise((resolve) => { release = resolve; }); + world.holds.push({ + match: (method, _url, body) => method === "POST" && body?.type === "promote_queued_message", + produce: () => acknowledgement, + }); + let promotion; + await act(async () => { promotion = old.latest.promoteQueuedToSteer("target"); }); + await act(async () => { old.renderer.unmount(); }); + activeRenderers.delete(old.renderer); + const current = await mountSession("new-promotion"); + await act(async () => { await current.latest.handleFollowUp("target"); }); + await act(async () => { + release(response); + await promotion; + }); + assert.deepEqual(current.latest.queuedMessages, { steering: [], followUp: ["target"] }); + assert.deepEqual(current.latest.notices, []); + assert.equal(world.calls.some((call) => call.url === "/api/agent/new-promotion" && call.body?.type === "promote_queued_message"), false); + assert.deepEqual(JSON.parse(queueStore.get("omp-queue-old-promotion")), response.status + ? { steering: [], followUp: ["target"] } + : { steering: ["target"], followUp: [] }, "an unobserved session still records the native result"); + }); + } +}); test("full run over fake SSE: optimistic bubble, coalesced streaming, terminal reload", async (t) => { t.after(unmountAll); diff --git a/hooks/useAgentSession.ts b/hooks/useAgentSession.ts index f34fc3c2..a65f1bc6 100644 --- a/hooks/useAgentSession.ts +++ b/hooks/useAgentSession.ts @@ -47,8 +47,11 @@ import { EMPTY_QUEUE, clearPersistedQueue, isEmptyQueue, + pendingQueuedPromotions, persistQueue, + publishQueueChange, readPersistedQueue, + subscribeQueueChanges, } from "./useAgentSession-queue"; import type { QueuedMessages } from "./useAgentSession-queue"; import { @@ -304,7 +307,16 @@ export function useAgentSession(opts: UseAgentSessionOptions) { const [extensionCustomUi, setExtensionCustomUi] = useState(null); const [extensionStatuses, setExtensionStatuses] = useState([]); const [extensionWidgets, setExtensionWidgets] = useState([]); - const [queuedMessages, setQueuedMessages] = useState({ steering: [], followUp: [] }); + const [queuedMessages, setQueuedMessagesState] = useState(EMPTY_QUEUE); + const queuedMessagesRef = useRef(EMPTY_QUEUE); + // Serialize queue transitions in event/effect handlers, not React's + // replayable render-phase updaters. Promotion bookkeeping and its queue + // removal must observe the same transition, even before React commits. + const updateQueuedMessages = useCallback((update: QueuedMessages | ((prev: QueuedMessages) => QueuedMessages)) => { + const next = typeof update === "function" ? update(queuedMessagesRef.current) : update; + queuedMessagesRef.current = next; + setQueuedMessagesState(next); + }, []); const [subagents, setSubagents] = useState([]); const [subagentEvents, setSubagentEvents] = useState>({}); const [subagentTranscriptVersions, setSubagentTranscriptVersions] = useState>({}); @@ -334,6 +346,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { // get_state snapshots may lag behind the RPC round-trip, so a snapshot // reporting queuedMessageCount === 0 must not wipe a queue we just wrote. const queueMutatedAtRef = useRef(0); + const queuedRemovalRef = useRef<{ sessionId: string } | null>(null); const agentRunningRef = useRef(false); const bashRunningRef = useRef(false); const bashRecoveryIdRef = useRef(0); @@ -386,6 +399,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { // True once this mount has persisted a non-empty queue: gates removal so a // just-mounted empty state cannot wipe a stored queue before restore runs. const queuePersistDirtyRef = useRef(false); + const publishedQueueRef = useRef(null); const eventCoalescerRef = useRef(null); if (eventCoalescerRef.current === null) { eventCoalescerRef.current = createMessageUpdateCoalescer((event) => { @@ -755,9 +769,9 @@ export function useAgentSession(opts: UseAgentSessionOptions) { if (liveState.extensionStatuses !== undefined) setExtensionStatuses(liveState.extensionStatuses ?? []); if (liveState.extensionWidgets !== undefined) setExtensionWidgets(liveState.extensionWidgets ?? []); if (liveState.todoPhases !== undefined) setTodoPhases(liveState.todoPhases ?? []); - if (liveState.queuedMessageCount === 0 && Date.now() - queueMutatedAtRef.current >= 5000) setQueuedMessages(EMPTY_QUEUE); + if (liveState.queuedMessageCount === 0 && Date.now() - queueMutatedAtRef.current >= 5000) updateQueuedMessages(EMPTY_QUEUE); } else if (!agentState.running && Date.now() - queueMutatedAtRef.current >= 5000) { - setQueuedMessages(EMPTY_QUEUE); + updateQueuedMessages(EMPTY_QUEUE); } if (showLoading) setLoading(false); return agentState; @@ -782,7 +796,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { // Ensure the flag is cleared even if the pre-state early-return path was taken if (showLoading && includeState && !messagesLoaded) initialHydrationPendingRef.current = false; } - }, [refreshSubagentHistory, applyAuthoritativeModel, beginAuthoritativeModelSync]); + }, [refreshSubagentHistory, applyAuthoritativeModel, beginAuthoritativeModelSync, updateQueuedMessages]); const loadContext = useCallback(async (sid: string, leafId: string | null, includePreCompaction = false): Promise => { const seq = ++contextRequestSeqRef.current; @@ -995,35 +1009,6 @@ export function useAgentSession(opts: UseAgentSessionOptions) { }); }, [eventCoalescer]); - const respondToExtensionUi = useCallback(async ( - request: ExtensionUiDialogRequest, - response: { value: string } | { confirmed: boolean } | { cancelled: true }, - ) => { - const sid = sessionIdRef.current; - if (!sid) { - setExtensionDialog((current) => current?.id === request.id ? null : current); - return; - } - try { - await sendAgentCommand(sid, { - type: "extension_ui_response", - id: request.id, - ...response, - }); - } catch (e) { - console.error("Failed to send extension UI response:", e); - } finally { - // OMP commonly emits the next Ask select immediately after this response. - // Keep the current panel mounted for a short hand-off window so the composer - // never flashes empty between sequential questions. - if (extensionDialogClearTimerRef.current) clearTimeout(extensionDialogClearTimerRef.current); - extensionDialogClearTimerRef.current = setTimeout(() => { - setExtensionDialog((current) => current?.id === request.id ? null : current); - extensionDialogClearTimerRef.current = null; - }, 250); - } - }, []); - // --------------------------------------------------------------------- // Host-tool bridge: omp-web registers tools the AGENT can call. The server // emits host_tool_call frames; this UI executes them and answers with @@ -1238,6 +1223,36 @@ export function useAgentSession(opts: UseAgentSessionOptions) { }); }, []); + const respondToExtensionUi = useCallback(async ( + request: ExtensionUiDialogRequest, + response: { value: string } | { confirmed: boolean } | { cancelled: true }, + ) => { + const sid = sessionIdRef.current; + if (!sid) { + setExtensionDialog((current) => current?.id === request.id ? null : current); + return; + } + try { + await sendAgentCommand(sid, { + type: "extension_ui_response", + id: request.id, + ...response, + }); + if (!hookAliveRef.current || sessionIdRef.current !== sid) return; + // Keep the answered panel mounted briefly while the next question arrives. + clearTimeout(extensionDialogClearTimerRef.current ?? undefined); + extensionDialogClearTimerRef.current = setTimeout(() => { + setExtensionDialog((current) => current?.id === request.id ? null : current); + extensionDialogClearTimerRef.current = null; + }, 250); + } catch (e) { + console.error("Failed to send extension UI response:", e); + if (hookAliveRef.current && sessionIdRef.current === sid) { + addNotice({ type: "error", message: e instanceof Error ? e.message : String(e) }); + } + } + }, [addNotice]); + const dismissNotice = useCallback((id: string) => { dispatchNotice({ type: "remove", id }); }, []); @@ -1488,7 +1503,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { if (state?.todoPhases !== undefined) setTodoPhases(state.todoPhases ?? []); // And the only reliable re-sync for a missed subagent lifecycle frame. void refreshSubagentRoster(sid); - if ((!state || state.queuedMessageCount === 0) && Date.now() - queueMutatedAtRef.current >= 5000) setQueuedMessages(EMPTY_QUEUE); + if ((!state || state.queuedMessageCount === 0) && Date.now() - queueMutatedAtRef.current >= 5000) updateQueuedMessages(EMPTY_QUEUE); const busy = data.running && state && (state.isStreaming || state.isPromptRunning || state.isCompacting); if (busy || !agentRunningRef.current) return; @@ -1512,7 +1527,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { void reconcileAgentState(sid); } } - }, [finishPromptWithoutStream, refreshSubagentRoster]); + }, [finishPromptWithoutStream, refreshSubagentRoster, updateQueuedMessages]); // Recovery net for missed SSE events: while the agent is running, verify // against the server periodically and whenever the tab returns to the @@ -1585,45 +1600,114 @@ export function useAgentSession(opts: UseAgentSessionOptions) { const consumeQueuedMessage = useCallback((text: string) => { if (!text) return; - setQueuedMessages((prev) => { + const sid = sessionIdRef.current; + const promotion = sid ? pendingQueuedPromotions.get(sid)?.get(text) : undefined; + updateQueuedMessages((prev) => { const si = prev.steering.indexOf(text); if (si !== -1) return { ...prev, steering: prev.steering.filter((_, i) => i !== si) }; const fi = prev.followUp.indexOf(text); - if (fi !== -1) return { ...prev, followUp: prev.followUp.filter((_, i) => i !== fi) }; + if (fi !== -1) { + // A pending acknowledgement must not promote the next same-text item. + if (promotion) promotion.consumed = true; + return { ...prev, followUp: prev.followUp.filter((_, i) => i !== fi) }; + } return prev; }); - }, []); + }, [updateQueuedMessages]); - /** Remove one queued message from the client-side queue mirror. omp's RPC - * protocol has no queue-mutation commands, so this only affects the queue - * panel: a message removed here may still be delivered by the running agent - * (it then arrives in the chat like any delivered turn). */ - const removeQueuedMessage = useCallback((text: string) => { - if (!text) return; - setQueuedMessages((prev) => { - const si = prev.steering.indexOf(text); - const fi = prev.followUp.indexOf(text); - if (si === -1 && fi === -1) return prev; - return { - steering: si === -1 ? prev.steering : prev.steering.filter((_, i) => i !== si), - followUp: fi === -1 ? prev.followUp : prev.followUp.filter((_, i) => i !== fi), + /** Only remove a chip after omp confirms cancellation of its queued payload. */ + const removeQueuedMessage = useCallback(async (text: string, queue: keyof QueuedMessages): Promise => { + const sid = sessionIdRef.current; + if (!hookAliveRef.current || !sid || !text) return false; + if (!queuedMessagesRef.current[queue].includes(text)) { + addNotice({ type: "warning", message: translate("agentSession.queuedRemovalUnavailable") }); + return false; + } + // Do not race another removal or promotion against the same queue mirror. + if (queuedRemovalRef.current?.sessionId === sid || pendingQueuedPromotions.get(sid)?.has(text)) return false; + const removal = { sessionId: sid }; + queuedRemovalRef.current = removal; + try { + const result = await sendAgentCommand<{ removed: boolean }>(sid, { + type: "remove_queued_message", message: text, queue, + }); + if (result?.removed !== true) { + if (hookAliveRef.current && sessionIdRef.current === sid) { + addNotice({ type: "warning", message: translate("agentSession.queuedRemovalUnavailable") }); + } + return false; + } + const removeFromMirror = (prev: QueuedMessages): QueuedMessages => { + const index = prev[queue].indexOf(text); + return index < 0 ? prev : { ...prev, [queue]: prev[queue].filter((_, i) => i !== index) }; }; - }); - }, []); + const mirror = hookAliveRef.current && sessionIdRef.current === sid + ? queuedMessagesRef.current + : readPersistedQueue(sid); + if (mirror) publishQueueChange(sid, removeFromMirror(mirror)); + return true; + } catch (error) { + if (hookAliveRef.current && sessionIdRef.current === sid) { + addNotice({ type: "error", message: error instanceof Error ? error.message : String(error) }); + } + return false; + } finally { + if (queuedRemovalRef.current === removal) queuedRemovalRef.current = null; + } + }, [addNotice]); - /** Promote the first queued follow-up to a steering message (client-side - * relabel; the delivery order itself is owned by omp). */ - const promoteQueuedToSteer = useCallback((text: string) => { - if (!text) return; - setQueuedMessages((prev) => { - const fi = prev.followUp.indexOf(text); - if (fi === -1) return prev; - return { - steering: [...prev.steering, text], - followUp: prev.followUp.filter((_, i) => i !== fi), - }; - }); - }, []); + /** Move the first matching native follow-up into steering, then relabel its + * still-undelivered chip. Never enqueue a second copy via steer. */ + const promoteQueuedToSteer = useCallback(async (text: string) => { + const sid = sessionIdRef.current; + if (!hookAliveRef.current || !sid || !text || !queuedMessagesRef.current.followUp.includes(text)) return; + // Text is the existing chip identity. Suppress overlap, not later retries. + let queuedPromotions = pendingQueuedPromotions.get(sid); + if (queuedPromotions?.has(text) || queuedRemovalRef.current?.sessionId === sid) return; + if (!queuedPromotions) { + queuedPromotions = new Map(); + pendingQueuedPromotions.set(sid, queuedPromotions); + } + const promotion = { consumed: false }; + queuedPromotions.set(text, promotion); + try { + const result = await sendAgentCommand<{ promoted: boolean }>(sid, { + type: "promote_queued_message", + message: text, + }); + if (result?.promoted !== true) { + if (hookAliveRef.current && sessionIdRef.current === sid) { + addNotice({ type: "warning", message: translate("agentSession.queuedPromotionUnavailable") }); + } + return; + } + if (promotion.consumed) return; + const mirror = hookAliveRef.current && sessionIdRef.current === sid + ? queuedMessagesRef.current + : readPersistedQueue(sid); + if (!mirror) return; + const fi = mirror.followUp.indexOf(text); + if (fi === -1) return; + publishQueueChange(sid, { + steering: [...mirror.steering, text], + followUp: mirror.followUp.filter((_, i) => i !== fi), + }); + } catch (error) { + if (!hookAliveRef.current || sessionIdRef.current !== sid) return; + addNotice({ type: "error", message: error instanceof Error ? error.message : String(error) }); + } finally { + if (queuedPromotions.get(text) === promotion) queuedPromotions.delete(text); + if (queuedPromotions.size === 0) pendingQueuedPromotions.delete(sid); + } + }, [addNotice]); + + useEffect(() => subscribeQueueChanges((sid, queue) => { + if (!hookAliveRef.current || sessionIdRef.current !== sid) return; + queueMutatedAtRef.current = Date.now(); + publishedQueueRef.current = queue; + queuePersistDirtyRef.current = !isEmptyQueue(queue); + updateQueuedMessages(queue); + }), [updateQueuedMessages]); // Mirror queued texts into sessionStorage so a reload can restore them. // The dirty gate keeps the initial empty state from wiping a stored queue @@ -1631,6 +1715,8 @@ export function useAgentSession(opts: UseAgentSessionOptions) { useEffect(() => { const sid = sessionIdRef.current; if (!sid) return; + // A same-document notification already persisted this exact snapshot. + if (queuedMessages === publishedQueueRef.current) return; const empty = isEmptyQueue(queuedMessages); if (empty && !queuePersistDirtyRef.current) return; queuePersistDirtyRef.current = !empty; @@ -1738,7 +1824,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { if (d.state?.todoPhases !== undefined) setTodoPhases(d.state.todoPhases ?? []); // omp reports only a queued count; an empty (or dead) session // means the client-tracked queue texts are stale. - if ((!d.state || d.state.queuedMessageCount === 0) && Date.now() - queueMutatedAtRef.current >= 5000) setQueuedMessages(EMPTY_QUEUE); + if ((!d.state || d.state.queuedMessageCount === 0) && Date.now() - queueMutatedAtRef.current >= 5000) updateQueuedMessages(EMPTY_QUEUE); }) .catch(() => {}); } @@ -2200,7 +2286,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { handleExtensionUiRequest(event as unknown as IncomingExtensionUiRequest); break; } - }, [addNotice, clearLiveToolResults, clearTerminalReconcileTimer, consumeQueuedMessage, finishPromptWithoutStream, handleExtensionUiRequest, handleHostToolCall, handleHostUriRequest, loadSession, mergeSubagents, onAgentEnd, reconcileAgentState, resetSubagentActivityState, applyAuthoritativeModel, beginAuthoritativeModelSync, setLiveToolResult, surfaceQuotaOnStream]); + }, [addNotice, clearLiveToolResults, clearTerminalReconcileTimer, consumeQueuedMessage, finishPromptWithoutStream, handleExtensionUiRequest, handleHostToolCall, handleHostUriRequest, loadSession, mergeSubagents, onAgentEnd, reconcileAgentState, resetSubagentActivityState, applyAuthoritativeModel, beginAuthoritativeModelSync, setLiveToolResult, surfaceQuotaOnStream, updateQueuedMessages]); handleAgentEventRef.current = handleAgentEvent; const handleSend = useCallback(async (message: string, images?: AttachedImage[]): Promise => { @@ -2860,13 +2946,13 @@ export function useAgentSession(opts: UseAgentSessionOptions) { // omp emits no queue snapshots; track the queued text locally until it // is delivered (user message_end) or the queue count drops to zero. queueMutatedAtRef.current = Date.now(); - setQueuedMessages((prev) => ({ ...prev, steering: [...prev.steering, message] })); + updateQueuedMessages((prev) => ({ ...prev, steering: [...prev.steering, message] })); } catch (e) { console.error("Failed to steer:", e); addNotice({ type: "error", message: e instanceof Error ? e.message : String(e) }); opts.chatInputRef?.current?.insertIfEmpty(message); } - }, [addNotice, opts.chatInputRef]); + }, [addNotice, opts.chatInputRef, updateQueuedMessages]); const handlePromptWithStreamingBehavior = useCallback(async ( message: string, @@ -2884,7 +2970,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { ...(piImages?.length ? { images: piImages } : {}), }); queueMutatedAtRef.current = Date.now(); - setQueuedMessages((prev) => behavior === "steer" + updateQueuedMessages((prev) => behavior === "steer" ? { ...prev, steering: [...prev.steering, message] } : { ...prev, followUp: [...prev.followUp, message] }); } catch (e) { @@ -2892,7 +2978,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { addNotice({ type: "error", message: e instanceof Error ? e.message : String(e) }); opts.chatInputRef?.current?.insertIfEmpty(message); } - }, [addNotice, opts.chatInputRef]); + }, [addNotice, opts.chatInputRef, updateQueuedMessages]); const handleFollowUp = useCallback(async (message: string, images?: AttachedImage[]) => { const sid = sessionIdRef.current; @@ -2905,13 +2991,13 @@ export function useAgentSession(opts: UseAgentSessionOptions) { ...(piImages?.length ? { images: piImages } : {}), }); queueMutatedAtRef.current = Date.now(); - setQueuedMessages((prev) => ({ ...prev, followUp: [...prev.followUp, message] })); + updateQueuedMessages((prev) => ({ ...prev, followUp: [...prev.followUp, message] })); } catch (e) { console.error("Failed to follow up:", e); addNotice({ type: "error", message: e instanceof Error ? e.message : String(e) }); opts.chatInputRef?.current?.insertIfEmpty(message); } - }, [addNotice, opts.chatInputRef]); + }, [addNotice, opts.chatInputRef, updateQueuedMessages]); const handleAbortCompaction = useCallback(async () => { const sid = sessionIdRef.current; @@ -3044,7 +3130,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { if (agentState.state.extensionStatuses !== undefined) setExtensionStatuses(agentState.state.extensionStatuses ?? []); if (agentState.state.extensionWidgets !== undefined) setExtensionWidgets(agentState.state.extensionWidgets ?? []); if (agentState.state.queuedMessageCount === 0 && Date.now() - queueMutatedAtRef.current >= 5000) { - setQueuedMessages(EMPTY_QUEUE); + updateQueuedMessages(EMPTY_QUEUE); // The queue drained while the page was closed — a stored copy // from a previous page load is stale. clearPersistedQueue(session.id); @@ -3053,7 +3139,7 @@ export function useAgentSession(opts: UseAgentSessionOptions) { // texts persisted by the previous page load. const persisted = readPersistedQueue(session.id); if (persisted) { - setQueuedMessages((prev) => (isEmptyQueue(prev) ? persisted : prev)); + updateQueuedMessages((prev) => (isEmptyQueue(prev) ? persisted : prev)); } } } @@ -3171,9 +3257,12 @@ export function useAgentSession(opts: UseAgentSessionOptions) { } }, [messages, streamState, agentRunning, agentPhase, extensionWidgets, isCompacting, retryInfo, activeSubagentCount, todoPhases, scrollToBottom, loading]); - useEffect(() => () => { - hookAliveRef.current = false; - if (followScrollFrameRef.current !== null) cancelAnimationFrame(followScrollFrameRef.current); + useEffect(() => { + hookAliveRef.current = true; + return () => { + hookAliveRef.current = false; + if (followScrollFrameRef.current !== null) cancelAnimationFrame(followScrollFrameRef.current); + }; }, []); // Load model list diff --git a/lib/draft-store.ts b/lib/draft-store.ts index 10c2510e..8ba20b7f 100644 --- a/lib/draft-store.ts +++ b/lib/draft-store.ts @@ -20,6 +20,7 @@ export interface ChatDraft { declare global { var __ompChatDrafts: Map | undefined; var __ompChatDraftListeners: Set<() => void> | undefined; + var __ompChatDraftRecoveryListeners: Set<(key: string, text: string) => void> | undefined; } const MAX_DRAFTS = 50; @@ -51,6 +52,7 @@ function readStoredDrafts(): Map { } const listeners = (globalThis.__ompChatDraftListeners ??= new Set<() => void>()); +const recoveryListeners = (globalThis.__ompChatDraftRecoveryListeners ??= new Set<(key: string, text: string) => void>()); export function hasUnsentDrafts(): boolean { return drafts.size > 0; @@ -61,6 +63,11 @@ export function subscribeDrafts(listener: () => void): () => void { return () => { listeners.delete(listener); }; } +export function subscribeDraftRecovery(listener: (key: string, text: string) => void): () => void { + recoveryListeners.add(listener); + return () => { recoveryListeners.delete(listener); }; +} + function cloneDraft(draft: ChatDraft): ChatDraft { return { value: draft.value, @@ -78,6 +85,16 @@ export function getDraft(key: string): ChatDraft | null { return draft ? cloneDraft(draft) : null; } +export function recoverDraftText(key: string, text: string): void { + if (key) { + const draft = getDraft(key) ?? { value: "", images: [], files: [] }; + setDraft(key, { ...draft, value: draft.value ? `${text}\n\n${draft.value}` : text }); + } + // Publish the recovery intent separately: ordinary persistence must not + // reapply it, and the mounted composer may have React updates still queued. + for (const listener of recoveryListeners) listener(key, text); +} + export function getDraftSummary(key: string): { text: string; hasAttachments: boolean } { const draft = drafts.get(key); if (!draft) return { text: "", hasAttachments: false }; diff --git a/lib/i18n/locales/en.json b/lib/i18n/locales/en.json index 61e53e95..7afd2e7a 100644 --- a/lib/i18n/locales/en.json +++ b/lib/i18n/locales/en.json @@ -21,6 +21,8 @@ "agentSession.sendFailed": "Failed to send message: {detail}", "agentSession.sessionRenamed": "Session renamed to {name}", "agentSession.shellSessionFailed": "Unable to create a session for the shell command", + "agentSession.queuedPromotionUnavailable": "The queued follow-up is no longer available to steer.", + "agentSession.queuedRemovalUnavailable": "This message is no longer pending in that queue and could not be cancelled.", "agentSession.startingAgent": "Starting omp… the first launch of a session can take a while.", "agentSession.toolPresetNotice": "Tool presets apply when a session starts; this session keeps its current tools.", "archiveBrowser.title": "Archived Sessions", @@ -269,7 +271,7 @@ "chatInput.fastLabel": "Fast", "chatInput.queuedDelete": "Delete", "chatInput.queuedDeleteTitle": "Remove this queued message", - "chatInput.queuedDeleteConfirmBody": "Remove this message from the queue panel? This cannot be undone here. The agent may still deliver a message it has already queued.", + "chatInput.queuedDeleteConfirmBody": "Cancel this queued message? Once cancellation is confirmed, it will be removed from the agent's queue. A message already delivered to the agent cannot be cancelled.", "chatInput.cancel": "Cancel", "chatInput.queuedEdit": "Edit", "chatInput.queuedEditTitle": "Edit this queued message", diff --git a/lib/i18n/locales/ja.json b/lib/i18n/locales/ja.json index 5fc84494..93077c1d 100644 --- a/lib/i18n/locales/ja.json +++ b/lib/i18n/locales/ja.json @@ -21,6 +21,8 @@ "agentSession.sendFailed": "メッセージの送信に失敗しました: {detail}", "agentSession.sessionRenamed": "セッション名を {name} に変更しました", "agentSession.shellSessionFailed": "シェルコマンド用のセッションを作成できませんでした", + "agentSession.queuedPromotionUnavailable": "このフォローアップはキューに存在しないため、ステアリングに変更できません。", + "agentSession.queuedRemovalUnavailable": "このメッセージはそのキューに存在しないため、キャンセルできませんでした。", "agentSession.startingAgent": "omp を起動しています… セッションの初回起動には時間がかかる場合があります。", "agentSession.toolPresetNotice": "ツールプリセットはセッション開始時に適用されます。このセッションは現在のツールを維持します。", "archiveBrowser.title": "アーカイブされたセッション", @@ -256,7 +258,7 @@ "chatInput.fastLabel": "高速", "chatInput.queuedDelete": "削除", "chatInput.queuedDeleteTitle": "このキューされたメッセージを削除", - "chatInput.queuedDeleteConfirmBody": "このメッセージをキューパネルから削除しますか？ここでの削除は取り消せません。エージェントがすでにキューに保持しているメッセージは、引き続き配信される場合があります。", + "chatInput.queuedDeleteConfirmBody": "このキューされたメッセージをキャンセルしますか？キャンセルが確認されると、エージェントのキューから削除されます。すでにエージェントに配信されたメッセージはキャンセルできません。", "chatInput.cancel": "キャンセル", "chatInput.queuedEdit": "編集", "chatInput.queuedEditTitle": "このキューされたメッセージを編集", diff --git a/lib/i18n/locales/zh-CN.json b/lib/i18n/locales/zh-CN.json index 1bf44517..019d3161 100644 --- a/lib/i18n/locales/zh-CN.json +++ b/lib/i18n/locales/zh-CN.json @@ -21,6 +21,8 @@ "agentSession.sendFailed": "发送消息失败：{detail}", "agentSession.sessionRenamed": "会话已重命名为 {name}", "agentSession.shellSessionFailed": "无法为 shell 命令创建会话", + "agentSession.queuedPromotionUnavailable": "该后续消息已不在队列中，无法转为引导消息。", + "agentSession.queuedRemovalUnavailable": "该消息已不在此队列中，无法取消。", "agentSession.startingAgent": "正在启动 omp……会话首次启动可能需要一些时间。", "agentSession.toolPresetNotice": "工具预设在会话启动时生效；当前会话将保留现有工具。", "archiveBrowser.title": "已归档会话", @@ -256,7 +258,7 @@ "chatInput.fastLabel": "快速", "chatInput.queuedDelete": "删除", "chatInput.queuedDeleteTitle": "移除这条已排队消息", - "chatInput.queuedDeleteConfirmBody": "要从队列面板移除这条消息吗？此处的删除无法撤销。智能体仍可能发送其已加入队列的消息。", + "chatInput.queuedDeleteConfirmBody": "要取消这条已排队消息吗？取消成功后，它将从智能体的队列中移除。已发送给智能体的消息无法取消。", "chatInput.cancel": "取消", "chatInput.queuedEdit": "编辑", "chatInput.queuedEditTitle": "编辑这条已排队消息", diff --git a/lib/omp/rpc-process-runtime.test.mjs b/lib/omp/rpc-process-runtime.test.mjs index 76d9e5dd..03d5936e 100644 --- a/lib/omp/rpc-process-runtime.test.mjs +++ b/lib/omp/rpc-process-runtime.test.mjs @@ -7,6 +7,7 @@ import { createJiti } from "jiti"; const jiti = createJiti(import.meta.url); const { RpcProcess, RpcCommandTimeoutError } = jiti("./rpc-process.ts"); const { encodeRpcFrames } = jiti("./rpc-frame.ts"); +const { AgentSessionWrapper } = jiti("../rpc-manager.ts"); function makeTransport() { const stdin = new PassThrough(); @@ -158,3 +159,55 @@ test("RpcProcess times out commands with RpcCommandTimeoutError and ignores late assert.ok(frameEvents.some((f) => f.type === "response" && f.command === "slow_command")); await proc.dispose(0); }); + +for (const type of ["remove_queued_message", "promote_queued_message"]) { + test(`${type} times out without replaying, misrouting late ACKs, or interrupting long commands`, async (t) => { + const transport = makeTransport(); + const proc = startProcess(transport); + await proc.waitReady(); + t.mock.timers.enable({ apis: ["setTimeout"] }); + const wrapper = new AgentSessionWrapper(proc, process.cwd()); + wrapper.start(); + + try { + let loginSettled = false; + const login = wrapper.send({ type: "login", provider: "test" }); + void login.then(() => { loginSettled = true; }, () => { loginSettled = true; }); + let mutationError; + const mutation = wrapper.send({ type, messageId: "queued-first" }); + void mutation.catch((error) => { mutationError = error; }); + await new Promise((resolve) => setImmediate(resolve)); + const first = transport.commands.find((command) => command.type === type); + assert.ok(first, "the child accepted the queue mutation but withheld its ACK"); + + t.mock.timers.tick(5_000); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(mutationError?.name, "RpcCommandTimeoutError", "a hung queue mutation must reject within five seconds"); + assert.equal(mutationError.command, type); + assert.equal(wrapper.isAlive(), true, "queue timeout must not kill unrelated work"); + assert.equal(loginSettled, false); + + let nextSettled = false; + const next = wrapper.send({ type, messageId: "queued-next" }); + void next.then(() => { nextSettled = true; }, () => { nextSettled = true; }); + await new Promise((resolve) => setImmediate(resolve)); + const mutations = transport.commands.filter((command) => command.type === type); + assert.equal(mutations.length, 2, "the timed-out mutation must not be automatically replayed"); + const second = mutations[1]; + transport.send({ type: "response", id: first.id, command: type, success: true, data: { messageId: "queued-first" } }); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(nextSettled, false, "a late ACK must not settle a subsequent queue mutation"); + + transport.send({ type: "response", id: second.id, command: type, success: true, data: { messageId: "queued-next" } }); + assert.deepEqual(await next, { messageId: "queued-next" }); + t.mock.timers.tick(60_000); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(loginSettled, false, "long unrelated commands retain their unlimited transport wait"); + const loginCommand = transport.commands.find((command) => command.type === "login"); + transport.send({ type: "response", id: loginCommand.id, command: "login", success: true, data: { loggedIn: true } }); + assert.deepEqual(await login, { loggedIn: true }); + } finally { + await wrapper.destroyAndWait(); + } + }); +} diff --git a/lib/rpc-manager.ts b/lib/rpc-manager.ts index de80c1e9..211d1769 100644 --- a/lib/rpc-manager.ts +++ b/lib/rpc-manager.ts @@ -1089,6 +1089,14 @@ export class AgentSessionWrapper { return null; } + case "remove_queued_message": + case "promote_queued_message": { + // Queue mutations are synchronous control operations, like get_state. + // Expiry must not abort unrelated work or replay a possibly applied mutation. + const result = await this.proc.sendCommand(command as { type: string }, GET_STATE_TIMEOUT_MS); + return result ?? null; + } + case "abort": await this.withFinalRunningNotification(async () => { await this.proc.sendCommand({ type: "abort" });