fix(chat): recover replies completed between queue observations

apps/presentation/dashboard/src/data/conversation-returns.test.mjs

@@ -167,6 +167,28 @@ assert.deepEqual(distinct.map(row => row.marker), ["stream", "foreign", "another
   "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 fast queued Turn can finish before its start/stream is observed. Its
+// canonical answer replaces the waiting receipt, never a live stream.
+const queuedReply = {sourceSessionId: "current", sourceTurnId: "same-turn", role: "assistant",
+  text: "", queued: true, pending: false};
+const completedQueue = reconcileConversationHistory([queuedReply], [storedRoles[1]], createHistory);
+assert.equal(completedQueue.length, 1);
+assert.equal(completedQueue[0].text, storedRoles[1].text);
+assert.equal(completedQueue[0].queued, false);
+assert.equal(completedQueue[0].sourceMessageId, "agent-message");
+assert.equal(reconcileConversationHistory(completedQueue, [storedRoles[1]], createHistory), completedQueue);
+const unboundQueuedReply = {...queuedReply, sourceTurnId: undefined, sourceMessageId: "agent-message"};
+const unboundAnswer = reconcileConversationReturns([unboundQueuedReply], "current", [{...storedRoles[1], turn_id: undefined}], createHistory);
+assert.equal(unboundAnswer[0].text, "");
+assert.equal(unboundAnswer[0].queued, true, "Two absent Turn identities do not prove queue execution");
+for (const change of [{pending: true, text: "Live fragment"}, {queued: false, text: "Existing answer"},
+  {sourceSessionId: "foreign"}, {sourceTurnId: "another-turn"}]) {
+  const original = {...queuedReply, ...change};
+  const observed = reconcileConversationHistory([original], [storedRoles[1]], createHistory);
+  assert.equal(observed.find(row => row === original || row.sourceTurnId === original.sourceTurnId
+    && row.sourceSessionId === original.sourceSessionId)?.text, original.text);
+}
+
 // 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}]},

apps/presentation/dashboard/src/data/conversation-returns.ts

@@ -9,6 +9,7 @@ type ConversationMessage = {
   sourceCreatedAt?: string;
   role?: string;
   pending?: boolean;
+  queued?: boolean;
   collaboration?: ChatVisibleMessage["collaboration"];
   collaborations?: ChatVisibleMessage["collaborations"];
   returnDelivery?: ChatVisibleMessage["return_delivery"];
@@ -81,16 +82,19 @@ export function reconcileConversationReturns<T extends ConversationMessage>(
       && 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
+    const queueAnswered = row.queued === true && !row.pending && row.role === "assistant"
+      && Boolean(row.sourceTurnId) && source.role !== "user" && source.origin !== "manager_followup" && source.turn_id === row.sourceTurnId;
+    const queued = queueAnswered ? false : row.queued;
+    const text = inputChanged || queueAnswered ? 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
-      && row.queueInputRevision === queueInputRevision && row.queueInputCancelled === queueInputCancelled) return row;
+      && row.queueInputRevision === queueInputRevision && row.queueInputCancelled === queueInputCancelled && row.queued === queued) return row;
     changed = true;
     return { ...row, sourceMessageId: source.message_id, sourceCreatedAt, returnDelivery, collaboration, collaborations, replyContext, text,
-      queueInputRevision, queueInputCancelled };
+      queueInputRevision, queueInputCancelled, ...(queueAnswered ? {queued: false} : {}) };
   });
   for (const message of messages) {
     if (message.origin !== "manager_followup" || seen.has(message.message_id)) continue;

apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts

@@ -581,7 +581,7 @@ export type PersonalWorkspaceCallbacks = {
   onInterruptConversationTurn?: (contextId: string, turnId: string) => Promise<void>;
   onSteerConversationTurn?: (contextId: string, turnId: string, message: string, ingressId: string) => Promise<void>;
   onQueueConversationTurn?: (contextId: string, sessionId: string, message: string, ingressId: string, attachments?: WorkspaceImageAttachment[]) => Promise<void>;
-  onObserveConversationTurn?: (contextId: string, sessionId: string, turnId: string) => Promise<void>;
+  onObserveConversationQueue?: (contextId: string, sessionId: string) => Promise<void>;
   onOpenGoal?: (goalId: string) => void | Promise<void>;
   onOpenGoalView?: (tab: WorkspaceGoalTab) => void;
   onOpenRunSession?: (run: WorkspaceRun) => void | Promise<void>;

apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx

@@ -1310,18 +1310,25 @@ export function PersonalWorkspacePage({
     refreshKey: `${managerMessages.length}:${Boolean(managerMessages.at(-1)?.pending)}:${steering}`,
   });
   const queuedTurnKeys = new Set(conversationQueue.turns.map(turn => JSON.stringify([conversationSessionId, turn.turn_id])));
-  const observedQueueTurn = useRef("");
+  const observedQueue = useRef<{scope: string; snapshot: string} | null>(null);
   useEffect(() => {
     const turnId = conversationQueue.activeTurnId;
-    if (!conversationSessionId || !turnId || turnId === runningMessage?.sourceTurnId || !callbacks.onObserveConversationTurn) return;
-    const key = JSON.stringify([statusSourceControl?.activeSource.statusUrl, conversationKey, conversationSessionId, turnId]);
-    if (observedQueueTurn.current === key) return;
-    observedQueueTurn.current = key;
-    void callbacks.onObserveConversationTurn(conversationKey, conversationSessionId, turnId).catch(() => {
-      if (observedQueueTurn.current === key) observedQueueTurn.current = "";
+    if (!conversationSessionId || !conversationQueue.readReady || !callbacks.onObserveConversationQueue) return;
+    const scope = JSON.stringify([statusSourceControl?.activeSource.statusUrl, conversationKey, conversationSessionId]);
+    const snapshot = JSON.stringify([turnId, conversationQueue.turns.map(turn => turn.turn_id)]);
+    const previous = observedQueue.current;
+    if (previous?.scope === scope && previous.snapshot === snapshot) return;
+    observedQueue.current = {scope, snapshot};
+    if (previous?.scope !== scope && (!turnId || turnId === runningMessage?.sourceTurnId)) return;
+    // A queued Turn can start and finish between polls after reconnect. Its
+    // disappearance still requires original history, even with no live Turn.
+    void callbacks.onObserveConversationQueue(conversationKey, conversationSessionId).catch(() => {
+      if (observedQueue.current?.scope === scope && observedQueue.current.snapshot === snapshot) {
+        observedQueue.current = {scope, snapshot: ""};
+      }
     });
-  }, [conversationQueue.activeTurnId, conversationQueue.turns, conversationSessionId, runningMessage?.sourceTurnId,
-    callbacks.onObserveConversationTurn, conversationKey, statusSourceControl?.activeSource.statusUrl]);
+  }, [conversationQueue.activeTurnId, conversationQueue.turns, conversationQueue.readReady, conversationSessionId, runningMessage?.sourceTurnId,
+    callbacks.onObserveConversationQueue, conversationKey, statusSourceControl?.activeSource.statusUrl]);
   const legacyDrafts = conversationSessionId ? legacySteeringRequests(conversationSessionId, composerDraftKey) : [];
   function restoreLegacyDraft(key: string) {
     if (composer.trim() || !conversationSessionId) return;

apps/presentation/dashboard/src/views/dashboard-page.tsx

@@ -1,7 +1,7 @@
 import {useManagedConversation} from "../data/use-managed-conversation";
 import { normalizeGoalDraft, type GoalDraft } from "../../../../../loopx/control_plane/collaboration/goal_draft.js";
 import { withTurnActivity, type TurnStep } from "../data/turn-steps";
-import { conversationReturnSessions, conversationPendingReturnSessions, reconcileConversationHistory, reconcileConversationReturns } from "../data/conversation-returns";
+import { conversationReturnSessions, conversationPendingReturnSessions, reconcileConversationHistory } from "../data/conversation-returns";
 import { readConversationReturns, readTaskSessions } from "../data/conversation-return-observation";
 import { currentChannelSession, useConversationHistory } from "../data/use-conversation-history";
 import { useChatProjects } from "../data/use-chat-projects";
@@ -82,6 +82,7 @@ import {
   type ChatSessionSummary,
   type ChatImageAttachment,
   type ChatVisibleMessage,
+  mergeChatSessionMessages,
   type ManagerChannelBinding,
   type ManagerRuntimeSessionReadback,
   type AgentResponse,
@@ -1585,6 +1586,23 @@ function PersonalGoalHome({
     return target.kind === "managed" ? target.agentId : target.kind === "manager" ? t("header.manager") : goalFallback;
   };
 
+  function conversationHistoryMessage(targetContextId: string, message: ChatVisibleMessage, sessionLabel: string): PersonalManagerMessage {
+    return {
+      sourceMessageId: message.message_id, sourceSessionId: message.session_id,
+      sourceTurnId: message.turn_id ?? undefined, sourceCreatedAt: message.created_at,
+      goalDraft: normalizeGoalDraft(message.goal_draft),
+      agentLabel: message.role === "user" ? message.origin === "peer" ? (locale === "zh-CN" ? "协作者" : "Collaborator") : undefined
+        : message.origin === "manager_followup" ? "协作回执" : answerIdentityLabel(targetContextId, sessionLabel),
+      attachments: workspaceImageAttachments(message.attachments),
+      id: managerMessageId.current++, lines: [], 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" ? "本地会话记录"
+        : targetContextId === "manager" ? `恢复的${t("header.manager")}会话` : `恢复的 ${sessionLabel} 会话`,
+      text: message.role === "user" ? message.reply_context?.source_text ?? message.text : visibleAgentMessage(message.text),
+    };
+  }
+
   useEffect(() => {
     const resolved = resolveLocalStatusUrl(
       statusSourceControl.activeSource.statusUrl,
@@ -1658,13 +1676,14 @@ function PersonalGoalHome({
               for (const [targetContextId, ids] of Object.entries(contexts)) {
                 if (!ids.includes(sessionId)) continue;
                 const previous = current[targetContextId] ?? [];
-                const updated = reconcileConversationReturns(previous, sessionId, snapshot.messages, (row) => ({
-                  id: managerMessageId.current++, sourceMessageId: row.message_id,
-                  sourceSessionId: sessionId, sourceCreatedAt: row.created_at,
-                  role: "assistant" as const,
-                  agentLabel: "协作回执", sourceLabel: "协作回执", text: visibleAgentMessage(row.text), lines: [],
-                  returnDelivery: row.return_delivery, collaboration: row.collaboration, collaborations: row.collaborations,
-                }));
+                // A terminal queue read can precede the final transcript write.
+                // The existing revision observer must also adopt later native
+                // rows after reload has retired their optimistic placeholders.
+                const messages = mergeChatSessionMessages([snapshot]);
+                const sessionLabel = runtimeAgents.find(agent => agent.agent_id === snapshot.session.agent_id)?.display_name
+                  ?? snapshot.session.agent_id;
+                const updated = reconcileConversationHistory(previous, messages,
+                  row => conversationHistoryMessage(targetContextId, row, sessionLabel));
                 if (updated !== previous) next = { ...next, [targetContextId]: updated };
               }
               return next;
@@ -1780,24 +1799,8 @@ function PersonalGoalHome({
       : conversationHistory.history.messages;
     const sessionLabels = new Map(conversationHistory.history.sessions.map(session => [session.session_id,
       runtimeAgents.find(agent => agent.agent_id === session.agent_id)?.display_name ?? session.agent_id]));
-    return reconcileConversationHistory(previous, messages, (message): PersonalManagerMessage => ({
-      sourceMessageId: message.message_id,
-      sourceSessionId: message.session_id,
-      sourceTurnId: message.turn_id ?? undefined,
-      sourceCreatedAt: message.created_at,
-      goalDraft: normalizeGoalDraft(message.goal_draft),
-      agentLabel: message.role === "user" ? message.origin === "peer" ? (locale === "zh-CN" ? "协作者" : "Collaborator") : undefined : message.origin === "manager_followup"
-        ? "协作回执" : answerIdentityLabel(contextId, sessionLabels.get(message.session_id ?? "") ?? selectedAgent.label),
-      attachments: workspaceImageAttachments(message.attachments),
-      id: managerMessageId.current++, lines: [],
-      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),
-    }));
+    return reconcileConversationHistory(previous, messages,
+      message => conversationHistoryMessage(targetContextId, message, sessionLabels.get(message.session_id ?? "") ?? selectedAgent.label));
   };
   useEffect(() => {
     if (!conversationHistory.history) return;
@@ -3347,17 +3350,17 @@ function PersonalGoalHome({
             void sendManagerQuestion(message, { agentId: runtimeBindings[targetContextId]?.agentId,
               contextId: targetContextId, attachments, queue: { sessionId, ingressId, onAccepted: resolve } }).catch(reject);
           }),
-          onObserveConversationTurn: async (targetContextId, sessionId, turnId) => {
+          onObserveConversationQueue: async (targetContextId, sessionId) => {
             const key = `${targetContextId}:${selectedAgent.agentId}`;
-            if (activeConversationContext.current !== targetContextId || sessionIds.current.get(key) !== sessionId
-                || (activeTurnIds.current.get(targetContextId) === turnId
-                  && streamControllers.current.get(targetContextId)?.signal.aborted === false)) return;
+            if (activeConversationContext.current !== targetContextId || sessionIds.current.get(key) !== sessionId) return;
             // Queue advancement is a read observation. The existing history
             // recovery owner adopts the exact current Session/Turn and stream.
             const history = await conversationHistory.refresh();
             const current = currentChannelSession(history, selectedAgent.agentId);
             if (activeConversationContext.current === targetContextId && sessionIds.current.get(key) === sessionId
-                && current?.session_id === sessionId && current.active_turn_id === turnId) {
+                && current?.session_id === sessionId && current.active_turn_id
+                && (activeTurnIds.current.get(targetContextId) !== current.active_turn_id
+                  || streamControllers.current.get(targetContextId)?.signal.aborted !== false)) {
               setTurnRecoveryRequest(value => value + 1);
             }
           },

examples/personal-workspace-browser/attached-host-follow-up.mjs

@@ -231,6 +231,50 @@ export const attachedHostFollowUpScenario = {
       await recovery.getByRole("button", {name: "清除草稿", exact: true}).click();
       await recovery.waitFor({state: "hidden"});
       if (posts.length !== 2) throw new Error("Reading or clearing retained edits sent another Turn");
+      // Reconnect can miss the entire execution of a fast queued follow-up.
+      // Queue consumption must read its original answer even without a new
+      // observed active Turn or any live completion event.
+      await page.getByRole("navigation", {name: "Goal 视图"}).getByRole("button", {name: /^(Chat|对话)$/}).click();
+      await composer.fill("继续处理下一条，完成后在这里返回结果");
+      await sendButton.click();
+      await page.locator(".personal-conversation-queue li").filter({hasText: "继续处理下一条"}).waitFor();
+      await composer.fill("尚未发送的独立草稿");
+      const runtime = page.__loopxRuntime;
+      const reply = "快速完成的排队回复，无需重新打开会话。";
+      runtime.messages.get(attachedSessionId).push({message_id: "fast-queued-answer", turn_id: "turn-queued-3",
+        role: "agent", text: reply, created_at: "2026-08-13T01:00:04Z"});
+      runtime.sessions.set(attachedSessionId, {...runtime.sessions.get(attachedSessionId), active_turn_id: null, status: "ready"});
+      runtime.queuedTurns.set(attachedSessionId, []);
+      await page.getByText(reply, {exact: true}).waitFor({timeout: 15_000});
+      if (await page.getByText(reply, {exact: true}).count() !== 1 || posts.length !== 3
+          || await composer.inputValue() !== "尚未发送的独立草稿") {
+        throw new Error("Reading a consumed queue duplicated execution/replies or replaced unsent input");
+      }
+      // Reload retires optimistic placeholders and their SSE subscriptions.
+      // The queue still owns observation when its entire execution is missed.
+      await page.reload();
+      await page.getByText(reply, {exact: true}).waitFor();
+      await composer.fill("刷新后仍在原会话接收这条队列的结果");
+      await sendButton.click();
+      await page.locator(".personal-conversation-queue li").filter({hasText: "刷新后仍在原会话"}).waitFor();
+      await composer.fill("刷新后仍保留的未发送草稿");
+      await page.reload();
+      await page.locator(".personal-conversation-queue li").filter({hasText: "刷新后仍在原会话"}).waitFor();
+      const recoveredReply = "刷新后快速完成的原队列回复。";
+      // The queue can become idle before the native transcript write is
+      // visible. Its later revision must recover an answer with no remaining
+      // queue transition, optimistic row, SSE or user navigation.
+      runtime.queuedTurns.set(attachedSessionId, []);
+      await page.reload();
+      await page.getByText(reply, {exact: true}).waitFor();
+      await page.locator(".personal-conversation-queue").waitFor({state: "hidden"});
+      runtime.messages.get(attachedSessionId).push({message_id: "recovered-queued-answer", turn_id: "turn-queued-4",
+        role: "agent", text: recoveredReply, created_at: "2026-08-13T01:00:05Z"});
+      await page.getByText(recoveredReply, {exact: true}).waitFor({timeout: 15_000});
+      if (await page.getByText(recoveredReply, {exact: true}).count() !== 1 || posts.length !== 4
+          || await composer.inputValue() !== "刷新后仍保留的未发送草稿") {
+        throw new Error("Queue recovery requires a result readback without replay or draft loss");
+      }
     } finally {
       for (const route of heldStreams.splice(0)) await route.abort().catch(() => {});
       await context.close();