diff --git a/apps/presentation/dashboard/package.json b/apps/presentation/dashboard/package.json index 1aeead3fd8..a4dd4cdcec 100644 --- a/apps/presentation/dashboard/package.json +++ b/apps/presentation/dashboard/package.json @@ -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", diff --git a/apps/presentation/dashboard/smoke/queue-input-recovery-smoke.mjs b/apps/presentation/dashboard/smoke/queue-input-recovery-smoke.mjs new file mode 100644 index 0000000000..bc7cc657c1 --- /dev/null +++ b/apps/presentation/dashboard/smoke/queue-input-recovery-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"); diff --git a/apps/presentation/dashboard/src/data/conversation-returns.test.mjs b/apps/presentation/dashboard/src/data/conversation-returns.test.mjs index 0cebf54a37..db2c63cc08 100644 --- a/apps/presentation/dashboard/src/data/conversation-returns.test.mjs +++ b/apps/presentation/dashboard/src/data/conversation-returns.test.mjs @@ -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}]}, diff --git a/apps/presentation/dashboard/src/data/conversation-returns.ts b/apps/presentation/dashboard/src/data/conversation-returns.ts index 20b4e4c5fc..6404e98339 100644 --- a/apps/presentation/dashboard/src/data/conversation-returns.ts +++ b/apps/presentation/dashboard/src/data/conversation-returns.ts @@ -112,10 +112,15 @@ export function reconcileConversationHistory( } // 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(); 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; diff --git a/apps/presentation/dashboard/src/data/use-conversation-queue.ts b/apps/presentation/dashboard/src/data/use-conversation-queue.ts index 76de2df8e1..a869d42ff1 100644 --- a/apps/presentation/dashboard/src/data/use-conversation-queue.ts +++ b/apps/presentation/dashboard/src/data/use-conversation-queue.ts @@ -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)}; } 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 daabb32619..273217988a 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css @@ -22,3 +22,9 @@ .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; } +.personal-queue-recovery > details { border: 1px solid var(--pw-line-strong); border-radius: 8px; padding: 8px 10px; } +.personal-queue-recovery summary { cursor: pointer; } +.personal-queue-retained-input { margin-top: 8px; } +.personal-queue-retained-input p { margin: 6px 0; color: var(--pw-muted); } +.personal-queue-retained-input textarea { box-sizing: border-box; width: 100%; resize: vertical; padding: 8px; font: inherit; font-size: var(--text-body); line-height: var(--leading-body); color: var(--pw-text); background: var(--pw-surface, #fff); border: 1px solid var(--pw-line-strong); border-radius: 6px; } +.personal-queue-retained-actions { display: flex; flex-wrap: wrap; gap: 6px; } 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 b734e12b05..939a9fa733 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx @@ -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(recovery.request); @@ -61,8 +58,7 @@ 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 || @@ -70,7 +66,7 @@ function QueuedMessage({sessionId, turn, index, targetTurnId, onChange, onInputC "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 ; } -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
+ return <>{turns.length || error ?
{zh ? "待处理" : "Queued"}{error ? "" : ` · ${turns.length}`} {error ? <>{zh ? "读取失败" : "Read failed"}

{error.message}

 : null}
    {turns.map((turn, index) => )}
-
; +
: null} + {readReady && !error ? : null} + ; } 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 e76c1c78ce..b390d1071f 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."}

: null} {conversationSessionId ? {conversationQueue.retry(); conversationHistoryState?.retry();}} targetTurnId={steeringTurnId} turns={conversationQueue.turns} error={conversationQueue.error} onRetry={conversationQueue.retry}/> : null}
(); + 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}; +} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/retained-queue-inputs.tsx b/apps/presentation/dashboard/src/features/personal-workspace/retained-queue-inputs.tsx new file mode 100644 index 0000000000..6df45b7070 --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/retained-queue-inputs.tsx @@ -0,0 +1,72 @@ +import { useEffect, useRef, useState } from "react"; +import { ChatApiError, mutateQueuedChatTurn, type ChatQueuedTurn } from "../../data/chat"; +import { listQueueInputRecovery, queueInputKeys, retireQueueInputRequest, type QueueInputRecovery } from "./queue-input-recovery"; +import { useWorkspaceI18n } from "./i18n"; + +function readSessionRecovery(sessionId: string) { + try { + return typeof window === "undefined" ? {entries: [], failed: false} : listQueueInputRecovery(window.localStorage, sessionId); + } catch {return {entries: [], failed: true};} +} + +function RetainedInput({sessionId, input, onChange}: {sessionId: string; input: QueueInputRecovery; onChange: () => void}) { + const zh = useWorkspaceI18n().locale === "zh-CN"; + const [error, setError] = useState(""); + const [busy, setBusy] = useState(false); + const lock = useRef(false), text = useRef(null); + const draft = input.editor?.message ?? (input.request?.action === "edit" ? input.request.message : null); + async function check() { + if (!input.request || lock.current) return; + lock.current = true; setBusy(true); setError(""); + try { + // Retry only the retained operation. The owner can replay its receipt + // after dispatch; a new edit against an already started Turn is rejected. + await mutateQueuedChatTurn(sessionId, input.turnId, input.request); + retireQueueInputRequest(localStorage, sessionId, input.turnId, input.request, true); + } catch (cause) { + const code = cause instanceof ChatApiError ? cause.payload.error_code : ""; + if (["queue_turn_already_started", "queue_input_revision_conflict", "queue_mutation_identity_conflict"].includes(String(code))) { + // Keep the full text, including an edit retained only in the request. + try {retireQueueInputRequest(localStorage, sessionId, input.turnId, input.request, false);} + catch { /* An unreadable record must not be replaced with a fresh send. */ } + setError(zh ? "修改未应用，草稿已保留。" : "The change was not applied. Your draft is retained."); + } else setError(zh ? "原操作尚未确认，恢复记录已保留。" : "The original operation is unconfirmed. Recovery is retained."); + } finally {lock.current = false; setBusy(false); onChange();} + } + function discard() { + try {localStorage.removeItem(queueInputKeys(sessionId, input.turnId).draft); onChange();} + catch {setError(zh ? "草稿无法清除，请保留文本后重试。" : "The draft could not be cleared. Retain the text and retry.");} + } + return
+ {draft !== null ? (zh ? "保留的编辑草稿" : "Retained edit draft") : (zh ? "未确认的队列修改" : "Unconfirmed queue change")} +

{zh ? "原消息已不在待处理队列中。" : "The original message is no longer in the pending queue."}

+ {draft !== null ?