From a4093afc262d15fa04dacc0e3299c65a9223c8b7 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 02:19:44 +0800 Subject: [PATCH 1/3] fix(chat): recover replies completed between queue observations Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../src/data/conversation-returns.test.mjs | 22 +++++++++++ .../src/data/conversation-returns.ts | 10 +++-- .../personal-workspace-model.ts | 2 +- .../personal-workspace-page.tsx | 25 +++++++----- .../dashboard/src/views/dashboard-page.tsx | 10 ++--- .../attached-host-follow-up.mjs | 38 +++++++++++++++++++ 6 files changed, 89 insertions(+), 18 deletions(-) diff --git a/apps/presentation/dashboard/src/data/conversation-returns.test.mjs b/apps/presentation/dashboard/src/data/conversation-returns.test.mjs index db2c63cc08..4243c8f10f 100644 --- a/apps/presentation/dashboard/src/data/conversation-returns.test.mjs +++ b/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}]}, diff --git a/apps/presentation/dashboard/src/data/conversation-returns.ts b/apps/presentation/dashboard/src/data/conversation-returns.ts index 6404e98339..904156b0b4 100644 --- a/apps/presentation/dashboard/src/data/conversation-returns.ts +++ b/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( && 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; 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 7aaf8187eb..b1d20a29f3 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 @@ -581,7 +581,7 @@ export type PersonalWorkspaceCallbacks = { onInterruptConversationTurn?: (contextId: string, turnId: string) => Promise; onSteerConversationTurn?: (contextId: string, turnId: string, message: string, ingressId: string) => Promise; onQueueConversationTurn?: (contextId: string, sessionId: string, message: string, ingressId: string, attachments?: WorkspaceImageAttachment[]) => Promise; - onObserveConversationTurn?: (contextId: string, sessionId: string, turnId: string) => Promise; + onObserveConversationQueue?: (contextId: string, sessionId: string) => Promise; onOpenGoal?: (goalId: string) => void | Promise; onOpenGoalView?: (tab: WorkspaceGoalTab) => void; onOpenRunSession?: (run: WorkspaceRun) => void | Promise; 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 a396966098..a5862c0ba6 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 @@ -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; diff --git a/apps/presentation/dashboard/src/views/dashboard-page.tsx b/apps/presentation/dashboard/src/views/dashboard-page.tsx index 35dfe74069..b3437a827c 100644 --- a/apps/presentation/dashboard/src/views/dashboard-page.tsx +++ b/apps/presentation/dashboard/src/views/dashboard-page.tsx @@ -3347,17 +3347,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); } }, diff --git a/examples/personal-workspace-browser/attached-host-follow-up.mjs b/examples/personal-workspace-browser/attached-host-follow-up.mjs index a0c4ebc083..9b3987dfb7 100644 --- a/examples/personal-workspace-browser/attached-host-follow-up.mjs +++ b/examples/personal-workspace-browser/attached-host-follow-up.mjs @@ -231,6 +231,44 @@ 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 = "刷新后快速完成的原队列回复。"; + 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"}); + runtime.queuedTurns.set(attachedSessionId, []); + 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(); From d48a3145565f2fed559177a03ce2345db14dbe54 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 05:09:29 +0800 Subject: [PATCH 2/3] test: retain committed side effect in lost-response fixture Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/control_plane/test_goal_create_completion.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/control_plane/test_goal_create_completion.py b/tests/control_plane/test_goal_create_completion.py index c03bb20600..18a0af6e7a 100644 --- a/tests/control_plane/test_goal_create_completion.py +++ b/tests/control_plane/test_goal_create_completion.py @@ -181,7 +181,7 @@ def test_managed_creation_retains_requirements_before_one_host_dispatch(environm original = getattr(target, method) def lose_response(*args, **kwargs): - result = original(*args, **kwargs) + original(*args, **kwargs) raise TimeoutError("Committed creation response was lost") monkeypatch.setattr(target, method, lose_response) From be9bec80fe89dfd76035ded1945250d8a741305b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 05:33:07 +0800 Subject: [PATCH 3/3] fix(chat): adopt native replies committed after queue drain Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../dashboard/src/views/dashboard-page.tsx | 55 ++++++++++--------- .../attached-host-follow-up.mjs | 8 ++- 2 files changed, 36 insertions(+), 27 deletions(-) diff --git a/apps/presentation/dashboard/src/views/dashboard-page.tsx b/apps/presentation/dashboard/src/views/dashboard-page.tsx index b3437a827c..f635802772 100644 --- a/apps/presentation/dashboard/src/views/dashboard-page.tsx +++ b/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; diff --git a/examples/personal-workspace-browser/attached-host-follow-up.mjs b/examples/personal-workspace-browser/attached-host-follow-up.mjs index 9b3987dfb7..2f70547ce1 100644 --- a/examples/personal-workspace-browser/attached-host-follow-up.mjs +++ b/examples/personal-workspace-browser/attached-host-follow-up.mjs @@ -261,9 +261,15 @@ export const attachedHostFollowUpScenario = { 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"}); - runtime.queuedTurns.set(attachedSessionId, []); 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() !== "刷新后仍保留的未发送草稿") {