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<typeof normalizeGoalDraft>;
   /** 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<ChatSessionSnapshot>(`/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<typeof chatQueueInputRequestSchema>;
+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<typeof chatQueuedInputSchema>;
+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<Record<string, unknown>>(
+    `/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<T extends ConversationMessage>(
     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" ? <span className="personal-message-avatar"><Bot size={17} /></span> : null}
         <div>
           <header><strong>{item.message.role === "user" ? item.message.agentLabel ?? t("common.you") : item.message.agentLabel ?? t("header.manager")}</strong>{item.message.time ? <time>{item.message.time}</time> : null}</header>
+          {item.message.queueInputCancelled ? <span>{locale === "zh-CN" ? "已取消排队" : "Cancelled before starting"}</span> : null}
           {item.message.attachments?.length ? <div className="personal-message-images">{item.message.attachments.map((attachment) => isTextDocument(attachment.mimeType) ? <DocumentAttachment key={attachment.id} {...attachment} zh={locale === "zh-CN"}/> : <img alt={attachment.name} key={attachment.id} src={attachment.dataUrl} />)}</div> : null}
           {item.message.role === "user" ? <p>{item.message.text}</p> : item.message.text ? <HistoricalMessageText text={item.message.text}
             initiallyCollapsed={Boolean(historical?.has(messageKey(item.message)))}
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css
index 5ec1bbeb40..daabb32619 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css
+++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css
@@ -16,3 +16,9 @@
 .personal-queue-message p { white-space: pre-wrap; overflow-wrap: anywhere; }
 .personal-queue-message img { display: block; max-width: 100%; max-height: 120px; object-fit: contain; }
 .personal-queue-feedback { margin: 0; padding: 0 10px 8px; color: var(--pw-muted); overflow-wrap: anywhere; }
+.personal-queue-actions { display: flex; flex-direction: column; align-items: flex-end; flex-shrink: 0; }
+.personal-queue-menu { display: flex; flex-direction: column; align-items: stretch; }
+.personal-queue-menu button { justify-content: flex-start; white-space: nowrap; }
+.personal-queue-editor { padding: 0 10px 8px; }
+.personal-queue-editor textarea { display: block; box-sizing: border-box; width: 100%; resize: vertical; min-height: 72px; padding: 8px; border: 1px solid var(--pw-line-strong); border-radius: 6px; font: inherit; font-size: var(--text-body); color: var(--pw-text); background: #fff; }
+.personal-queue-editor > 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<ChatVisibleMessage | null>(null);
+  const [message, setMessage] = useState<ChatQueuedInput | null>(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<ChatQueueInputRequest | null>(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<string | null>(recovery.editor?.message ?? null);
+  const [editRevision, setEditRevision] = useState<number | null>(recovery.editor?.revision ?? null);
+  const textarea = useRef<HTMLTextAreaElement>(null), menuButton = useRef<HTMLButtonElement>(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 <li onKeyDown={event => {
+    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 ? <div className="personal-queue-message">
       {failed ? <p role="alert">{zh ? "完整消息读取失败" : "Full message unavailable"}<button type="button" onClick={() => setRetry(value => value + 1)}>{zh ? "重试" : "Retry"}</button></p>
         : !message ? <p role="status">{zh ? "正在读取完整消息…" : "Reading the full message…"}</p>
-        : <><p>{message.reply_context?.source_text ?? message.text}</p>
+        : <><p>{message.message}</p>
           {message.attachments?.map(attachment => isTextDocument(attachment.mime_type)
             ? <DocumentAttachment key={attachment.id} dataUrl={attachment.data_url} name={attachment.name} size={attachment.size} zh={zh}/>
             : <img key={attachment.id} src={attachment.data_url} alt={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 ? <RefreshCw size={14}/> : <CornerUpLeft size={14}/>}<span>{held ? (zh ? "核对插入" : "Check insertion") : (zh ? "插入本轮" : "Steer")}</span>
-  </button></div>
-    {held || actionError || turn.steering?.status === "failed" ? <p className="personal-queue-feedback" role={actionError ? "alert" : "status"}>
+  </button><div className="personal-queue-actions"><button ref={menuButton} type="button" disabled={busy || !editable}
+    aria-label={zh ? `更多操作：待处理消息 ${index + 1}` : `More actions for queued message ${index + 1}`}
+    aria-expanded={menu} onClick={() => setMenu(!menu)}><MoreHorizontal size={16}/></button>
+    {menu ? <div className="personal-queue-menu">
+      <button type="button" onClick={() => {setMenu(false); setEditing(true); setOpen(true);}}><Pencil size={14}/>{zh ? "编辑" : "Edit"}</button>
+      <button type="button" onClick={() => void mutate({action: "cancel", operation_id: crypto.randomUUID(),
+        expected_revision: turn.input_revision})}><X size={14}/>{zh ? "取消排队" : "Cancel queued message"}</button>
+    </div> : null}</div></div>
+    {editing ? <div className="personal-queue-editor">
+      {draft === null ? <p role="status">{failed ? (zh ? "完整消息读取失败" : "Full message unavailable") : (zh ? "正在读取…" : "Reading…")}</p>
+        : <textarea ref={textarea} aria-label={zh ? `编辑待处理消息 ${index + 1}` : `Edit queued message ${index + 1}`}
+          maxLength={24000} value={draft} onChange={event => {
+            setDraft(event.target.value);
+            try {localStorage.setItem(draftKey, JSON.stringify({message: event.target.value, revision: editRevision}));}
+            catch {setActionError(zh ? "草稿无法保存，切换前请保留文本。" : "Draft could not be saved. Retain the text before switching.");}
+          }} onCompositionStart={() => {composing.current = true;}} onCompositionEnd={() => {composing.current = false;}}
+          disabled={busy || Boolean(unresolved)} rows={3}/>}
+      <div><button type="button" onClick={save} disabled={busy || !draft?.trim() || Boolean(unresolved)}>{zh ? "保存" : "Save"}</button>
+        <button type="button" disabled={busy || Boolean(unresolved)} onClick={discard}>{zh ? "放弃修改" : "Discard edit"}</button></div>
+    </div> : null}
+    {unresolved ? <div className="personal-queue-feedback"><span>{zh ? "修改尚未确认" : "Change unconfirmed"}</span>
+      <button type="button" disabled={busy} onClick={() => void mutate(unresolved)}>{zh ? "核对修改" : "Check change"}</button></div> : null}
+    {held || actionError || recovery.failed || turn.steering?.status === "failed" ? <p className="personal-queue-feedback" role={actionError || recovery.failed ? "alert" : "status"}>
+      {recovery.failed ? (zh ? "恢复信息不可读，请保留当前草稿并重新打开会话核对。" : "Recovery information is unavailable. Retain your draft and reopen the conversation to check.") : null}
       {actionError || (held ? (zh ? "插入尚未确认，已暂停自动出队。" : "Insertion is unconfirmed; automatic dispatch is paused.")
         : (zh ? "上次插入未送达，消息仍在队列中。" : "The previous insertion was not delivered. The message is still queued."))}
     </p> : null}
   </li>;
 }
 
-export function ConversationQueue({sessionId, turns, error, onRetry, targetTurnId}: {
-  sessionId: string; turns: ChatQueuedTurn[]; error: Error | null; onRetry: () => void; targetTurnId?: string;
+export function ConversationQueue({sessionId, turns, error, onRetry, targetTurnId, onInputChanged = onRetry}: {
+  sessionId: string; turns: ChatQueuedTurn[]; error: Error | null; onRetry: () => void; targetTurnId?: string; onInputChanged?: () => void;
 }) {
   const { locale } = useWorkspaceI18n();
   const zh = locale === "zh-CN";
@@ -89,6 +189,6 @@ export function ConversationQueue({sessionId, turns, error, onRetry, targetTurnI
         aria-label={zh ? "重试读取待处理消息" : "Retry queued messages"}><RefreshCw size={14}/></button>
         <ContextHelp label={zh ? "队列读取失败原因" : "Queue read error"}><p>{error.message}</p></ContextHelp></> : null}</header>
     <ol>{turns.map((turn, index) => <QueuedMessage key={`${sessionId}:${turn.turn_id}`} sessionId={sessionId} turn={turn} index={index}
-      targetTurnId={targetTurnId} onChange={onRetry}/>)}</ol>
+      targetTurnId={targetTurnId} onChange={onRetry} onInputChanged={onInputChanged}/>)}</ol>
   </section>;
 }
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts
index 3d6bbf718a..994f3f712b 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts
+++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts
@@ -393,6 +393,7 @@ export type WorkspaceActionPreview = {
 };
 
 export type WorkspaceMessage = {
+  queueInputCancelled?: true;
   /** A native queue receipt, not an assistant answer or active execution. */
   queued?: boolean;
   replyContext?: ChatReplyContext;
diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx
index d1460739f5..e76c1c78ce 100644
--- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx
+++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx
@@ -3224,6 +3224,7 @@ export function PersonalWorkspacePage({
             ? "上次投递尚未确认，发送将重试原请求。"
             : "The previous delivery is unconfirmed. Send retries the original request."}</p> : null}
           {conversationSessionId ? <ConversationQueue sessionId={conversationSessionId}
+            onInputChanged={() => {conversationQueue.retry(); conversationHistoryState?.retry();}}
             targetTurnId={steeringTurnId} turns={conversationQueue.turns} error={conversationQueue.error} onRetry={conversationQueue.retry}/> : null}
             <div
               className="personal-channel-composer"
diff --git a/apps/presentation/dashboard/src/views/dashboard-page.tsx b/apps/presentation/dashboard/src/views/dashboard-page.tsx
index 77a108a16b..542093f88b 100644
--- a/apps/presentation/dashboard/src/views/dashboard-page.tsx
+++ b/apps/presentation/dashboard/src/views/dashboard-page.tsx
@@ -531,6 +531,8 @@ type PersonalHomeModel = {
   workers?: WorkspaceWorker[];
 };
 type PersonalManagerMessage = {
+  queueInputRevision?: number;
+  queueInputCancelled?: true;
   queued?: boolean;
   replyContext?: ChatVisibleMessage["reply_context"];
   goalDraft?: GoalDraft | null;
@@ -1758,6 +1760,7 @@ function PersonalGoalHome({
         role: message.role === "user" ? "user" : "assistant",
         returnDelivery: message.return_delivery, collaboration: message.collaboration, collaborations: message.collaborations,
         replyContext: message.reply_context,
+        queueInputRevision: message.queue_input_revision, queueInputCancelled: message.queue_input_cancelled,
         sourceLabel: message.role === "user" ? undefined : message.role === "error" ? "本地会话记录"
           : contextId === "manager" ? `恢复的${t("header.manager")}会话` : `恢复的 ${sessionLabels.get(message.session_id ?? "") ?? selectedAgent.label} 会话`,
         text: message.role === "user" ? message.reply_context?.source_text ?? message.text : visibleAgentMessage(message.text),
@@ -2931,6 +2934,7 @@ function PersonalGoalHome({
         returnDelivery: message.returnDelivery,
         collaboration: message.collaboration, collaborations: message.collaborations,
         replyContext: message.replyContext,
+        queueInputCancelled: message.queueInputCancelled,
         goalDraft: message.goalDraft,
         role: message.role,
         sourceTurnId: message.sourceTurnId,
@@ -3139,6 +3143,7 @@ function PersonalGoalHome({
                 agentLabel: message.role === "user" ? undefined : run.agentLabel,
                 attachments: workspaceImageAttachments(message.attachments),
                 replyContext: message.reply_context,
+                queueInputRevision: message.queue_input_revision, queueInputCancelled: message.queue_input_cancelled,
                 id: managerMessageId.current++,
                 lines: [],
                 role: message.role === "user" ? "user" : "assistant",
diff --git a/docs/integrations/session-runtime-control-plane-adapter.md b/docs/integrations/session-runtime-control-plane-adapter.md
index 868911c095..4163514740 100644
--- a/docs/integrations/session-runtime-control-plane-adapter.md
+++ b/docs/integrations/session-runtime-control-plane-adapter.md
@@ -618,6 +618,28 @@ composer, without recording an uncertain delivery for an attempt that was never
 dispatched. A genuinely uncertain request still retries its exact durable identity.
 This does not widen project grants or authorize another Session.
 
+Pending web inputs offer **Edit** and **Cancel queued message** in the queue's
+existing actions. Editing preserves attachments and the original text in the
+message's edit history. Cancellation retains the sent history and never calls
+the active Turn's interrupt operation. External, bound, collaboration and LoopX
+execution inputs continue through their original source entry.
+
+The existing paged history reader returns visible input and `queue_input_revision`
+from the same canonical user message, retaining original-source audience projection.
+`POST /api/chat/sessions/{session_id}/queue/{turn_id}/input` accepts an
+`action` (`edit` or `cancel`), `operation_id`, `expected_revision`, and text for
+an edit. The typed queue owner checks the revision and unclaimed Turn under the
+same Session/Turn fences as consumption and steering. A started Turn rejects a
+late change; an already applied operation replays its own receipt without
+changing newer input. A partially written operation reserves the item until its
+original identity finishes transcript/event publication. It cannot be steered
+or dispatched meanwhile. The browser retains unconfirmed operations and edit
+drafts per Session/Turn; **Check change** resolves the original operation rather
+than resending a message. A draft's revision remains fixed while editing, so a
+concurrent edit requires reconciliation instead of silently overwriting it.
+History reads show the confirmed edited text and a pre-start cancellation marker;
+Session metadata invalidates cached history without inventing worker activity.
+
 Queued acknowledgements do not show execution duration or completion. Short
 answers and steering instructions reconcile their canonical message identities
 and timestamps as well as long answers; an admission/history race cannot add
diff --git a/loopx/capabilities/native_chat/queue_api.py b/loopx/capabilities/native_chat/queue_api.py
index 52081a76c2..17be78f6d3 100644
--- a/loopx/capabilities/native_chat/queue_api.py
+++ b/loopx/capabilities/native_chat/queue_api.py
@@ -6,6 +6,20 @@
 
 
 class ChatQueueRequestMixin:
+    def _dispatch_chat_queue(self, path: str, *, apply: bool) -> bool:
+        parts = path.strip("/").split("/")
+        if parts[:3] != ["api", "chat", "sessions"] or len(parts) < 5 or parts[4] != "queue":
+            return False
+        if len(parts) == 5 and not apply:
+            self._session_queue(parts[3])
+        elif len(parts) == 7 and parts[6] == "input" and apply:
+            self._mutate_queued_turn(parts[3], parts[5])
+        elif len(parts) == 7 and parts[6] == "steer" and apply:
+            self._steer_queued_turn(parts[3], parts[5])
+        else:
+            return False
+        return True
+
     def _session_queue(self, session_id: str) -> None:
         try:
             store = self.server.chat_store
@@ -27,6 +41,8 @@ def _session_queue(self, session_id: str) -> None:
                     "character_count": len(text),
                     "attachment_count": len(turn.get("attachments") or []),
                     "steering": turn.get("_queue_steering"),
+                    "input_revision": turn.get("input_revision", 0),
+                    "mutation": turn.get("_queue_mutation"),
                 })
             self._send_json({"ok": True, "schema_version": CHAT_STORE_SCHEMA_VERSION,
                              "session_id": session_id, "active_turn_id": session.get("active_turn_id"), "turns": rows})
@@ -37,6 +53,23 @@ def _session_queue(self, session_id: str) -> None:
         except (OSError, RuntimeError):
             self._send_error("Queued messages are temporarily unavailable", status=503)
 
+    def _mutate_queued_turn(self, session_id: str, queued_turn_id: str) -> None:
+        try:
+            controller = self.server.runtime_controller
+            # Reuse current Session/source admission without requiring native
+            # steering support. Bound sources retain their original input entry.
+            controller._require_steering_session(session_id)
+            result = controller.store.mutate_queued_input(session_id, queued_turn_id, self._read_json())
+            self._send_json({"ok": True, **result})
+        except KeyError:
+            self._send_error("chat turn was not found", status=404)
+        except ValueError as exc:
+            self._send_error(str(exc), status=400)
+        except RuntimeError as exc:
+            self._send_error(str(exc), status=409, error_code=str(exc))
+        except OSError:
+            self._send_error("Queue change is unconfirmed; retry the original operation", status=503)
+
     def _steer_queued_turn(self, session_id: str, queued_turn_id: str) -> None:
         try:
             body = self._read_json()
diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py
index aced12cdf5..571576b60c 100644
--- a/loopx/chat_ingress.py
+++ b/loopx/chat_ingress.py
@@ -131,6 +131,67 @@ def queue_steering_lock(self, session_id: str, turn_id: str):
             stack.close()
             raise
 
+    def mutate_queued_input(self, session_id: str, turn_id: str, request: dict[str, Any]) -> dict[str, Any]:
+        """Execute the TS owner's recoverable input plan under the consumer fences."""
+        from .chat_store import _atomic_write_json, _read_jsonl, utc_now
+        from .control_plane.chat_turn_acceptance import plan_queued_chat_mutation
+        from .file_lock import write_bytes_atomic
+
+        with self.queue_steering_lock(session_id, turn_id):
+            session, turn = self.load_session(session_id), self.load_turn(session_id, turn_id)
+            if session is None or turn is None:
+                raise KeyError("chat turn was not found")
+            plan = plan_queued_chat_mutation(session, turn, request)
+            if plan["action"] == "reject":
+                raise RuntimeError(plan["error_code"])
+            receipt = plan["receipt"]
+            if plan["action"] == "apply":
+                if turn.get("_queue_mutation") is None:
+                    self._recover_queued_turn_input(session_id, turn)
+                path = self._turn_path(session_id, turn_id)
+                # The receipt precedes all other writes. A partial write stays
+                # reserved, including across process death; only this identity
+                # can finish it. No native call or active-Turn interrupt occurs.
+                turn = {**turn, "_queue_mutation": receipt,
+                        "_queue_original_message": turn.get("_queue_original_message", turn["message"])}
+                _atomic_write_json(path, turn, preserve_mode=True)
+                messages_path = self._session_dir(session_id) / "messages.jsonl"
+                with exclusive_file_lock(messages_path, agent_id="loopx-chat", operation="update_queued_input"):
+                    rows = _read_jsonl(messages_path, raise_on_error=True, strict=True)
+                    matches = [r for r in rows if r.get("turn_id") == turn_id and r.get("role") == "user"]
+                    if len(matches) != 1:
+                        raise OSError("Original queued input is unavailable")
+                    row = matches[0]
+                    if row.get("queue_input_revision", 0) != receipt["input_revision"]:
+                        if row.get("text") != turn["message"] or row.get("attachments") != turn.get("attachments"):
+                            raise OSError("Original queued input changed")
+                        if "text" in plan["message_patch"]:
+                            row["queue_edit_history"] = [*row.get("queue_edit_history", []),
+                                {"text": row["text"], "input_revision": turn.get("input_revision", 0),
+                                 "operation_id": receipt["operation_id"]}]
+                        row.update(plan["message_patch"])
+                        write_bytes_atomic(messages_path,
+                            ("".join(json.dumps(r, ensure_ascii=False) + "\n" for r in rows)).encode(), preserve_mode=True)
+                    elif any(row.get(key) != value for key, value in plan["message_patch"].items()):
+                        raise OSError("Queued input revision conflicts with its receipt")
+                event = plan["event"]
+                if not any(e.get("kind") == event["kind"] and e.get("payload") == event["payload"]
+                           for e in self.events_after(session_id, turn_id, None)):
+                    self.append_event(session_id, turn_id, kind=event["kind"], payload=event["payload"])
+                # Metadata edits do not invent worker activity; invalidate the
+                # existing Session history cache without changing liveness.
+                session["updated_at"] = utc_now()
+                _atomic_write_json(self._session_path(session_id), session, preserve_mode=True)
+                turn.update(plan["turn_patch"])
+                turn["_queue_operations"] = [*turn.get("_queue_operations", []), receipt]
+                turn.pop("_queue_mutation")
+                if receipt["action"] == "cancel":
+                    turn["completed_at"] = utc_now()
+                _atomic_write_json(path, turn, preserve_mode=True)
+            return {"session_id": session_id, "queued_turn_id": turn_id,
+                    "operation_id": receipt["operation_id"], "action": receipt["action"],
+                    "input_revision": receipt["input_revision"], "status": "applied"}
+
     def steering_receipt(self, session_id: str, ingress_id: str):
         """A corrupt receipt cannot be treated as absence and resent."""
         from .chat_store import CHAT_INGRESS_SCHEMA_VERSION
@@ -241,7 +302,7 @@ def create_queued_turn(
                 existing = self.turn_for_client(session_id, client_id)
                 if existing is not None:
                     require_matching_replay(
-                        existing,
+                        {**existing, "message": existing.get("_queue_original_message", existing.get("message"))},
                         identity="client_turn_id",
                         request={
                             "message": str(message),
diff --git a/loopx/chat_server.py b/loopx/chat_server.py
index b60461503a..c8c7a0f8bc 100644
--- a/loopx/chat_server.py
+++ b/loopx/chat_server.py
@@ -1739,8 +1739,8 @@ def do_GET(self) -> None:
         if len(action_parts) == 3 and action_parts[:2] == ["api", "actions"]:
             return self._action_snapshot(action_parts[2])
         session_parts = path.strip("/").split("/")
-        if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "queue":
-            return self._session_queue(session_parts[3])
+        if self._dispatch_chat_queue(path, apply=False):
+            return
         if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "native-turn":
             return self._managed_native_turn(session_parts[3])
         if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "loopx":
@@ -1765,8 +1765,8 @@ def do_POST(self) -> None:
         if path == "/api/chat/collaboration":
             return self._project_collaboration(apply=True)
         peer_parts = path.strip("/").split("/")
-        if len(peer_parts) == 7 and peer_parts[:3] == ["api", "chat", "sessions"] and peer_parts[4] == "queue" and peer_parts[6] == "steer":
-            return self._steer_queued_turn(peer_parts[3], peer_parts[5])
+        if self._dispatch_chat_queue(path, apply=True):
+            return
         if len(peer_parts) == 5 and peer_parts[:3] == ["api", "chat", "sessions"] and peer_parts[4] == "collaboration":
             return self._project_collaboration(peer_parts[3], apply=True)
         if self._dispatch_zcode_goal(path, apply=True):
diff --git a/loopx/control_plane/chat_turn_acceptance.py b/loopx/control_plane/chat_turn_acceptance.py
index 4f0396fbcb..de7a9fab8f 100644
--- a/loopx/control_plane/chat_turn_acceptance.py
+++ b/loopx/control_plane/chat_turn_acceptance.py
@@ -179,7 +179,8 @@ def queued_chat_turn_ids(
         "active_turn_id": active_turn_id, "include_reserved": include_reserved,
         "capacity_at": capacity_at,
         "turns": [{"turn_id": t["turn_id"], "status": t["status"],
-                   "expires_at": t.get("expires_at"), "reservation": t.get("_queue_steering")} for t in turns],
+                   "expires_at": t.get("expires_at"), "reservation": t.get("_queue_steering"),
+                   "mutation": t.get("_queue_mutation")} for t in turns],
     })["turn_ids"]
 
 
@@ -197,7 +198,7 @@ def plan_queued_chat_steering(
                  "has_collaboration_request": turn.get("collaboration_request") is not None,
                  "has_external_agent_target": turn.get("external_agent_target") is not None,
                  "attachment_count": len(turn.get("attachments") or []),
-                 "reservation": turn.get("_queue_steering")},
+                 "reservation": turn.get("_queue_steering"), "mutation": turn.get("_queue_mutation")},
         "receipt": ({key: receipt.get(key) for key in (
             "status", "mode", "expected_turn_id", "active_turn_id", "error_code",
         )} if receipt is not None else None),
@@ -205,6 +206,19 @@ def plan_queued_chat_steering(
     })
 
 
+def plan_queued_chat_mutation(session: dict[str, Any], turn: dict[str, Any], request: dict[str, Any]) -> dict[str, Any]:
+    return effect_runtime_result("chat.queue.mutate", {
+        "session": {"status": session.get("status"), "active_turn_id": session.get("active_turn_id")},
+        "turn": {"turn_id": turn["turn_id"], "status": turn["status"], "origin": turn.get("origin"),
+                 "has_collaboration_request": turn.get("collaboration_request") is not None,
+                 "has_external_agent_target": turn.get("external_agent_target") is not None,
+                 "loopx_execution": turn.get("loopx_execution", False), "has_acceptance": "_acceptance" in turn,
+                 "input_revision": turn.get("input_revision", 0), "reservation": turn.get("_queue_steering"),
+                 "mutation": turn.get("_queue_mutation"), "operations": turn.get("_queue_operations", [])},
+        "request": request,
+    })
+
+
 def _required_bool(payload: dict[str, Any], field: str) -> bool:
     value = payload.get(field)
     if not isinstance(value, bool):
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index 5e135a5af9..85ec873139 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -621,6 +621,7 @@ export function createEffectRuntimeHandlers(
     ["chat.turn.accept", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({planChatTurnAcceptance}) => planChatTurnAcceptance)],
     ["chat.queue.select", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({selectQueuedChatTurns}) => selectQueuedChatTurns)],
     ["chat.queue.steer", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatSteering}) => planQueuedChatSteering)],
+    ["chat.queue.mutate", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatMutation}) => planQueuedChatMutation)],
     ["chat.turn.execution_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayContinueChatTurn}) => mayContinueChatTurn)],
     ["chat.turn.recovery_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayRecoverChatTurn}) => mayRecoverChatTurn)],
     ["collaboration.delegation.observe", lazyHandler(() => import("./collaboration/delegation.ts"), ({transitionDelegationObservation}) => transitionDelegationObservation)],
diff --git a/loopx/control_plane/turn_driver/chat_queue_steering.ts b/loopx/control_plane/turn_driver/chat_queue_steering.ts
index 69d1ddb338..5ed1c4a2f3 100644
--- a/loopx/control_plane/turn_driver/chat_queue_steering.ts
+++ b/loopx/control_plane/turn_driver/chat_queue_steering.ts
@@ -28,15 +28,27 @@ function reservation(value: unknown): JsonObject | null {
   return row;
 }
 
+function mutation(value: unknown): JsonObject | null {
+  if (value == null) return null;
+  const row = requireJsonObject(value, "queue input receipt");
+  requireNonEmptyString(row.operation_id, "queue input operation_id");
+  if (!["edit", "cancel"].includes(String(row.action)) ||
+      !Number.isSafeInteger(row.expected_revision) || Number(row.expected_revision) < 0 ||
+      row.input_revision !== Number(row.expected_revision) + 1 ||
+      row.action === "edit" && typeof row.message !== "string") throw new Error("Invalid queue input receipt");
+  return row;
+}
+
 export function selectQueuedChatTurns(input: JsonObject): JsonObject {
   if (!Array.isArray(input.turns)) throw new Error("Queue turns must be an array");
   const ids = input.turns.flatMap(value => {
     const turn = requireJsonObject(value, "queue turn");
     const held = reservation(turn.reservation);
-    if (input.capacity_at && held?.status !== "pending" && typeof turn.expires_at === "string" &&
+    const editing = mutation(turn.mutation);
+    if (input.capacity_at && held?.status !== "pending" && !editing && typeof turn.expires_at === "string" &&
         Date.parse(turn.expires_at) <= Date.parse(String(input.capacity_at))) return [];
     return turn.status === "queued" && turn.turn_id !== input.active_turn_id &&
-      (input.include_reserved === true || held?.status !== "pending")
+      (input.include_reserved === true || (held?.status !== "pending" && !editing))
       ? [requireNonEmptyString(turn.turn_id, "turn.turn_id")] : [];
   });
   return {turn_ids: ids};
@@ -51,6 +63,7 @@ export function planQueuedChatSteering(input: JsonObject): QueueSteeringPlan {
   const held = reservation(turn.reservation);
   const reject = (code: string): QueueSteeringPlan => ({action: "reject", error_code: code});
   if (session.status === "closed") return reject("session_closed");
+  if (turn.mutation != null) return reject("queue_mutation_in_progress");
   if (held?.status === "pending" &&
       (held.client_ingress_id !== id || held.target_turn_id !== target)) {
     return reject("queue_steering_in_progress");
@@ -97,3 +110,54 @@ export function planQueuedChatSteering(input: JsonObject): QueueSteeringPlan {
     target_turn_id: target, operation_id: requireNonEmptyString(request.operation_id, "operation_id"), status: "pending",
   }};
 }
+
+/** Pending input changes share the queue consumer's Session/Turn fence.
+ * The pending receipt excludes dispatch until transcript/event IO is finished.
+ * Retried operation identities describe the same request, even after a claim.
+ */
+export function planQueuedChatMutation(input: JsonObject): JsonObject {
+  const session = requireJsonObject(input.session, "session");
+  const turn = requireJsonObject(input.turn, "queued turn");
+  const request = requireJsonObject(input.request, "queue input request");
+  const operationId = requireNonEmptyString(request.operation_id, "operation_id");
+  if (operationId.length > 160 || !/^[A-Za-z0-9._-]+$/.test(operationId) ||
+      !Number.isSafeInteger(request.expected_revision) || Number(request.expected_revision) < 0 ||
+      !["edit", "cancel"].includes(String(request.action)) ||
+      Object.keys(request).some(key => !["operation_id", "expected_revision", "action", "message"].includes(key))) {
+    throw new Error("Invalid queue input request");
+  }
+  if (request.action === "edit" && (typeof request.message !== "string" ||
+      !request.message.trim() || [...request.message].length > 12000) ||
+      request.action === "cancel" && "message" in request) throw new Error("Invalid queued message");
+  const reject = (error_code: string): JsonObject => ({action: "reject", error_code});
+  if (session.status === "closed") return reject("session_closed");
+  if (turn.origin !== "web" || turn.has_collaboration_request === true ||
+      turn.has_external_agent_target === true || turn.loopx_execution === true || turn.has_acceptance === true) {
+    return reject("queue_mutation_source_unavailable");
+  }
+  const pending = mutation(turn.mutation);
+  const operations = turn.operations ?? [];
+  if (!Array.isArray(operations)) throw new Error("Invalid queue input receipts");
+  const previous = operations.map(row => mutation(row)!)
+    .find(row => row.operation_id === operationId);
+  const held = pending ?? previous;
+  if (held) {
+    if (held.operation_id !== operationId) return reject("queue_mutation_in_progress");
+    if (held.action !== request.action || held.expected_revision !== request.expected_revision ||
+        held.message !== request.message) return reject("queue_mutation_identity_conflict");
+    if (!pending) return {action: "replay", receipt: held};
+  }
+  if (turn.status !== "queued" || session.active_turn_id === turn.turn_id) return reject("queue_turn_already_started");
+  if (reservation(turn.reservation)?.status === "pending") return reject("queue_steering_in_progress");
+  if (request.expected_revision !== (turn.input_revision ?? 0)) return reject("queue_input_revision_conflict");
+  const receipt = held ?? {...request, input_revision: Number(request.expected_revision) + 1};
+  return {action: "apply", receipt,
+    message_patch: request.action === "edit" ? {text: request.message, queue_input_revision: receipt.input_revision}
+      : {queue_input_cancelled: true, queue_input_revision: receipt.input_revision},
+    turn_patch: request.action === "edit" ? {message: request.message, input_revision: receipt.input_revision}
+      : {status: "interrupted", input_revision: receipt.input_revision},
+    event: request.action === "edit" ? {kind: "turn.input_edited", payload: {
+      operation_id: operationId, input_revision: receipt.input_revision}}
+      : {kind: "turn.interrupted", payload: {reason: "queued_input_cancelled", operation_id: operationId}},
+  };
+}
diff --git a/tests/control_plane_ts/chat_queue_steering.test.ts b/tests/control_plane_ts/chat_queue_steering.test.ts
index 41681d24ab..871c93edc1 100644
--- a/tests/control_plane_ts/chat_queue_steering.test.ts
+++ b/tests/control_plane_ts/chat_queue_steering.test.ts
@@ -1,6 +1,6 @@
 import assert from "node:assert/strict";
 import test from "node:test";
-import {planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts";
+import {planQueuedChatMutation, planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts";
 
 function facts() {
   return {phase: "prepare", session: {status: "busy", active_turn_id: "active"}, active_status: "running",
@@ -11,6 +11,40 @@ function facts() {
 const held = {schema_version: QUEUE_STEERING_SCHEMA, status: "pending", client_ingress_id: "native-op", operation_id: "op", target_turn_id: "active"};
 const receipt = {status: "delivered", expected_turn_id: "active", active_turn_id: "active", mode: "live_steering"};
 
+test("input changes require the exact unclaimed web input revision, without native steering", () => {
+  const f = facts();
+  const request = {action: "edit", expected_revision: 0, operation_id: "edit-1", message: "Corrected input"};
+  const input = {...f, request};
+  assert.equal(planQueuedChatMutation(input).action, "apply");
+  for (const status of ["starting", "running", "completed", "failed", "interrupted"]) {
+    assert.equal(planQueuedChatMutation({...input, turn: {...input.turn, status}}).error_code, "queue_turn_already_started");
+  }
+  assert.equal(planQueuedChatMutation({...input, turn: {...input.turn, input_revision: 1}}).error_code, "queue_input_revision_conflict");
+  for (const extra of [{origin: "external"}, {origin: "peer"}, {has_collaboration_request: true}, {loopx_execution: true}, {has_acceptance: true}]) {
+    assert.equal(planQueuedChatMutation({...input, turn: {...input.turn, ...extra}}).error_code, "queue_mutation_source_unavailable");
+  }
+  assert.equal(planQueuedChatMutation({...input, turn: {...input.turn, reservation: held}}).error_code, "queue_steering_in_progress");
+  assert.throws(() => planQueuedChatMutation({...input, request: {...request, message: " "}}));
+  assert.throws(() => planQueuedChatMutation({...input, request: {...request, action: "cancel"}}));
+});
+
+test("partial input writes exclude both FIFO and steering; exact retries finish or replay", () => {
+  const f = facts();
+  const request = {action: "cancel", expected_revision: 0, operation_id: "cancel-1"};
+  const pending = { ...request, input_revision: 1 };
+  const turn = {...f.turn, mutation: pending};
+  assert.deepEqual(selectQueuedChatTurns({turns: [turn]}).turn_ids, []);
+  assert.deepEqual(selectQueuedChatTurns({turns: [turn], include_reserved: true}).turn_ids, ["queued"]);
+  assert.deepEqual(selectQueuedChatTurns({turns: [{...turn, expires_at: "2020-01-01T00:00:00Z"}],
+    capacity_at: "2026-01-01T00:00:00Z", include_reserved: true}).turn_ids, ["queued"]);
+  assert.equal(planQueuedChatSteering({...f, turn}).error_code, "queue_mutation_in_progress");
+  assert.equal(planQueuedChatMutation({...f, turn, request}).action, "apply");
+  assert.equal(planQueuedChatMutation({...f, turn, request: {...request, operation_id: "different"}}).error_code, "queue_mutation_in_progress");
+  const ended = {...f.turn, status: "interrupted", input_revision: 1, operations: [pending]};
+  assert.equal(planQueuedChatMutation({...f, turn: ended, request}).action, "replay");
+  assert.equal(planQueuedChatMutation({...f, turn: ended, request: {...request, expected_revision: 1}}).error_code, "queue_mutation_identity_conflict");
+});
+
 test("queued steering requires an unclaimed original and exact live target", () => {
   const f = facts(); assert.equal(planQueuedChatSteering(f).action, "dispatch");
   for (const status of ["starting", "running", "interrupting", "completed", "failed"]) {
diff --git a/tests/presentation/test_chat_turn_queue.py b/tests/presentation/test_chat_turn_queue.py
index 76d324d158..7545dccaaf 100644
--- a/tests/presentation/test_chat_turn_queue.py
+++ b/tests/presentation/test_chat_turn_queue.py
@@ -8,6 +8,181 @@
 from test_chat_ordinary_project import ordinary  # noqa: F401
 
 
+def queue_input_body(action="edit", revision=0, operation="input-1", message="Corrected queued input."):
+    return {"action": action, "expected_revision": revision, "operation_id": operation,
+            **({"message": message} if action == "edit" else {})}
+
+
+def test_edit_cancel_and_admission_retries_keep_exact_input_history_and_active_turn(ordinary):  # noqa: F811
+    store, runtime, contexts, request, capture, *_ = ordinary
+    sid, active, queued, original = held_queue(ordinary)
+    path = f"/api/chat/sessions/{sid}/queue/{queued}/input"
+    before = capture.read_bytes()
+    assert request(path)[0] == 404
+    assert store.load_turn(sid, queued)["message"] == original["message"]  # GET never mutates.
+    body = queue_input_body()
+    code, edited = request(path, body)
+    assert code == 200 and edited["input_revision"] == 1
+    assert request(path, body) == (code, edited)
+    assert request(path, queue_input_body(operation="stale"))[1]["error_code"] == "queue_input_revision_conflict"
+    # A lost original admission ACK cannot overwrite the explicit edit.
+    assert request(f"/api/chat/sessions/{sid}/turns", original)[1]["turn_id"] == queued
+    assert store.load_turn(sid, queued)["message"] == body["message"]
+    transcript = next(m for m in store.messages(sid) if m.get("turn_id") == queued)
+    assert transcript["text"] == body["message"]
+    assert transcript["queue_edit_history"] == [{"text": original["message"], "input_revision": 0, "operation_id": "input-1"}]
+    cancel = queue_input_body("cancel", 1, "cancel-1")
+    result = request(path, cancel)
+    assert result[0] == 200 and request(path, cancel) == result
+    assert request(path, body) == (code, edited)  # receipt replay, not another edit
+    assert store.load_turn(sid, queued)["status"] == "interrupted"
+    snapshot = request(f"/api/chat/sessions/{sid}")[1]
+    cancelled = next(row for row in snapshot["messages"] if row.get("turn_id") == queued)
+    assert cancelled["text"] == body["message"] and cancelled["queue_input_cancelled"] is True
+    assert cancelled["queue_input_revision"] == 2
+    assert store.load_session(sid)["active_turn_id"] == active
+    assert request(f"/api/chat/sessions/{sid}/queue")[1]["turns"] == []
+    assert len([e for e in store.events_after(sid, queued, None) if e["kind"] == "turn.interrupted"]) == 1
+    assert capture.read_bytes() == before  # edit/cancel never issue native I/O
+
+
+def test_partial_edit_is_reserved_and_recovers_the_same_operation_after_reopening(ordinary, monkeypatch):  # noqa: F811
+    import loopx.chat_store as store_io
+    from loopx.chat_store import ChatSessionStore
+    store, runtime, contexts, request, capture, *_ = ordinary
+    sid, active, queued, _ = held_queue(ordinary)
+    path = f"/api/chat/sessions/{sid}/queue/{queued}/input"
+    body = queue_input_body()
+    write = store_io._atomic_write_json
+
+    def fail_commit(path, value, **kwargs):
+        if value.get("_queue_operations") and not value.get("_queue_mutation"):
+            raise OSError("lost commit")
+        return write(path, value, **kwargs)
+
+    with monkeypatch.context() as patch:
+        patch.setattr(store_io, "_atomic_write_json", fail_commit)
+        assert request(path, body)[0] == 503
+    reopened = ChatSessionStore(store.root.parent)
+    assert reopened.queued_turns(sid) == []
+    assert reopened.queued_turns(sid, include_reserved=True)[0]["_queue_mutation"]["operation_id"] == "input-1"
+    assert request(path, queue_input_body(operation="replacement"))[1]["error_code"] == "queue_mutation_in_progress"
+    assert request(f"/api/chat/sessions/{sid}/queue/{queued}/steer", {
+        "target_turn_id": active, "client_ingress_id": "do-not-send"})[1]["error_code"] == "queue_mutation_in_progress"
+    assert request(path, body)[0] == 200
+    assert len(next(m for m in store.messages(sid) if m.get("turn_id") == queued)["queue_edit_history"]) == 1
+    assert len([e for e in store.events_after(sid, queued, None) if e["kind"] == "turn.input_edited"]) == 1
+
+
+@pytest.mark.parametrize("action", ["edit", "cancel"])
+def test_input_change_and_queue_claim_have_both_serial_orders(ordinary, monkeypatch, action):  # noqa: F811
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import Event
+    import loopx.chat_store as store_io
+    store, runtime, contexts, request, capture, *_ = ordinary
+    monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: None)
+    sid, active, queued, _ = held_queue(ordinary)
+    # Stop this isolated native fixture through its real owner; no automatic
+    # queue worker is installed, so the following claims have controlled order.
+    runtime.interrupt_turn(session_id=sid, turn_id=active)
+    runtime.wait_for_turn(session_id=sid, turn_id=active, timeout_sec=10)
+    entered, proceed = Event(), Event()
+    write = store_io._atomic_write_json
+
+    def pause(path, value, **kwargs):
+        if value.get("_queue_mutation"):
+            entered.set()
+            assert proceed.wait(5)
+        return write(path, value, **kwargs)
+
+    with monkeypatch.context() as patch, ThreadPoolExecutor(max_workers=2) as pool:
+        patch.setattr(store_io, "_atomic_write_json", pause)
+        changing = pool.submit(request, f"/api/chat/sessions/{sid}/queue/{queued}/input", queue_input_body(action))
+        assert entered.wait(5)
+        claiming = pool.submit(store.claim_next_queued_turn, sid, host_claim_id="test-host")
+        assert not claiming.done()
+        proceed.set()
+        assert changing.result(timeout=10)[0] == 200
+        claimed = claiming.result(timeout=10)
+    if action == "edit":
+        assert claimed["message"] == "Corrected queued input."
+    else:
+        assert claimed is None
+    other, _ = store.create_queued_turn(sid, client_turn_id="other", message="Already claimed.", origin="web")
+    if claimed:
+        store.release_active_turn(sid, queued)
+    assert store.claim_next_queued_turn(sid, host_claim_id="test-host")["turn_id"] == other["turn_id"]
+    before = store.load_turn(sid, other["turn_id"])
+    rejected = request(f"/api/chat/sessions/{sid}/queue/{other['turn_id']}/input", queue_input_body(action))
+    assert rejected[0] == 409 and rejected[1]["error_code"] == "queue_turn_already_started"
+    assert store.load_turn(sid, other["turn_id"]) == before
+
+
+def test_mutation_respects_revoked_scope_external_source_and_other_session(ordinary):  # noqa: F811
+    from loopx.capabilities.native_chat.project_context import ChatProjectContexts
+    store, runtime, contexts, request, capture, *_ = ordinary
+    sid, active, queued, _ = held_queue(ordinary)
+    external, _ = store.create_queued_turn(sid, client_turn_id="external", message="Original external request", origin="external")
+    assert request(f"/api/chat/sessions/{sid}/queue/{external['turn_id']}/input", queue_input_body())[1]["error_code"] == "queue_mutation_source_unavailable"
+    original = store.load_turn(sid, queued)
+    other = store.create_session(goal_id="other", agent_id="codex", adapter_kind="codex", upstream_thread_id="other-thread")
+    assert request(f"/api/chat/sessions/{other['session_id']}/queue/{queued}/input", queue_input_body())[0] == 404
+    runtime.project_contexts = ChatProjectContexts([])
+    assert request(f"/api/chat/sessions/{sid}/queue/{queued}/input", queue_input_body())[0] == 400
+    assert store.load_turn(sid, queued) == original
+
+
+def test_edited_input_and_attachments_are_actually_dispatched_once(ordinary):  # noqa: F811
+    import base64
+    store, runtime, contexts, request, capture, *_ = ordinary
+    data = b"# Original source\nKeep this attachment.\n"
+    attachment = {"id": "source", "name": "source.md", "mime_type": "text/markdown",
+                  "size": len(data), "data_url": "data:text/markdown;base64," + base64.b64encode(data).decode()}
+    sid, active, queued, _ = held_queue(ordinary, attachments=[attachment])
+    body = queue_input_body()
+    assert request(f"/api/chat/sessions/{sid}/queue/{queued}/input", body)[0] == 200
+    runtime.steer_active_turn(session_id=sid, expected_turn_id=active, client_ingress_id="finish-current", message="Finish current work.")
+    assert runtime.wait_for_turn(session_id=sid, turn_id=queued, timeout_sec=10)["status"] == "completed"
+    calls = [json.loads(line) for line in capture.read_text().splitlines()]
+    starts = [c for c in calls if c["method"] == "turn/start"]
+    assert len(starts) == 2
+    inputs = json.dumps(starts[1]["params"]["input"])
+    assert body["message"] in inputs and "Keep this attachment." in inputs
+    assert store.load_turn(sid, queued)["attachments"] == [attachment]
+    # Rechecking the applied edit after native execution cannot start it again.
+    before = capture.read_bytes()
+    assert request(f"/api/chat/sessions/{sid}/queue/{queued}/input", body)[0] == 200
+    assert capture.read_bytes() == before
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_goal_queue_input_http_with_canonical_goal_providers(ordinary, tmp_path, monkeypatch, provider):  # noqa: F811
+    from tests.control_plane.canonical_authority_fixture import promoted_create_fixture, isolate_sqlite_runtime
+    from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted
+    (tmp_path / "provider").mkdir()
+    isolate_sqlite_runtime(tmp_path / "provider", monkeypatch)
+    registry, root, state = promoted_create_fixture(tmp_path / "goal-source", provider=provider)
+    store, runtime, contexts, request, capture, *_ = ordinary
+    server = next(c.cell_contents for c in request.__closure__ if hasattr(c.cell_contents, "serve_forever"))
+    server.registry_path, runtime.registry_path = registry, registry
+    server.runtime_root_override = str(root)
+    before = read_canonical_todos_if_promoted(runtime_root=root, goal_id="goal-a")
+    session = store.create_session(goal_id="goal-a", agent_id="codex", adapter_kind="codex", upstream_thread_id="goal-thread")
+    sid = session["session_id"]
+    queued, _ = store.create_queued_turn(sid, client_turn_id="goal-input", message="Original Goal input.", origin="web")
+    qid = queued["turn_id"]
+    assert request(f"/api/chat/sessions/{sid}/queue")[0] == 200
+    path = f"/api/chat/sessions/{sid}/queue/{qid}/input"
+    assert request(path, queue_input_body())[0] == 200
+    history = request(f"/api/chat/sessions/{sid}?turn_id={qid}&message_limit=200")[1]["messages"]
+    assert history[0]["text"] == "Corrected queued input." and history[0]["queue_input_revision"] == 1
+    assert request(path, queue_input_body("cancel", 1, "goal-cancel"))[0] == 200
+    assert store.load_turn(sid, qid)["status"] == "interrupted"
+    after = read_canonical_todos_if_promoted(runtime_root=root, goal_id="goal-a")
+    assert after["provider_revision"] == before["provider_revision"]
+    assert after["todos"] == before["todos"] and not capture.exists()
+
+
 def wait_for_active_turn(store, session_id, turn_id):
     deadline = time.monotonic() + 5
     while time.monotonic() < deadline: