From 17b96b83dc420c4b50855102aa6a8809470a8848 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 16:42:22 +0800 Subject: [PATCH] fix(chat): allow cancelling delegation reads in place Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- apps/presentation/dashboard/package.json | 5 +- .../dashboard/smoke/delegation-read-cancel.ts | 172 ++++++++++++++++++ .../conversation-requests.tsx | 27 ++- .../steward_requests_fixture.py | 32 +++- 4 files changed, 224 insertions(+), 12 deletions(-) create mode 100644 apps/presentation/dashboard/smoke/delegation-read-cancel.ts diff --git a/apps/presentation/dashboard/package.json b/apps/presentation/dashboard/package.json index 6c8f2b23ee..1aeead3fd8 100644 --- a/apps/presentation/dashboard/package.json +++ b/apps/presentation/dashboard/package.json @@ -81,12 +81,13 @@ "smoke:recent-completions": "tsc --ignoreConfig --target ES2022 --module CommonJS --moduleResolution Node --ignoreDeprecations 6.0 --resolveJsonModule --esModuleInterop --jsx react-jsx --skipLibCheck --strict --types node --outDir /tmp/loopx-recent-completions-smoke smoke/recent-completions-smoke.ts && NODE_PATH=\"$PWD/node_modules\" node /tmp/loopx-recent-completions-smoke/apps/presentation/dashboard/smoke/recent-completions-smoke.js", "smoke:task-inspector-route": "npm run smoke:goal-work-browse && LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/task-inspector-route-smoke.mjs", "smoke:bound-conversation": "vite build --ssr smoke/bound-conversation-smoke.ts --outDir node_modules/.cache/loopx-bound-conversation --emptyOutDir && node node_modules/.cache/loopx-bound-conversation/bound-conversation-smoke.js", - "smoke:delegation-search": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/delegation-search-browser-smoke.mjs", + "smoke:delegation-search": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node smoke/delegation-search-browser-smoke.mjs && LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node --experimental-strip-types smoke/delegation-read-cancel.ts", "smoke:workspace-command-palette": "node --experimental-strip-types --test src/features/personal-workspace/workspace-command-entries.test.mjs && node --experimental-strip-types smoke/workspace-command-palette-browser-smoke.mjs && node --experimental-strip-types smoke/workspace-navigation-browser-smoke.mjs", "smoke:app-steward": "vite build --ssr smoke/app-steward-smoke.ts --outDir node_modules/.cache/loopx-app-steward --emptyOutDir && node node_modules/.cache/loopx-app-steward/app-steward-smoke.js", "smoke:conversation-input": "node --no-warnings --experimental-strip-types --test src/features/personal-workspace/steering-recovery.test.mjs && vite build --ssr smoke/conversation-input-smoke.ts --outDir node_modules/.cache/loopx-conversation-input --emptyOutDir && node node_modules/.cache/loopx-conversation-input/conversation-input-smoke.js", "smoke:answer-report": "vite build --ssr smoke/answer-report-smoke.tsx --outDir node_modules/.cache/loopx-answer-report --emptyOutDir && node node_modules/.cache/loopx-answer-report/answer-report-smoke.js", - "smoke:goal-work-browse": "vite build --ssr src/features/personal-workspace/goal-work-browse.test.mjs --outDir node_modules/.cache/loopx-work-browse --emptyOutDir && node node_modules/.cache/loopx-work-browse/goal-work-browse.test.js" + "smoke:goal-work-browse": "vite build --ssr src/features/personal-workspace/goal-work-browse.test.mjs --outDir node_modules/.cache/loopx-work-browse --emptyOutDir && node node_modules/.cache/loopx-work-browse/goal-work-browse.test.js", + "smoke:delegation-cancel": "LOOPX_PLAYWRIGHT_PACKAGE=\"$PWD/node_modules/playwright\" node --experimental-strip-types smoke/delegation-read-cancel.ts" }, "dependencies": { "@fontsource-variable/geist": "^5.3.0", diff --git a/apps/presentation/dashboard/smoke/delegation-read-cancel.ts b/apps/presentation/dashboard/smoke/delegation-read-cancel.ts new file mode 100644 index 0000000000..aa2328a77b --- /dev/null +++ b/apps/presentation/dashboard/smoke/delegation-read-cancel.ts @@ -0,0 +1,172 @@ +// Packaged reader over the real File/SQLite Chat/Inbox owners. Synthetic source +// data and a server-side gate only; no intercepted inventory or model execution. +import assert from "node:assert/strict"; +import {spawn} from "node:child_process"; +import {once} from "node:events"; +import {mkdir, writeFile} from "node:fs/promises"; +import {resolve} from "node:path"; +import {createInterface} from "node:readline"; +import type {Browser, Page} from "playwright"; +import {testPythonCommand} from "../../../../scripts/test-python.mjs"; +import {launchBrowser, loadPlaywright} from "../../../../examples/dashboard-browser-smoke-support.mjs"; + +const repo = resolve(import.meta.dirname, "../../../.."); +const output = resolve(repo, "output/playwright/delegation-cancel"); +await mkdir(output, {recursive: true}); +const results: unknown[] = []; +for (const storage of ["file", "sqlite"]) { + const path = resolve(repo, "examples/personal-workspace-browser/steward_requests_fixture.py"); + const command = testPythonCommand(`import runpy; runpy.run_path(${JSON.stringify(path)}, run_name="__main__")`, {repoRoot: repo}); + const child = spawn(command.command, [...command.args, "--status-file", resolve(repo, "examples/status.example.json"), + "--requests", "61", "--long-history", "--canonical-storage", storage, + ...(process.env.LOOPX_CANCEL_ASSETS ? ["--assets-dir", process.env.LOOPX_CANCEL_ASSETS] : [])], + {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: "0"}, stdio: ["pipe", "pipe", "pipe"]}); + const exited = once(child, "exit"); + let stderr = ""; + child.stderr.on("data", data => stderr += data); + type Event = {event: string; origin: string; posts: number; source_unchanged: boolean; chat_unchanged: boolean}; + const queue: Event[] = [], waiting: {kind: string; receive: (event: Event) => void}[] = []; + createInterface({input: child.stdout}).on("line", line => { + try { + const value: Event = JSON.parse(line), index = waiting.findIndex(item => item.kind === value.event); + if (index >= 0) waiting.splice(index, 1)[0]!.receive(value); else queue.push(value); + } + catch { /* Owner startup output is not a fixture event. */ } + }); + const next = (kind: string) => new Promise((accept, reject) => { + const timer = setTimeout(() => reject(Error(`Fixture timeout: ${stderr}`)), 90000); + const receive = (value: Event) => {clearTimeout(timer); accept(value);}; + const index = queue.findIndex(item => item.event === kind); + if (index >= 0) receive(queue.splice(index, 1)[0]!); else waiting.push({kind, receive}); + }); + const fixture = async (action: string) => { + child.stdin.write(action + "\n"); return next(action); + }; + let browser: Browser | undefined, page: Page | undefined; + try { + const ready = await next("ready"); + browser = await launchBrowser(loadPlaywright().chromium); + page = await browser!.newPage({viewport: {width: 1440, height: 1000}}); + await page.addInitScript(() => {if (!localStorage.getItem("loopx-pw-locale")) localStorage.setItem("loopx-pw-locale", "zh-CN");}); + const errors: string[] = []; + page.on("pageerror", error => errors.push(error.message)); + await page.goto(ready.origin + "/chat/"); + await page.getByRole("navigation", {name: "管家视图"}).getByRole("button", {name: /^(Chat|对话)$/}).click(); + await page.getByText("复核迁移说明中的示例", {exact: true}).filter({visible: true}).first().waitFor(); + const before = await fixture("inspect"); + const writes: string[] = []; + page.on("request", request => {if (request.method() !== "GET") writes.push(request.url());}); + const trigger = page.getByRole("button", {name: /委派，.*条待跟进/}); + const dialog = page.getByRole("dialog", {name: "委派", exact: true}); + const settled = () => page!.waitForFunction(() => document.querySelector(".conversation-requests-body")?.getAttribute("aria-busy") === "false"); + const rows = async (count: number) => { + await settled(); await page!.waitForFunction(expected => document.querySelectorAll(".conversation-request-row").length === expected, count); + }; + const cancel = dialog.getByRole("button", {name: "取消读取委派", exact: true}); + const refresh = dialog.getByRole("button", {name: "刷新委派", exact: true}); + const held = () => next("inventory-held"); + const release = async () => {await fixture("release"); await next("inventory-finished");}; + const status = dialog.locator(".conversation-requests-body > [role=status]"); + const snapshot = () => dialog.locator(".conversation-request-row > summary").allTextContents(); + const search = async (text: string) => { + await dialog.getByRole("searchbox", {name: "搜索委派"}).fill(text); + await dialog.getByRole("button", {name: "搜索", exact: true}).click(); + }; + await fixture("hold"); await trigger.click(); await held(); + if (process.env.LOOPX_CANCEL_BEFORE === "1") { + await page.screenshot({path: resolve(output, `before-${storage}-desktop.png`)}); + await page.setViewportSize({width: 390, height: 844}); + await page.screenshot({path: resolve(output, `before-${storage}-narrow.png`)}); + continue; + } + // No successful page exists yet. Cancellation is neither empty success nor + // a source error; the same dialog and keyboard focus survive it. + assert.equal(await cancel.count(), 1, "A stalled read needs an in-place cancellation control"); + await cancel.focus(); await cancel.press("Enter"); await settled(); + assert.equal(await dialog.isVisible(), true); + assert.equal(await status.innerText(), "已取消读取。"); + assert.equal(await dialog.locator(".conversation-request-row").count(), 0); + assert.equal(await dialog.getByRole("alert").count(), 0); + assert.equal(await refresh.evaluate(node => node === document.activeElement), true); + await fixture("continue"); await refresh.press("Enter"); await rows(12); + const firstPage = await snapshot(); + await release(); + assert.deepEqual(await snapshot(), firstPage); + + // Switching views keeps the published source visible. The cancelled + // requested filter remains directly retryable, even when it is unchanged. + await fixture("hold"); await dialog.getByRole("button", {name: "全部记录", exact: true}).click(); await held(); + await page.screenshot({path: resolve(output, `loading-${storage}-desktop.png`)}); + await cancel.click(); await settled(); assert.deepEqual(await snapshot(), firstPage); + assert.match(await status.innerText(), /显示上次记录/); + await fixture("continue"); await dialog.getByRole("button", {name: "全部记录", exact: true}).click(); await rows(12); + assert.equal(await dialog.getByRole("button", {name: "全部记录", exact: true}).getAttribute("aria-pressed"), "true"); + await release(); + + // A late cancelled search must never overwrite a newer query's real page. + await fixture("hold"); await search("核对账单导出的字段映射"); await held(); await cancel.click(); await settled(); + assert.equal(await dialog.locator(".conversation-request-row").count(), 0); + await fixture("continue"); await dialog.getByRole("button", {name: "清空", exact: true}).click(); await rows(12); + await release(); + assert.deepEqual(await snapshot(), firstPage); + + // More and a multi-window refresh retain all already published rows and + // an expanded original, rather than publishing a half-read snapshot. + const more = dialog.getByRole("button", {name: "加载下一页", exact: true}); + await fixture("hold-window"); await more.click(); await held(); await cancel.click(); await rows(12); + await fixture("continue"); await more.click(); await rows(24); await release(); + for (const count of [36, 48, 60, 61]) {await more.click(); await rows(count);} + const original = dialog.locator(".conversation-request-row").filter({has: page.getByText("核对账单导出的字段映射", {exact: true})}).first(); + await original.locator(":scope > summary").click(); + await original.getByText("原对话", {exact: true}).click(); + const articles = original.locator("article"); await articles.nth(1).waitFor(); + const body = "核对账单导出的字段映射\n\n" + "完整原始要求。Original evidence. ".repeat(600) + "完整正文结束。"; + assert.equal(await articles.first().locator(":scope > .personal-md").innerText(), body); + const allRows = await snapshot(); + await fixture("hold-window"); await refresh.click(); await held(); + await cancel.focus(); await cancel.press("Enter"); await rows(61); + assert.deepEqual(await snapshot(), allRows); + assert.equal(await articles.first().locator(":scope > .personal-md").innerText(), body); + assert.equal(await original.getByText("原对话", {exact: true}).evaluate(node => node.parentElement!.hasAttribute("open")), true); + await page.screenshot({path: resolve(output, `cancelled-${storage}-desktop.png`)}); + await page.setViewportSize({width: 390, height: 844}); + assert.ok(await dialog.evaluate(node => node.scrollWidth <= node.clientWidth + 1)); + await page.screenshot({path: resolve(output, `cancelled-${storage}-narrow.png`)}); + await fixture("continue"); await refresh.press("Enter"); await rows(61); await release(); + assert.equal(await status.count(), 0); + + // Source loss still withholds content. Cancellation must not turn that + // refusal into empty success or keep another source's previous records. + await fixture("local-source-unavailable"); await refresh.click(); await dialog.getByRole("alert").waitFor(); + assert.equal(await dialog.locator(".conversation-request-row").count(), 0); + await fixture("local-source-recover"); await dialog.getByRole("button", {name: "重试", exact: true}).click(); await rows(12); + const after = await fixture("inspect"); + assert.equal(after.source_unchanged, true); assert.equal(after.chat_unchanged, true); + assert.equal(after.posts, before.posts); assert.deepEqual(writes, []); assert.deepEqual(errors, []); + await page.keyboard.press("Escape"); + assert.equal(await trigger.evaluate(node => node === document.activeElement), true); + if (storage === "sqlite") { + await page.evaluate(() => localStorage.setItem("loopx-pw-locale", "en")); await page.reload(); + await page.getByRole("button", {name: /Delegations,.*awaiting follow-up/}).click(); await rows(12); + const english = page.getByRole("dialog", {name: "Delegations", exact: true}); + await fixture("hold"); await english.getByRole("button", {name: "All records", exact: true}).click(); await held(); + await page.screenshot({path: resolve(output, "loading-en-narrow.png")}); + await english.getByRole("button", {name: "Cancel delegation read", exact: true}).press("Enter"); await rows(12); + assert.match(await english.locator(".conversation-requests-body > [role=status]").innerText(), /Read cancelled.*previous records/); + await page.screenshot({path: resolve(output, "cancelled-en-narrow.png")}); + assert.ok(await english.evaluate(node => node.scrollWidth <= node.clientWidth + 1)); + await fixture("continue"); await english.getByRole("button", {name: "All records", exact: true}).click(); await rows(12); await release(); + assert.equal(await english.getByRole("button", {name: "All records", exact: true}).getAttribute("aria-pressed"), "true"); + await page.keyboard.press("Escape"); + } + results.push({storage, requests: 61, originalBodyChars: [...body].length, sourceUnchanged: true, + initial: true, filterRetry: true, staleSearch: true, more: true, multiWindow: true, sourceRecovery: true, writes: 0}); + } finally { + // Release before shutdown: cancelled clients do not stop server work. + child.stdin.write("release\nstop\n"); + await browser?.close(); + const [code] = await exited; assert.equal(code, 0, stderr); + } +} +await writeFile(resolve(output, "result.json"), JSON.stringify({results}, null, 2)); +console.log(JSON.stringify({ok: true, results})); diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx index 36ca5afb60..611ccdaffb 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -236,6 +236,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const [history, setHistory] = useState(false); const [pages, setPages] = useState([]); const [busy, setBusy] = useState(false); + const [cancelled, setCancelled] = useState(false); const [error, setError] = useState(false); const [noSession, setNoSession] = useState(false); const [revision, setRevision] = useState(0); @@ -259,7 +260,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se && (sessionId ? loadedSource.current === sessionId : ownerFallback) && loadedScope.current === otherViewScope; useEffect(() => { - setOpen(false); setHistory(false); setPages([]); setError(false); setSearch(""); setSearchDraft(""); + setOpen(false); setHistory(false); setPages([]); setError(false); setCancelled(false); setBusy(false); setSearch(""); setSearchDraft(""); loadedScope.current = null; loadedSource.current = null; requestedRows.current = 12; }, [sessionId, ownerFallback]); @@ -281,7 +282,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const current = ++generation.current; requestAbort.current?.abort(); const abort = new AbortController(); requestAbort.current = abort; - setBusy(true); setError(false); setNoSession(false); setConversationChanged(false); + setBusy(true); setCancelled(false); setError(false); setNoSession(false); setConversationChanged(false); const sameScope = loadedScope.current === scope; const loadedRows = pages.reduce((count, page) => count + page.rows.length, 0); // A fresh snapshot or parent refresh must retain the page requested by @@ -349,19 +350,26 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const recipients = [...new Set(rows.map(row => `${row.goal_id}\n${row.agent_id}`))]; const count = error || !page ? "—" : unknown ? (page.pending_count ? `${page.pending_count}+` : "—") : page.pending_count; const title = zh ? "委派" : "Delegations"; + const cancelRead = () => { + // Abort only this reader. The server may still finish its read; its late + // response must not replace the last published view or a subsequent retry. + generation.current++; requestAbort.current?.abort(); requestAbort.current = null; + setBusy(false); setCancelled(true); + }; const changeHistory = (value: boolean) => { - if (value === history) return; + if (value === history && !cancelled) return; generation.current++; requestAbort.current?.abort(); - setBusy(true); setError(false); setHistory(value); + setBusy(true); setCancelled(false); setError(false); setHistory(value); + setRevision(value => value + 1); }; const applySearch = (value: string) => { // Fence immediately, including the interval before the effect starts. generation.current++; requestAbort.current?.abort(); requestedRows.current = 12; - setPages([]); setError(false); setBusy(true); setSearch(value.trim()); + setPages([]); setError(false); setCancelled(false); setBusy(true); setSearch(value.trim()); setRevision(value => value + 1); }; - const close = () => {generation.current++; requestAbort.current?.abort(); requestedRows.current = 12; setBusy(false); setOpen(false); trigger.current?.focus();}; + const close = () => {generation.current++; requestAbort.current?.abort(); requestedRows.current = 12; setBusy(false); setCancelled(false); setOpen(false); trigger.current?.focus();}; return
{conversationChanged ? {zh ? "对话有更新" : "Conversation updated"} : null} - +
{event.preventDefault(); applySearch(searchDraft);}}> {setSearchDraft(""); applySearch("");}}>{zh ? "清空" : "Clear"} : null}
+ {cancelled ?

{rows.length + ? (zh ? "已取消读取，显示上次记录。" : "Read cancelled. Showing previous records.") + : (zh ? "已取消读取。" : "Read cancelled.")}

: null} {error ?

{zh ? "委派记录暂时无法核验。" : "Delegation records cannot be verified."}

: null} {busy && !rows.length ? <>

{zh ? "正在读取委派…" : "Loading delegations…"}

: null} diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index 179107d860..13085b6a21 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/examples/personal-workspace-browser/steward_requests_fixture.py @@ -6,7 +6,7 @@ Commands on stdin: inspect, fault, fault-window, deliver-window, report-window, recover, deliver, source-unavailable, source-recover, local-source-unavailable, local-source-recover, history-corrupt/history-recover (--long-history only), -report-partial, complete-partial, stop. Window faults/returns apply +report-partial, complete-partial, hold, hold-window, continue, release, stop. Window faults/returns apply after the first page to challenge multi-window refresh readback. """ from __future__ import annotations @@ -148,12 +148,24 @@ def _goal_contexts(self): def _request_inventory_read(self): self.server.inventory_reads += 1 later_window = parse_qs(urlparse(self.path).query).get("offset", ["0"])[0] != "0" + held = self.server.inventory_hold and (not self.server.hold_window or later_window) + if held: + print(json.dumps({"event": "inventory-held", "later_window": later_window}), flush=True) + if not self.server.inventory_gate.wait(60): + return self._send_error("Synthetic read gate was not released", status=503) if self.server.fault or (later_window and self.server.fault_window): return self._send_error("Synthetic inventory read failure", status=503) if later_window and self.server.deliver_window: self.server.deliver_window = False drain(runtime, registry, store, lambda *_: None) - return super()._request_inventory_read() + try: + return super()._request_inventory_read() + except (BrokenPipeError, ConnectionResetError): + if not held: + raise + finally: + if held: + print(json.dumps({"event": "inventory-finished"}), flush=True) def do_GET(self): if urlparse(self.path).path == "/status.json": query = parse_qs(urlparse(self.path).query) @@ -183,6 +195,9 @@ def do_GET(self): server.runtime_root_override, server.selected_goal_id, server.verbose = str(runtime), None, False server.assets_dir = selected.assets_dir or default_chat_assets_dir() server.fault, server.posts, server.inventory_reads = False, 0, 0 + server.inventory_hold, server.hold_window = False, False + server.inventory_gate = threading.Event() + server.inventory_gate.set() server.fault_window, server.deliver_window = False, False server.runtime_controller = ChatRuntimeController(store=store, registry_path=registry, codex_bin="unused-fixture-codex") server.runtime_controller.capabilities = lambda: [{"agent_id": "codex", "display_name": "Codex", @@ -207,7 +222,17 @@ def source_hashes(): command = line.strip() if command == "stop": break - if command == "fault": + if command in {"hold", "hold-window"}: + server.inventory_gate.clear() + server.inventory_hold, server.hold_window = True, command == "hold-window" + elif command == "continue": + # Admit new reads while the cancelled HTTP read remains + # held, so its real late response can challenge a retry. + server.inventory_hold = False + elif command == "release": + server.inventory_hold = False + server.inventory_gate.set() + elif command == "fault": server.fault = True elif command == "fault-window": server.fault_window = True @@ -264,6 +289,7 @@ def source_hashes(): "chat_unchanged": journal.read_bytes() == original_journal, "turn_count": len(list(store.sessions_root.glob("*/turns/*.json")))}), flush=True) finally: + server.inventory_gate.set() server.shutdown() server.server_close() thread.join(5)