loopx-agent · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026 · Nov 5, 2026
@@ -5,6 +5,7 @@
   "license": "Apache-2.0",
   "type": "module",
   "scripts": {
+    "smoke:queue-input-recovery": "vite build --ssr smoke/queue-input-recovery-smoke.mjs --outDir node_modules/.cache/queue-input-recovery --emptyOutDir && node node_modules/.cache/queue-input-recovery/queue-input-recovery-smoke.js",
     "smoke:todo-ownership": "vite build --ssr smoke/todo-ownership-smoke.mjs --outDir node_modules/.cache/todo-ownership --emptyOutDir && node node_modules/.cache/todo-ownership/todo-ownership-smoke.js",
     "smoke:workspace-progressive-loader": "node smoke/workspace-progressive-status-smoke.mjs",
     "smoke:workspace-progressive": "node ../../../examples/workspace-progressive-loading-browser-smoke.mjs",

@@ -0,0 +1,53 @@
+import assert from "node:assert/strict";
+import { listQueueInputRecovery, queueInputKeys, readQueueInputRecovery, retireQueueInputRequest } from "../src/features/personal-workspace/queue-input-recovery";
+
+class MemoryStorage {
+  rows = new Map();
+  get length() {return this.rows.size;}
+  key(index) {return [...this.rows.keys()][index] ?? null;}
+  getItem(key) {return this.rows.get(key) ?? null;}
+  setItem(key, value) {this.rows.set(key, value);}
+  removeItem(key) {this.rows.delete(key);}
+}
+const storage = new MemoryStorage(), session = "original-session", turn = "original-turn";
+const keys = queueInputKeys(session, turn);
+const request = {action: "edit", operation_id: "original-operation", expected_revision: 0, message: "First correction"};
+function retain(editor = {message: request.message, revision: 0}) {
+  storage.setItem(keys.request, JSON.stringify(request));
+  storage.setItem(keys.draft, JSON.stringify(editor));
+}
+retain();
+assert.deepEqual(listQueueInputRecovery(storage, "other-session").entries, [], "Session isolation applies to local recovery");
+assert.equal(listQueueInputRecovery(storage, session).entries.length, 1, "Request and draft describe one retained input");
+// While A's response is pending, another page changes its draft to B.
+retain({message: "Newer correction from another page", revision: 0});
+retireQueueInputRequest(storage, session, turn, request, true);
+assert.equal(readQueueInputRecovery(storage, session, turn).editor.message, "Newer correction from another page");
+assert.equal(readQueueInputRecovery(storage, session, turn).request, null);
+// A's response must not retire a later operation identity either.
+const newer = {...request, operation_id: "later-operation", message: "Later request"};
+storage.setItem(keys.request, JSON.stringify(newer));
+retireQueueInputRequest(storage, session, turn, request, true);
+assert.deepEqual(readQueueInputRecovery(storage, session, turn).request, newer);
+// A request retained before IO can be rejected after the original starts.
+storage.removeItem(keys.draft); storage.setItem(keys.request, JSON.stringify(request));
+retireQueueInputRequest(storage, session, turn, request, false);
+assert.equal(readQueueInputRecovery(storage, session, turn).editor.message, request.message);
+assert.equal(readQueueInputRecovery(storage, session, turn).request, null);
+// Successful cleanup removes only matching text and revision.
+retain({message: request.message, revision: 1});
+retireQueueInputRequest(storage, session, turn, request, true);
+assert.equal(readQueueInputRecovery(storage, session, turn).editor.revision, 1);
+retain(); retireQueueInputRequest(storage, session, turn, request, true);
+assert.deepEqual(listQueueInputRecovery(storage, session).entries, []);
+// Failure while retiring the draft keeps the original retry identity.
+retain();
+const remove = storage.removeItem.bind(storage);
+storage.removeItem = key => {if (key === keys.draft) throw new Error("Storage unavailable"); remove(key);};
+assert.throws(() => retireQueueInputRequest(storage, session, turn, request, true));
+assert.equal(readQueueInputRecovery(storage, session, turn).request.operation_id, request.operation_id);
+storage.removeItem = remove;
+storage.setItem(keys.draft, "0");
+assert.equal(listQueueInputRecovery(storage, session).failed, true, "Invalid recovery is not an empty success");
+assert.equal(listQueueInputRecovery({get length() {throw new Error("Storage blocked");}}, session).failed, true);
+console.log("queue-input-recovery smoke ok");
@@ -142,6 +142,31 @@ assert.equal(observedBoth[0].text, "Ask both");
 assert.deepEqual(conversationPendingReturnSessions(observedBoth), []);
 assert.equal(reconcileConversationReturns(observedBoth, "original", [{message_id: "multi"}], createReply), observedBoth);
 
+// History can be displayed before the original admission ACK supplies the
+// live row's identity. Its stream callback target must survive that handoff.
+const historyFirst = {sourceSessionId: "current", sourceMessageId: "agent-message", sourceTurnId: "same-turn",
+  role: "assistant", text: "Stored answer", marker: "history"};
+const lateBoundStream = {sourceSessionId: "current", sourceTurnId: "same-turn", role: "assistant",
+  text: "Live fragment", pending: true, marker: "stream", activity: ["Running"]};
+const admittedAfterRead = reconcileConversationHistory([historyFirst, lateBoundStream], [storedRoles[1]], createHistory);
+assert.equal(admittedAfterRead.length, 1, "One canonical reply remains when history arrived first");
+assert.equal(admittedAfterRead[0].marker, "stream", "The original stream callback target and controls survive");
+assert.equal(admittedAfterRead[0].sourceMessageId, "agent-message");
+assert.equal(admittedAfterRead[0].pending, true);
+assert.equal(admittedAfterRead[0].text, "Live fragment", "Readback does not overwrite streaming text");
+assert.deepEqual(admittedAfterRead[0].activity, ["Running"]);
+assert.equal(reconcileConversationHistory(admittedAfterRead, [storedRoles[1]], createHistory), admittedAfterRead);
+const separateReplies = [historyFirst, lateBoundStream,
+  {...historyFirst, sourceSessionId: "foreign", marker: "foreign"},
+  {...historyFirst, sourceMessageId: "another-answer", sourceTurnId: "another-turn", marker: "another-turn"},
+  {...historyFirst, sourceMessageId: "followup", marker: "followup"}];
+const separateSources = [storedRoles[1], {...storedRoles[1], message_id: "followup", origin: "manager_followup"},
+  {...storedRoles[1], message_id: "another-answer", turn_id: "another-turn"}];
+const distinct = reconcileConversationHistory(separateReplies, separateSources, createHistory);
+assert.deepEqual(distinct.map(row => row.marker), ["stream", "foreign", "another-turn", "followup"],
+  "Equal text, other Sessions/Turns and separate follow-up identities remain distinct");
+console.log("conversation-returns: history-before-admission preserves the live identity and distinct replies");
+
 // A revised brief keeps its return obligation despite a delivered old conclusion.
 const revised = [{sourceSessionId: "original", collaborations: [
   {request_id: "a", request_revision: 2, returns: [{phase: "conclusion", status: "delivered", request_revision: 1}]},

@@ -112,10 +112,15 @@ export function reconcileConversationHistory<T extends ConversationMessage>(
   }
   // A history read can arrive before admission hydrates the optimistic row.
   // Deduplicate only established Session/message identities, never body text.
+  // A live row owns the stream callbacks and controls even when the history
+  // row appeared first. Keep that row when both acquire the same identity.
+  const pendingIdentities = new Set(updated.filter(message => message.pending && message.sourceSessionId && message.sourceMessageId)
+    .map(message => JSON.stringify([message.sourceSessionId, message.sourceMessageId])));
   const identities = new Set<string>();
   const unique = updated.filter((message) => {
     if (!message.sourceSessionId || !message.sourceMessageId) return true;
     const key = JSON.stringify([message.sourceSessionId, message.sourceMessageId]);
+    if (pendingIdentities.has(key) && !message.pending) return false;
     if (identities.has(key)) return false;
     identities.add(key);
     return true;

@@ -36,6 +36,6 @@ export function useConversationQueue({sessionId, scope, active, refreshKey}: {
     return () => { controller.abort(); clearTimeout(timer); };
   }, [key, sessionId, active, refreshKey, attempt]);
   const current = sessionId && state?.key === key ? state : null;
-  return {turns: current?.turns ?? [], activeTurnId: current?.activeTurnId ?? null, error: current?.error ?? null,
+  return {turns: current?.turns ?? [], activeTurnId: current?.activeTurnId ?? null, error: current?.error ?? null, readReady: Boolean(current && !current.error),
     retry: () => setAttempt(value => value + 1)};
 }
@@ -5,6 +5,8 @@ import { ChatApiError, chatQueueInputRequestSchema, fetchQueuedChatInput, mutate
 import { composerKeyAction } from "./composer-keyboard";
 import { DocumentAttachment, isTextDocument } from "./document-attachment";
 import { ContextHelp } from "./context-help";
+import { queueInputKeys, readQueueInputRecovery, retireQueueInputRequest } from "./queue-input-recovery";
+import { RetainedQueueInputs } from "./retained-queue-inputs";
 import { useWorkspaceI18n } from "./i18n";
 import "./conversation-queue.css";
 
@@ -18,16 +20,11 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange, onInputC
   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 {request: recoveryKey, draft: draftKey} = queueInputKeys(sessionId, turn.turn_id);
   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};
+      return {...readQueueInputRecovery(localStorage, sessionId, turn.turn_id), failed: false};
     } catch {return {request: null, editor: null, failed: true};}
   });
   const [pending, setPending] = useState<ChatQueueInputRequest | null>(recovery.request);
@@ -61,16 +58,15 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange, onInputC
       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);
+      retireQueueInputRequest(localStorage, sessionId, turn.turn_id, request, true);
     } 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. */ }
+        try {retireQueueInputRequest(localStorage, sessionId, turn.turn_id, request, false);} 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.")
@@ -177,18 +173,19 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange, onInputC
   </li>;
 }
 
-export function ConversationQueue({sessionId, turns, error, onRetry, targetTurnId, onInputChanged = onRetry}: {
-  sessionId: string; turns: ChatQueuedTurn[]; error: Error | null; onRetry: () => void; targetTurnId?: string; onInputChanged?: () => void;
+export function ConversationQueue({sessionId, turns, error, readReady, onRetry, targetTurnId, onInputChanged = onRetry}: {
+  sessionId: string; turns: ChatQueuedTurn[]; error: Error | null; readReady: boolean; onRetry: () => void; targetTurnId?: string; onInputChanged?: () => void;
 }) {
   const { locale } = useWorkspaceI18n();
   const zh = locale === "zh-CN";
-  if (!turns.length && !error) return null;
-  return <section className="personal-conversation-queue" aria-label={zh ? "待处理消息" : "Queued messages"}>
+  return <>{turns.length || error ? <section className="personal-conversation-queue" aria-label={zh ? "待处理消息" : "Queued messages"}>
     <header><span>{zh ? "待处理" : "Queued"}{error ? "" : ` · ${turns.length}`}</span>
       {error ? <><span role="alert">{zh ? "读取失败" : "Read failed"}</span><button type="button" onClick={onRetry}
         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} onInputChanged={onInputChanged}/>)}</ol>
-  </section>;
+  </section> : null}
+    {readReady && !error ? <RetainedQueueInputs key={sessionId} sessionId={sessionId} turns={turns} onChange={onInputChanged}/> : null}
+  </>;
 }
@@ -3224,6 +3224,7 @@ export function PersonalWorkspacePage({
             ? "上次投递尚未确认，发送将重试原请求。"
             : "The previous delivery is unconfirmed. Send retries the original request."}</p> : null}
           {conversationSessionId ? <ConversationQueue sessionId={conversationSessionId}
+            readReady={conversationQueue.readReady}
             onInputChanged={() => {conversationQueue.retry(); conversationHistoryState?.retry();}}
             targetTurnId={steeringTurnId} turns={conversationQueue.turns} error={conversationQueue.error} onRetry={conversationQueue.retry}/> : null}
             <div

@@ -0,0 +1,57 @@
+import { chatQueueInputRequestSchema, type ChatQueueInputRequest } from "../../data/chat";
+
+export function queueInputKeys(sessionId: string, turnId: string) {
+  const request = `loopx.queue-input:${JSON.stringify([sessionId, turnId])}`;
+  return {request, draft: `${request}:draft`};
+}
+
+export function readQueueInputRecovery(storage: Storage, sessionId: string, turnId: string) {
+  const keys = queueInputKeys(sessionId, turnId);
+  const raw = storage.getItem(keys.request), saved = storage.getItem(keys.draft);
+  const editor: {message: string; revision: number} | null = saved === null ? null : JSON.parse(saved);
+  if (saved !== null && (!editor || typeof editor !== "object" || typeof editor.message !== "string" || !Number.isSafeInteger(editor.revision) || editor.revision < 0)) {
+    throw new Error("Invalid edit recovery");
+  }
+  return {request: raw === null ? null : chatQueueInputRequestSchema.parse(JSON.parse(raw)), editor};
+}
+
+export type QueueInputRecovery = {
+  turnId: string; request: ChatQueueInputRequest | null; editor: {message: string; revision: number} | null;
+};
+
+/** An old response may retire only its own retained intent, never a newer edit. */
+export function retireQueueInputRequest(storage: Storage, sessionId: string, turnId: string,
+  request: ChatQueueInputRequest, applied: boolean) {
+  const current = readQueueInputRecovery(storage, sessionId, turnId);
+  if (JSON.stringify(current.request) !== JSON.stringify(chatQueueInputRequestSchema.parse(request))) return;
+  const keys = queueInputKeys(sessionId, turnId);
+  if (!applied && request.action === "edit" && !current.editor) storage.setItem(keys.draft,
+    JSON.stringify({message: request.message, revision: request.expected_revision}));
+  if (applied && request.action === "edit" && current.editor?.message === request.message &&
+      current.editor.revision === request.expected_revision) storage.removeItem(keys.draft);
+  storage.removeItem(keys.request);
+}
+
+/** This is unsent input/retry identity, never a second queue or a Turn status. */
+export function listQueueInputRecovery(storage: Storage, sessionId: string) {
+  const ids = new Set<string>();
+  let failed = false;
+  const entries: QueueInputRecovery[] = [];
+  try {
+    for (let i = 0; i < storage.length; i += 1) {
+      const key = storage.key(i);
+      if (!key?.startsWith("loopx.queue-input:")) continue;
+      try {
+        const identity: unknown = JSON.parse(key.slice("loopx.queue-input:".length).replace(/:draft$/, ""));
+        if (Array.isArray(identity) && identity.length === 2 && identity[0] === sessionId && typeof identity[1] === "string") ids.add(identity[1]);
+      } catch { /* An unresolvable identity must not be assigned to this Session. */ }
+    }
+    for (const turnId of ids) {
+      try {
+        const input = readQueueInputRecovery(storage, sessionId, turnId);
+        if (input.request || input.editor) entries.push({turnId, ...input});
+      } catch {failed = true;}
+    }
+  } catch {failed = true;}
+  return {entries, failed};
+}