From 7774095cf61287ff891ba6daa52080736ba20f13 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 13:57:04 +0800 Subject: [PATCH] feat(chat): edit and cancel pending queue inputs Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- apps/presentation/dashboard/src/data/chat.ts | 38 ++++ .../src/data/conversation-returns.test.mjs | 14 ++ .../src/data/conversation-returns.ts | 15 +- .../personal-workspace/channel-timeline.tsx | 1 + .../personal-workspace/conversation-queue.css | 6 + .../personal-workspace/conversation-queue.tsx | 140 ++++++++++++-- .../personal-workspace-model.ts | 1 + .../personal-workspace-page.tsx | 1 + .../dashboard/src/views/dashboard-page.tsx | 5 + .../session-runtime-control-plane-adapter.md | 22 +++ loopx/capabilities/native_chat/queue_api.py | 33 ++++ loopx/chat_ingress.py | 63 ++++++- loopx/chat_server.py | 8 +- loopx/control_plane/chat_turn_acceptance.py | 18 +- .../control_plane/effect_runtime_handlers.ts | 1 + .../turn_driver/chat_queue_steering.ts | 68 ++++++- .../chat_queue_steering.test.ts | 36 +++- tests/presentation/test_chat_turn_queue.py | 175 ++++++++++++++++++ 18 files changed, 612 insertions(+), 33 deletions(-) diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 749832fee6..529da55fa0 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -895,6 +895,8 @@ export type ManagerRuntimeSessionReadback = { }; export type ChatVisibleMessage = { + queue_input_revision?: number; + queue_input_cancelled?: true; goal_draft?: ReturnType; /** Client-side lineage added when messages from several Sessions are merged. */ session_id?: string; @@ -965,6 +967,27 @@ export async function fetchChatSession(sessionId: string, signal?: AbortSignal) return requestJson(`/api/chat/sessions/${encodeURIComponent(sessionId)}`, { signal }); } +export const chatQueueInputRequestSchema = z.discriminatedUnion("action", [ + z.object({action: z.literal("edit"), operation_id: z.string().min(1).max(160), + expected_revision: z.number().int().nonnegative(), message: z.string().min(1).max(24000)}).strict(), + z.object({action: z.literal("cancel"), operation_id: z.string().min(1).max(160), + expected_revision: z.number().int().nonnegative()}).strict(), +]); +export type ChatQueueInputRequest = z.infer; +const chatQueuedInputSchema = z.object({input_revision: z.number().int().nonnegative(), message: z.string(), + attachments: z.array(z.object({id: z.string(), name: z.string(), size: z.number(), + mime_type: z.string(), data_url: z.string()})), +}); +export type ChatQueuedInput = z.infer; +export async function fetchQueuedChatInput(sessionId: string, turnId: string, signal?: AbortSignal) { + // Keep original-source audience projection and complete paging. The visible + // input and its revision come from one canonical user row, never raw model context. + const history = await fetchChatSessionHistory(sessionId, signal, turnId); + const original = history.messages.find(row => row.role === "user" && row.turn_id === turnId); + if (!original) throw new Error("Original queued input is unavailable"); + return chatQueuedInputSchema.parse({input_revision: original.queue_input_revision ?? 0, + message: original.reply_context?.source_text ?? original.text, attachments: original.attachments ?? []}); +} const chatQueueSchema = z.object({ ok: z.literal(true), schema_version: z.literal("loopx_chat_store_v1"), session_id: z.string().min(1), active_turn_id: z.string().min(1).nullable(), @@ -972,6 +995,10 @@ const chatQueueSchema = z.object({ turn_id: z.string().min(1), created_at: z.string(), origin: z.string().nullable(), preview: z.string(), character_count: z.number().int().nonnegative(), attachment_count: z.number().int().nonnegative(), + input_revision: z.number().int().nonnegative().default(0), + mutation: z.object({action: z.enum(["edit", "cancel"]), operation_id: z.string().min(1), + expected_revision: z.number().int().nonnegative(), input_revision: z.number().int().positive(), + message: z.string().optional()}).nullable().optional().default(null), steering: z.object({ schema_version: z.literal("loopx_chat_queue_steering_v1"), client_ingress_id: z.string().min(1), operation_id: z.string().min(1), @@ -1004,6 +1031,17 @@ export async function steerQueuedChatTurn(sessionId: string, queuedId: string, t } } +export async function mutateQueuedChatTurn(sessionId: string, queuedId: string, request: ChatQueueInputRequest) { + const value = await requestJson>( + `/api/chat/sessions/${encodeURIComponent(sessionId)}/queue/${encodeURIComponent(queuedId)}/input`, + {method: "POST", body: JSON.stringify(request), signal: AbortSignal.timeout(15_000)}); + if (value.ok !== true || value.session_id !== sessionId || value.queued_turn_id !== queuedId || + value.operation_id !== request.operation_id || value.action !== request.action || + value.input_revision !== request.expected_revision + 1 || value.status !== "applied") { + throw new Error("Queue input acknowledgement does not match the original request"); + } +} + const chatMessagePageSchema = z.object({ schema_version: z.literal("loopx_chat_message_page_v1"), offset: z.number().int().nonnegative(), total: z.number().int().nonnegative(), diff --git a/apps/presentation/dashboard/src/data/conversation-returns.test.mjs b/apps/presentation/dashboard/src/data/conversation-returns.test.mjs index e478d8688c..0cebf54a37 100644 --- a/apps/presentation/dashboard/src/data/conversation-returns.test.mjs +++ b/apps/presentation/dashboard/src/data/conversation-returns.test.mjs @@ -2,6 +2,20 @@ import assert from "node:assert/strict"; import { conversationReturnSessions, conversationPendingReturnSessions, reconcileConversationHistory, reconcileConversationReturns } from "./conversation-returns.ts"; const collaboration = { returns: [] }; +const queuedInput = [{sourceSessionId: "queued-session", sourceMessageId: "input", sourceTurnId: "queued-turn", + role: "user", text: "Original input"}]; +const editedInput = {message_id: "input", turn_id: "queued-turn", role: "user", text: "Corrected input", queue_input_revision: 1}; +const corrected = reconcileConversationReturns(queuedInput, "queued-session", [editedInput], () => assert.fail("no duplicate input")); +assert.equal(corrected[0].text, "Corrected input"); +assert.equal(reconcileConversationReturns(queuedInput, "queued-session", [{...editedInput, role: "agent"}], () => assert.fail()), queuedInput); +assert.equal(reconcileConversationReturns(queuedInput, "queued-session", [{...editedInput, turn_id: "other-turn"}], () => assert.fail()), queuedInput); +const cancelledInput = {...editedInput, queue_input_revision: 2, queue_input_cancelled: true}; +const cancelled = reconcileConversationReturns(corrected, "queued-session", [cancelledInput], () => assert.fail("no duplicate input")); +assert.equal(cancelled[0].queueInputCancelled, true); +assert.equal(reconcileConversationReturns(cancelled, "queued-session", [editedInput], () => assert.fail()), cancelled, "stale reads cannot revert cancellation"); +assert.equal(reconcileConversationReturns(queuedInput, "other-session", [cancelledInput], () => assert.fail()), queuedInput); +const liveAnswer = [{...queuedInput[0], role: "assistant", pending: true, text: "Live answer"}]; +assert.equal(reconcileConversationReturns(liveAnswer, "queued-session", [{...cancelledInput, role: "agent"}], () => assert.fail())[0].text, "Live answer"); const original = [ { sourceSessionId: "old", sourceMessageId: "brief", collaboration, text: "Original brief" }, { sourceSessionId: "current", sourceMessageId: "brief", text: "New conversation" }, diff --git a/apps/presentation/dashboard/src/data/conversation-returns.ts b/apps/presentation/dashboard/src/data/conversation-returns.ts index 4052d63f5c..20b4e4c5fc 100644 --- a/apps/presentation/dashboard/src/data/conversation-returns.ts +++ b/apps/presentation/dashboard/src/data/conversation-returns.ts @@ -1,6 +1,8 @@ import type { ChatVisibleMessage } from "./chat"; type ConversationMessage = { + queueInputRevision?: number; + queueInputCancelled?: true; sourceSessionId?: string; sourceMessageId?: string; sourceTurnId?: string; @@ -74,14 +76,21 @@ export function reconcileConversationReturns( const replyContext = source.reply_context ?? row.replyContext; // Only the exact canonical user body can switch to its source-backed view. // A background read never replaces a live answer or a different draft. - const text = row.role === "user" && source.reply_context && row.text === source.text + const inputChanged = row.role === "user" && source.role === "user" + && (!row.sourceTurnId || source.turn_id === row.sourceTurnId) && typeof source.queue_input_revision === "number" + && source.queue_input_revision > (row.queueInputRevision ?? 0); + const queueInputRevision = inputChanged ? source.queue_input_revision : row.queueInputRevision; + const queueInputCancelled = inputChanged ? source.queue_input_cancelled : row.queueInputCancelled; + const text = inputChanged ? source.text : row.role === "user" && source.reply_context && row.text === source.text ? source.reply_context.source_text : row.text; if (row.sourceMessageId === source.message_id && JSON.stringify(row.returnDelivery) === JSON.stringify(returnDelivery) && JSON.stringify(row.collaboration) === JSON.stringify(collaboration) && JSON.stringify(row.collaborations) === JSON.stringify(collaborations) && row.sourceCreatedAt === sourceCreatedAt - && JSON.stringify(row.replyContext) === JSON.stringify(replyContext) && row.text === text) return row; + && JSON.stringify(row.replyContext) === JSON.stringify(replyContext) && row.text === text + && row.queueInputRevision === queueInputRevision && row.queueInputCancelled === queueInputCancelled) return row; changed = true; - return { ...row, sourceMessageId: source.message_id, sourceCreatedAt, returnDelivery, collaboration, collaborations, replyContext, text }; + return { ...row, sourceMessageId: source.message_id, sourceCreatedAt, returnDelivery, collaboration, collaborations, replyContext, text, + queueInputRevision, queueInputCancelled }; }); for (const message of messages) { if (message.origin !== "manager_followup" || seen.has(message.message_id)) continue; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx b/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx index ecdbbdab7d..3c9884b3f3 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx @@ -207,6 +207,7 @@ export function ChannelTimeline({ {item.message.role !== "user" ? : null}
{item.message.role === "user" ? item.message.agentLabel ?? t("common.you") : item.message.agentLabel ?? t("header.manager")}{item.message.time ? : null}
+ {item.message.queueInputCancelled ? {locale === "zh-CN" ? "已取消排队" : "Cancelled before starting"} : null} {item.message.attachments?.length ?
{item.message.attachments.map((attachment) => isTextDocument(attachment.mimeType) ? : {attachment.name})}
: null} {item.message.role === "user" ?

{item.message.text}

: item.message.text ? div { display: flex; justify-content: flex-end; gap: 6px; margin-top: 4px; } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx index ee44b325d2..b734e12b05 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx @@ -1,24 +1,95 @@ import { useEffect, useRef, useState } from "react"; -import { CornerUpLeft, RefreshCw } from "lucide-react"; -import { ChatApiError, fetchChatSessionHistory, steerQueuedChatTurn, type ChatQueuedTurn, type ChatVisibleMessage } from "../../data/chat"; +import { CornerUpLeft, MoreHorizontal, Pencil, RefreshCw, X } from "lucide-react"; +import { ChatApiError, chatQueueInputRequestSchema, fetchQueuedChatInput, mutateQueuedChatTurn, steerQueuedChatTurn, + type ChatQueueInputRequest, type ChatQueuedInput, type ChatQueuedTurn } from "../../data/chat"; import { composerKeyAction } from "./composer-keyboard"; import { DocumentAttachment, isTextDocument } from "./document-attachment"; import { ContextHelp } from "./context-help"; import { useWorkspaceI18n } from "./i18n"; import "./conversation-queue.css"; -function QueuedMessage({sessionId, turn, index, targetTurnId, onChange}: { - sessionId: string; turn: ChatQueuedTurn; index: number; targetTurnId?: string; onChange: () => void; +function QueuedMessage({sessionId, turn, index, targetTurnId, onChange, onInputChanged}: { + sessionId: string; turn: ChatQueuedTurn; index: number; targetTurnId?: string; onChange: () => void; onInputChanged: () => void; }) { const { locale } = useWorkspaceI18n(); const zh = locale === "zh-CN"; const [open, setOpen] = useState(false), [retry, setRetry] = useState(0); - const [message, setMessage] = useState(null); + const [message, setMessage] = useState(null); const [failed, setFailed] = useState(false); const lock = useRef(false); const [busy, setBusy] = useState(false), [actionError, setActionError] = useState(""); + const recoveryKey = `loopx.queue-input:${JSON.stringify([sessionId, turn.turn_id])}`; + const draftKey = `${recoveryKey}:draft`; + const [recovery] = useState(() => { + try { + if (typeof window === "undefined") return {request: null, editor: null, failed: false}; + const raw = localStorage.getItem(recoveryKey); + const saved = localStorage.getItem(draftKey); + const editor = saved ? JSON.parse(saved) as {message: string; revision: number} : null; + if (editor && (typeof editor.message !== "string" || !Number.isSafeInteger(editor.revision) || editor.revision < 0)) throw new Error("Invalid edit recovery"); + return {request: raw ? chatQueueInputRequestSchema.parse(JSON.parse(raw)) : null, editor, failed: false}; + } catch {return {request: null, editor: null, failed: true};} + }); + const [pending, setPending] = useState(recovery.request); + const serverPending = turn.mutation ? chatQueueInputRequestSchema.parse({ + action: turn.mutation.action, operation_id: turn.mutation.operation_id, + expected_revision: turn.mutation.expected_revision, + ...(turn.mutation.action === "edit" ? {message: turn.mutation.message} : {}), + }) : null; + const unresolved = serverPending ?? pending; + const [menu, setMenu] = useState(false), [editing, setEditing] = useState(Boolean(recovery.editor)); + const [draft, setDraft] = useState(recovery.editor?.message ?? null); + const [editRevision, setEditRevision] = useState(recovery.editor?.revision ?? null); + const textarea = useRef(null), menuButton = useRef(null); + const composing = useRef(false); const held = turn.steering?.status === "pending" ? turn.steering : null; - const canSteer = Boolean(held || (targetTurnId && !turn.attachment_count)); + const canSteer = !unresolved && !editing && !recovery.failed && Boolean(held || (targetTurnId && !turn.attachment_count)); + const editable = !held && !unresolved && !recovery.failed && turn.origin === "web"; + useEffect(() => {setMessage(null);}, [turn.input_revision]); + useEffect(() => {if (editing && message && draft === null) { + setDraft(message.message); setEditRevision(message.input_revision); + }}, [editing, message, draft]); + useEffect(() => {if (editing && draft !== null) textarea.current?.focus();}, [editing, draft === null]); + async function mutate(request: ChatQueueInputRequest) { + if (lock.current) return; + lock.current = true; setBusy(true); setActionError(""); setMenu(false); + let retained = false; + try { + // Failure to durably retain the operation must precede any network effect. + localStorage.setItem(recoveryKey, JSON.stringify(request)); + retained = true; + setPending(request); + await mutateQueuedChatTurn(sessionId, turn.turn_id, request); + setPending(null); setEditing(false); setDraft(null); setMessage(null); setEditRevision(null); + localStorage.removeItem(recoveryKey); + localStorage.removeItem(draftKey); + } catch (cause) { + const code = cause instanceof ChatApiError ? cause.payload.error_code : ""; + const rejected = cause instanceof ChatApiError && (cause.payload.http_status === 400 || cause.payload.http_status === 404 || + ["queue_turn_already_started", "queue_input_revision_conflict", "queue_mutation_identity_conflict", + "queue_mutation_source_unavailable", "queue_steering_in_progress", "queue_mutation_in_progress", "session_closed"].includes(String(code))); + if (rejected) { + setPending(null); + try {localStorage.removeItem(recoveryKey);} catch { /* Preserve the draft if storage is unavailable. */ } + } + setActionError(!retained + ? (zh ? "恢复信息无法保存，修改尚未提交。请保留文本后重试。" : "Recovery could not be saved. The change was not submitted; retain the text and retry.") + : rejected + ? (zh ? "消息状态或权限已变化，修改未应用。请核对当前状态；编辑草稿已保留。" : "The message state or permission changed. Check its current state; your edit is retained.") + : (zh ? "修改尚未确认，请核对原操作；不会重新发送消息。" : "The change is unconfirmed. Check the original operation; the message will not be resent.")); + } finally {lock.current = false; setBusy(false); onInputChanged();} + } + function save() { + if (draft && [...draft].length > 12000) { + setActionError(zh ? "待处理消息最多 12000 个字符，请缩短后保存。" : "Queued messages support up to 12000 characters. Shorten the text before saving."); return; + } + if (draft?.trim() && editRevision !== null) void mutate({action: "edit", operation_id: crypto.randomUUID(), + expected_revision: editRevision, message: draft}); + } + function discard() { + try {localStorage.removeItem(draftKey);} catch {setActionError(zh ? "草稿恢复信息无法清除，请保留文本。" : "Draft recovery could not be cleared. Retain the text."); return;} + setEditing(false); setDraft(null); setEditRevision(null); setMenu(false); menuButton.current?.focus(); + } async function steer() { if (lock.current || !canSteer) return; lock.current = true; setBusy(true); setActionError(""); @@ -35,18 +106,25 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange}: { } finally {lock.current = false; setBusy(false); onChange();} } useEffect(() => { - if (!open || message) return; + if ((!open && !editing) || message) return; const controller = new AbortController(); setFailed(false); - void fetchChatSessionHistory(sessionId, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)]), turn.turn_id) - .then(snapshot => { - const original = snapshot.messages.find(row => row.role === "user" && row.turn_id === turn.turn_id); - if (!original) throw new Error("Original queued input is unavailable"); - if (!controller.signal.aborted) setMessage(original); - }).catch(() => { if (!controller.signal.aborted) setFailed(true); }); + void fetchQueuedChatInput(sessionId, turn.turn_id, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)])) + .then(input => {if (!controller.signal.aborted) setMessage(input);}) + .catch(() => { if (!controller.signal.aborted) setFailed(true); }); return () => controller.abort(); - }, [open, message, sessionId, turn.turn_id, retry]); + }, [open, editing, message, sessionId, turn.turn_id, turn.input_revision, retry]); return
  • { + if (composing.current || event.nativeEvent.isComposing || event.nativeEvent.keyCode === 229) return; + if (event.key === "Escape" && !busy && !unresolved) { + event.preventDefault(); event.stopPropagation(); discard(); return; + } + if (editing) { + if (composerKeyAction({...event, isComposing: event.nativeEvent.isComposing, keyCode: event.nativeEvent.keyCode}, false) === "steer") { + event.preventDefault(); event.stopPropagation(); save(); + } + return; + } if (composerKeyAction({...event, isComposing: event.nativeEvent.isComposing, keyCode: event.nativeEvent.keyCode}, false) === "steer" && canSteer) { event.preventDefault(); event.stopPropagation(); void steer(); } @@ -58,7 +136,7 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange}: { {open ?
    {failed ?

    {zh ? "完整消息读取失败" : "Full message unavailable"}

    : !message ?

    {zh ? "正在读取完整消息…" : "Reading the full message…"}

    - : <>

    {message.reply_context?.source_text ?? message.text}

    + : <>

    {message.message}

    {message.attachments?.map(attachment => isTextDocument(attachment.mime_type) ? : {attachment.name}/)}} @@ -69,16 +147,38 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange}: { : !targetTurnId && !held ? (zh ? "当前回合或执行器不支持插入，消息仍在队列中" : "This turn or executor cannot be steered; the message stays queued") : (zh ? "插入本轮 · Command+Enter" : "Steer · Command+Enter")}> {busy ? : }{held ? (zh ? "核对插入" : "Check insertion") : (zh ? "插入本轮" : "Steer")} -
    - {held || actionError || turn.steering?.status === "failed" ?

    +

    + {menu ?
    + + +
    : null}
    • + {editing ?
    + {draft === null ?

    {failed ? (zh ? "完整消息读取失败" : "Full message unavailable") : (zh ? "正在读取…" : "Reading…")}

    + :