From 3a2ebe592cd7974d558789b6ec0c8ac4dfef710a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 14:04:44 +0800 Subject: [PATCH 1/6] perf(collaboration): reuse lifetime fences in bounded inventory batches Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/manager_context/README.md | 9 ++ .../collaboration/conversation_inventory.py | 87 ++++++++++++----- .../test_chat_request_inventory.py | 93 +++++++++++++++++++ 3 files changed, 165 insertions(+), 24 deletions(-) diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index cfebf883eb..fa46d88e59 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -182,10 +182,19 @@ Goal visibility are checked before and after reading. Original-Chat delivery and explicit peer-return consumption remain different receipt paths. The list does not mark reads, accept work, start an Agent or certify an overall outcome. Unreadable/missing sources and incomplete scans keep counts visibly partial. +Within one inventory read, requests sharing an exact Goal lifetime reuse its +held guard in batches of at most one page. Every request retains its own lock, +source/Turn read, grant and typed receipt/lifecycle checks; each new batch +observes the current Goal lifetime. Core linked-work IO runs after those guards +are released. This reduces repeated fence IO without caching authorization +between requests or pages; pagination and search still scan the bounded history. 管家标题栏提供常驻委派入口，按接收方查看待跟进请求，并可展开当前交办内容、关联 工作和原对话。同一来源的较早会话不会因重开会话而丢失；其它用户、来源、群聊和 项目会话不混入。结论已回传与成果已验收不是同一事实，历史记录仍可回看。 +同一次读取中，相同 Goal 实例按至多一页大小复用持有中的生命周期锁；逐请求锁、 +来源与 Turn 读回、授权及 typed 回执检查保留，每批重新观察当前实例。关联 Core 工作 +在释放这些锁后读取，不跨请求或分页缓存授权；分页与搜索仍扫描有界历史。 The automatic agenda is a partial read (`read_complete=false`): it cannot fulfill the full Inbox read or replace its requirements. Before acting, read the current diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index eab30c355d..6c48c213b1 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/loopx/control_plane/collaboration/conversation_inventory.py @@ -46,6 +46,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core query = search.strip().casefold() paths = sorted((_root(root) / "entries").glob("*/*.json")) rows, unreadable, cache, turn_readers = [], 0, {}, {} + groups, work_reads = {}, [] for path in paths[:2000]: try: candidate = _read(path) @@ -67,29 +68,67 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core gid, aid = route["goal_id"], route["agent_id"] if goal_ids is not None and gid not in goal_ids: continue - sid = source_session["session_id"] - if sid not in turn_readers: - turn_readers[sid] = original_turn_reader(store, sid) - turn = turn_readers[sid](route["client_turn_id"]) - if turn is None: - raise ValueError("original source Turn unavailable") - with collaboration_goal_scope(registry, goal_id=gid, agents=(), caller_goal_ref=route.get("goal_ref")) as scope: - with _request_lock(root, candidate["request_id"], scope, path.with_suffix(".lock")): - row = _entry(root, gid, aid, candidate["request_id"], scope=scope) - if row.get("source_kind") == "peer": - continue - grant = source_context_target_authority(root, source_session, turn, {"goal_id": gid, "agent_id": aid}) - decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route, - initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"])) - observed = read_inbox_transaction(root, {"operation": "receipt_states", "rows": [row], "include_agenda": True}) - state, agenda = observed["items"][0], observed["agenda_items"][0] - controlled = state.get("readback") - read, read_error = _receipt(root, "reads", row) - decision, decision_error = _receipt(root, "decisions", row) - replies = reply_status(root, row) - # Request/return observations are version-consistent. Core work is - # a separate current observation; never hold its request lock over - # another authority's IO. + if not isinstance(gid, str) or not isinstance(aid, str): + raise ValueError("invalid request target") + key = (gid, json.dumps(route.get("goal_ref"), sort_keys=True)) + groups.setdefault(key, []).append((path, route, candidate["request_id"])) + except (OSError, ValueError, KeyError, TypeError): + unreadable += 1 + # Reuse only the held lifetime guard, never an authorization or request + # snapshot. Resolve sources outside it first: their provider checks can + # read another authority. Each source/Turn and request is reread below. + # Keep each guard bounded to a page-sized batch so a long inventory does + # not exclude Goal writers for the entire history scan. + batches = [(key, items[start:start + limit]) for key, items in groups.items() + for start in range(0, len(items), limit)] + for (gid, _), requests in batches: + start, failed = len(work_reads), 0 + try: + with collaboration_goal_scope(registry, goal_id=gid, agents=(), + caller_goal_ref=requests[0][1].get("goal_ref")) as scope: + for path, route, request_id in requests: + try: + source_session = store.load_session(route["session_id"]) + if not source_session or source_session.get("channel_id") != session["channel_id"]: + raise ValueError("original source Session unavailable") + validate_source_session(source_session) + sid = source_session["session_id"] + if sid not in turn_readers: + turn_readers[sid] = original_turn_reader(store, sid) + turn = turn_readers[sid](route["client_turn_id"]) + if turn is None: + raise ValueError("original source Turn unavailable") + with _request_lock(root, request_id, scope, path.with_suffix(".lock")): + row = _entry(root, gid, route["agent_id"], request_id, scope=scope) + if row.get("source_kind") == "peer": + continue + grant = source_context_target_authority(root, source_session, turn, + {"goal_id": gid, "agent_id": route["agent_id"]}) + decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route, + initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"])) + observed = read_inbox_transaction(root, {"operation": "receipt_states", "rows": [row], "include_agenda": True}) + state, agenda = observed["items"][0], observed["agenda_items"][0] + controlled = state.get("readback") + read, read_error = _receipt(root, "reads", row) + decision, decision_error = _receipt(root, "decisions", row) + replies = reply_status(root, row) + # Do not retain full model responses while waiting for + # linked-work IO; only the displayed question is used. + source_turn = {"turn_id": turn["turn_id"], "message": turn.get("message", "")} + work_reads.append((row, source_session["session_id"], source_turn, state, agenda, controlled, + read, read_error, decision, decision_error, replies)) + except (OSError, ValueError, KeyError, TypeError): + failed += 1 + except (OSError, ValueError, KeyError, TypeError): + del work_reads[start:] + unreadable += len(requests) + else: + unreadable += failed + # Request/return observations are version-consistent. Core work is a + # separate observation: release all lifetime/request guards before its IO. + for row, source_session_id, turn, state, agenda, controlled, read, read_error, decision, decision_error, replies in work_reads: + try: + gid, aid = row["goal_id"], row["agent_id"] work = read_linked_work(root, registry, row, cache, include_core_details=core_goal_ids is None or gid in core_goal_ids) warnings = list(state.get("warnings", [])) + work["warnings"] @@ -113,7 +152,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core "unavailable" if read_error or control_unavailable else "supplied" if read else "pending", "decision": assessment["status"], "decision_reason": assessment.get("reason"), "returns": replies, "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"], - "warnings": warnings, "source_session_id": source_session["session_id"], "source_turn_id": turn["turn_id"], + "warnings": warnings, "source_session_id": source_session_id, "source_turn_id": turn["turn_id"], "last_event_at": _last_event([row.get("delivered_at"), read_at, assessment.get("at"), *(reply.get(key) for reply in replies for key in ("created_at", "delivered_at"))])}) except (OSError, ValueError, KeyError, TypeError): diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index 2dfa337a62..57a31863c5 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -227,3 +227,96 @@ def observe_scope(reader, session_id): code, denied = _inventory(api, route["session_id"]) assert code == 400 and denied["error_code"] == "request_inventory_unavailable" assert "rows" not in denied and "Check the current contract" not in str(denied) + + +def test_inventory_bounds_lifetime_guards_and_keeps_core_io_outside_them(api, monkeypatch): + from contextlib import contextmanager + from loopx.control_plane.collaboration import conversation_inventory as inventory + + _, _, route = _source(api) + _, f, server = api + session = server.chat_store.load_session(route["session_id"]) + for number in range(12): + turn, _ = server.chat_store.create_turn(session["session_id"], client_turn_id=f"batch-{number}", + message=f"Check artifact {number}", origin="web") + delivered = deliver(f["runtime"], f["registry"], session=session, turn=turn, + request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-a"}, source_store=server.chat_store) + server.chat_store.update_turn(session["session_id"], turn["turn_id"], status="completed", + response={"context_handoff_receipt": delivered}) + server.chat_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) + original_scope, original_work = inventory.collaboration_goal_scope, inventory.read_linked_work + held, batches = [], [] + + @contextmanager + def scope(*args, **kwargs): + with original_scope(*args, **kwargs) as current: + held.append(current) + batches.append(0) + try: + yield current + finally: + held.pop() + + original_entry = inventory._entry + def entry(*args, **kwargs): + assert held, "A request must retain the Goal lifetime guard" + batches[-1] += 1 + return original_entry(*args, **kwargs) + + def work(*args, **kwargs): + assert not held, "Core IO must not nest under a Goal lifetime guard" + return original_work(*args, **kwargs) + + monkeypatch.setattr(inventory, "collaboration_goal_scope", scope) + monkeypatch.setattr(inventory, "_entry", entry) + monkeypatch.setattr(inventory, "read_linked_work", work) + code, value = _inventory(api, route["session_id"], view="history") + assert code == 200 and value["matched"] == value["pending_count"] == 13 + assert value["coverage"] == {"scan_complete": True, "unreadable": 0} + assert sum(batches) == 13 and len(batches) == 2 and max(batches) <= 12 + + +def test_inventory_rereads_goal_lifetime_between_batches(api, monkeypatch): + from contextlib import contextmanager + from loopx.control_plane.collaboration import conversation_inventory as inventory + from loopx.control_plane.goals.source_session_recreation import RecreateGoalRequest, recreate_goal_instance + + _, _, route = _source(api) + _, f, server = api + session = server.chat_store.load_session(route["session_id"]) + for number in range(12): + turn, _ = server.chat_store.create_turn(session["session_id"], client_turn_id=f"replace-{number}", + message=f"Check artifact {number}", origin="web") + delivered = deliver(f["runtime"], f["registry"], session=session, turn=turn, + request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-a"}, source_store=server.chat_store) + server.chat_store.update_turn(session["session_id"], turn["turn_id"], status="completed", + response={"context_handoff_receipt": delivered}) + server.chat_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) + original_scope = inventory.collaboration_goal_scope + original_decide = inventory.decide_collaboration_lifecycle + calls, modes = [], [] + + @contextmanager + def scope(*args, **kwargs): + if calls: + replaced = recreate_goal_instance(RecreateGoalRequest(registry_path=f["registry"], + goal_id=f["ref"]["goal_id"], goal_instance_id=f["ref"]["goal_instance_id"], operation_id="replace-during-read")) + assert replaced["ok"] + with original_scope(*args, **kwargs) as current: + calls.append(current.current_goal_ref) + yield current + + def decide(*args, **kwargs): + result = original_decide(*args, **kwargs) + modes.append(result.get("mode")) + return result + + monkeypatch.setattr(inventory, "collaboration_goal_scope", scope) + monkeypatch.setattr(inventory, "decide_collaboration_lifecycle", decide) + code, value = _inventory(api, route["session_id"], view="history") + assert code == 200 and len(calls) == 2 + assert calls[0] == f["ref"] and calls[1] != f["ref"] + # The existing typed owner permits proven historical inspection after a + # replacement; it must observe the new lifetime rather than reuse the old. + assert modes == ["current_instance"] * 12 + ["historical_read"] + assert value["matched"] == 13 and value["coverage"] == {"scan_complete": True, "unreadable": 0} From 1e454b3bd43f07a8a980d193ff301c7c48e8d841 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 19:14:58 +0800 Subject: [PATCH 2/6] fix: batch loaded delegation pages during refresh Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../src/data/conversation-requests.ts | 3 +- .../conversation-requests.tsx | 13 ++-- .../steward_requests_fixture.py | 31 +++++++-- loopx/capabilities/manager_context/README.md | 6 ++ .../collaboration/conversation_inventory.py | 4 +- loopx/presentation/request_control_api.py | 4 +- .../test_chat_request_inventory.py | 69 ++++++++++++++++++- 7 files changed, 113 insertions(+), 17 deletions(-) diff --git a/apps/presentation/dashboard/src/data/conversation-requests.ts b/apps/presentation/dashboard/src/data/conversation-requests.ts index 2c58144b16..faac0c7dac 100644 --- a/apps/presentation/dashboard/src/data/conversation-requests.ts +++ b/apps/presentation/dashboard/src/data/conversation-requests.ts @@ -21,8 +21,9 @@ const inventorySchema = z.object({ok: z.literal(true), session_id: z.string(), r export type ConversationRequest = z.infer; export type ConversationRequestInventory = z.infer; -export async function fetchConversationRequests(sessionId: string, history: boolean, offset = 0, signal?: AbortSignal, search = "") { +export async function fetchConversationRequests(sessionId: string, history: boolean, offset = 0, signal?: AbortSignal, search = "", limit = 12) { const query = new URLSearchParams({session_id: sessionId, view: history ? "history" : "unfinished", offset: String(offset), search}); + if (limit !== 12) query.set("limit", String(limit)); const result = inventorySchema.parse(await requestJson(`/api/chat/request-inventory?${query}`, {signal})); if (result.session_id !== sessionId) throw new Error("Conversation request scope changed."); return result; 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 5e15d62ffa..65a50a995e 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -231,7 +231,8 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se setBusy(true); setError(false); setNoSession(false); const scope = JSON.stringify([sessionId, ownerFallback, history, search]); const sameScope = loadedScope.current === scope; - const pageCount = sameScope ? Math.max(1, pages.length) : 1; + const loadedRows = pages.reduce((count, page) => count + page.rows.length, 0); + const refreshRows = sameScope ? Math.max(12, Math.ceil(loadedRows / 12) * 12) : 12; // Keep verified rows during a same-source refresh, but never across scope // changes or a failed read. Publish refreshed pages together below. if (!sameScope) {setPages([]); loadedSource.current = null;} @@ -249,14 +250,16 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se loadedSource.current = source; const refreshed: ConversationRequestInventory[] = []; let cursor: number | null = offset; - const limit = offset ? 1 : sameSource ? pageCount : 1; - for (let index = 0; index < limit && cursor !== null; index++) { - const value = await fetchConversationRequests(source, history, cursor, abort.signal, search); + const targetRows = offset ? 12 : sameSource ? refreshRows : 12; + // Each HTTP read verifies the whole inventory. Restore the loaded extent + // in bounded windows; opening and More still read one normal page. + for (let remaining = targetRows; remaining > 0 && cursor !== null; remaining -= 48) { + const value = await fetchConversationRequests(source, history, cursor, abort.signal, search, Math.min(48, remaining)); if (current !== generation.current) return; // Returns can reorder a live inventory. A stale More cursor gets one // fresh read; mixed snapshots during that read fail visibly for retry. if (offset && value.snapshot_digest !== pages[0]?.snapshot_digest) {void load(); return;} - if (index && value.snapshot_digest !== refreshed[0].snapshot_digest) throw new Error("Request snapshot changed."); + if (refreshed.length && value.snapshot_digest !== refreshed[0].snapshot_digest) throw new Error("Request snapshot changed."); if (value.next_offset !== null && value.next_offset <= cursor) throw new Error("Request paging did not advance."); refreshed.push(value); cursor = value.next_offset; } diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index 7b39fe9b01..7d4535411c 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/examples/personal-workspace-browser/steward_requests_fixture.py @@ -3,8 +3,9 @@ Pass a public synthetic status file for whole-workspace visual review. Runtime availability is display-only fixture metadata; POSTs cannot execute it. Requests, decisions and return delivery below use the production owners in a temp runtime. -Commands on stdin: inspect, fault, recover, deliver, source-unavailable, -source-recover, stop. +Commands on stdin: inspect, fault, fault-window, deliver-window, recover, +deliver, source-unavailable, source-recover, stop. Window faults/returns apply +after the first page to challenge multi-window refresh readback. """ from __future__ import annotations @@ -36,7 +37,10 @@ def main(): args = argparse.ArgumentParser() args.add_argument("--status-file", type=Path, required=True) args.add_argument("--assets-dir", type=Path, default=default_chat_assets_dir()) + args.add_argument("--requests", type=int, default=13) selected = args.parse_args() + if not 1 <= selected.requests <= 120: + args.error("--requests must be between 1 and 120") status = json.loads(selected.status_file.read_text()) with TemporaryDirectory(prefix="loopx-steward-requests-") as directory: root = Path(directory).resolve() @@ -61,7 +65,10 @@ def main(): "检查超时后的安全重试", "复核下载文件中的版本信息", "检查月底跨时区的账期边界", "整理旧接口的调用方清单", "核对缺失凭证时的处理方式", "检查手机上的导出入口", "复核迁移说明中的示例"] last_receipt = None - for index, title in enumerate(titles): + for index in range(selected.requests): + title = titles[index % len(titles)] + if index >= len(titles): + title += f" · 记录 {index + 1}" target = {"goal_id": "billing-exports", "agent_id": "reviewer" if index % 3 == 2 else "builder"} turn, _ = store.create_turn(session["session_id"], client_turn_id=f"synthetic-{index}", message=title, origin="web") brief = {"schema_version": "collaboration_brief_v0", "purpose": title, @@ -78,7 +85,10 @@ def main(): acknowledge(runtime, "billing-exports", target["agent_id"], receipt["request_id"], "adopt", "已接到原工作，正在核对实际产物；尚未形成完成结论。", registry=registry, caller_goal_ref=receipt["goal_ref"]) last_receipt = receipt - report(runtime, "billing-exports", "builder", last_receipt["request_id"], "conclusion", + if (selected.requests - 1) % 3 == 2: + acknowledge(runtime, "billing-exports", "reviewer", last_receipt["request_id"], "adopt", + "已接到原工作，正在核对实际产物；尚未形成完成结论。", registry=registry, caller_goal_ref=last_receipt["goal_ref"]) + report(runtime, "billing-exports", target["agent_id"], last_receipt["request_id"], "conclusion", "已核对本次迁移说明与示例，版本一致，原始检查结果已保留。", registry=registry, caller_goal_ref=last_receipt["goal_ref"]) @@ -92,8 +102,12 @@ def _goal_contexts(self): self._send_json({"ok": True, "schema_version": "loopx_chat_goal_contexts_v0", "goals": []}) def _request_inventory_read(self): self.server.inventory_reads += 1 - if self.server.fault: + later_window = parse_qs(urlparse(self.path).query).get("offset", ["0"])[0] != "0" + 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() def do_GET(self): if urlparse(self.path).path == "/status.json": @@ -122,6 +136,7 @@ def do_GET(self): server.lark_cli_resolution = LarkCliResolution(None, False, "synthetic", None, "lark_cli_not_installed") server.runtime_root_override, server.selected_goal_id, server.verbose = str(runtime), None, False server.assets_dir, server.fault, server.posts, server.inventory_reads = selected.assets_dir, False, 0, 0 + 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", "adapter_kind": "codex_app_server", "available": True, "streaming": True, "resume": True, "interrupt": True}] @@ -143,8 +158,12 @@ def source_hashes(): break if command == "fault": server.fault = True + elif command == "fault-window": + server.fault_window = True + elif command == "deliver-window": + server.deliver_window = True elif command == "recover": - server.fault = False + server.fault = server.fault_window = server.deliver_window = False elif command == "deliver": drain(runtime, registry, store, lambda *_: None) elif command == "source-unavailable": diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 2553cf66e1..230ec768d9 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -188,6 +188,10 @@ request/Goal/Agent identities. This is a refusal readback, not a readable brief, cancellation or completion. The App's warning glyph exposes that cause on demand; other unreadable sources remain unknown. Refresh rechecks the original owners; the read neither repairs historical receipts nor grants execution/return authority. +Pages default to 12 rows. An explicit `limit` of 1–48 lets the App refresh up to +four already loaded pages with one verified scan; larger loaded views use more +bounded windows. Refresh publishes only matching snapshots. No cross-request +content/authorization cache is added; this does not qualify a faster initial read. 管家标题栏提供常驻委派入口，按接收方查看待跟进请求，并可展开当前交办内容、关联 工作和原对话。同一来源的较早会话不会因重开会话而丢失；其它用户、来源、群聊和 @@ -195,6 +199,8 @@ the read neither repairs historical receipts nor grants execution/return authori 来源与原路匹配但发送回执冲突时，保留不完整计数，以警示小符号按需提供至多 12 条 具体请求身份和原因；不展示未获读取资格的交办正文，不推定已取消或完成。刷新重新 核验原 owner，不修改历史回执，也不解除执行或返回权限；其它读取缺口继续保持未知。 +默认每页 12 条；显式 `limit`（1–48）让刷新在一次核验中读取至多四个已加载页，更长列表 +分批读取，快照一致后一起展示。不新增跨请求内容或授权缓存，也不据此宣称首次读取已提速。 The automatic agenda is a partial read (`read_complete=false`): it cannot fulfill the full Inbox read or replace its requirements. Before acting, read the current diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index f24c334285..e94e87de71 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/loopx/control_plane/collaboration/conversation_inventory.py @@ -39,7 +39,9 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core The caller supplies provider-validated Goal visibility and rechecks it after this read. A model/caller-supplied channel, path or source id is never used. """ - if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 12: + # Normal pages remain 12 rows. A refresh may restore four loaded pages + # together, reusing this read's full scan and authorization checks. + if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 48: raise ValueError("invalid request page") if not isinstance(search, str) or len(search) > 200: raise ValueError("request search must be at most 200 characters") diff --git a/loopx/presentation/request_control_api.py b/loopx/presentation/request_control_api.py index ee51b0ef3a..074c6f9049 100644 --- a/loopx/presentation/request_control_api.py +++ b/loopx/presentation/request_control_api.py @@ -81,7 +81,7 @@ def _request_inventory_read(self) -> None: try: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) - if (set(query) - {"session_id", "view", "offset", "search"} or "session_id" not in query + if (set(query) - {"session_id", "view", "offset", "search", "limit"} or "session_id" not in query or any(len(values) != 1 for values in query.values())): raise ValueError("select one original conversation") view = query.get("view", ["unfinished"])[0] @@ -107,7 +107,7 @@ def validate_source(source): result = read_conversation_requests(controller.coordination_runtime_root, self.server.registry_path, self.server.chat_store, session, goal_ids=goals, core_goal_ids=core, validate_source_session=validate_source, history=view == "history", offset=int(query.get("offset", ["0"])[0]), - search=query.get("search", [""])[0]) + limit=int(query.get("limit", ["12"])[0]), search=query.get("search", [""])[0]) for sid, proof in sources.items(): current, current_goals, current_core = self._request_inventory_scope(sid) if (source_basis(current), current_goals, current_core) != proof: diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index 7282fb9d07..ea0c190d77 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -1,6 +1,8 @@ """Original steward requests, not inferred worker activity or a second ledger.""" from urllib.parse import urlencode +import pytest + from test_chat_goal_amendment import api as api from test_goal_amendment_authority import source_goal as source_goal from test_chat_request_control import _source @@ -213,7 +215,70 @@ def test_turn_locator_rereads_content_and_rejects_lost_identity(api): assert _inventory(api, route["session_id"])[1]["pending_count"] == 1 -def test_inventory_withholds_all_content_when_closing_source_scope_changes(api, monkeypatch): +def test_refresh_window_preserves_full_pages_and_source_checks(api, monkeypatch): + from loopx.presentation.request_control_api import RequestControlMixin + + _, _, route = _source(api) + _, f, server = api + store = server.chat_store + session = store.load_session(route["session_id"]) + expected_ids = {route["request_id"]} + for number in range(60): + turn, _ = store.create_turn(session["session_id"], client_turn_id=f"retained-{number}", + message=f"Inspect retained artifact {number}: " + "Keep the complete source and its unknowns. " * 30, origin="web") + receipt = deliver(f["runtime"], f["registry"], session=session, turn=turn, + request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-a" if number % 2 else "agent-b"}, source_store=store) + expected_ids.add(receipt["request_id"]) + store.update_turn(session["session_id"], turn["turn_id"], status="completed", response={"context_handoff_receipt": receipt}) + store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) + saved = {path: path.read_bytes() for path in _root(f["runtime"]).rglob("*.json")} + scope_read = RequestControlMixin._request_inventory_scope + verified = [] + def observe_scope(reader, session_id): + verified.append(session_id) + return scope_read(reader, session_id) + monkeypatch.setattr(RequestControlMixin, "_request_inventory_scope", observe_scope) + + rows, cursor, snapshots = [], 0, set() + while cursor is not None: + code, page = _inventory(api, route["session_id"], view="history", offset=cursor) + assert code == 200 and len(page["rows"]) <= 12 + rows.extend(page["rows"]) + snapshots.add(page["snapshot_digest"]) + cursor = page["next_offset"] + assert len(rows) == len({row["request_id"] for row in rows}) == 61 + assert {row["request_id"] for row in rows} == expected_ids + assert len(verified) == 12 and len(snapshots) == 1 + verified.clear() + code, first = _inventory(api, route["session_id"], view="history", limit=48) + assert code == 200 and len(first["rows"]) == 48 and first["next_offset"] == 48 + code, rest = _inventory(api, route["session_id"], view="history", offset=48, limit=24) + assert code == 200 and rest["next_offset"] is None + assert first["rows"] + rest["rows"] == rows + assert first["pending_count"] == rest["pending_count"] == first["matched"] == 61 + assert first["snapshot_digest"] == rest["snapshot_digest"] == snapshots.pop() + assert first["coverage"] == rest["coverage"] == {"scan_complete": True, "unreadable": 0, "diagnostics": []} + assert len(verified) == 4, "Each window still verifies the source before and after its full scan" + assert {path: path.read_bytes() for path in saved} == saved + + +def test_inventory_refresh_window_is_bounded_and_preserves_default_page(api): + _, _, route = _source(api) + for limit in (0, -1, 49, 1000, "12.5", "", "true"): + code, denied = _inventory(api, route["session_id"], limit=limit) + assert code == 400 and denied["error_code"] == "request_inventory_unavailable" + assert "rows" not in denied + call, _, server = api + server.selected_goal_id = None + code, denied = call(path="/api/chat/request-inventory?" + urlencode({"session_id": route["session_id"]}) + "&limit=12&limit=48") + assert code == 400 and "rows" not in denied + code, default = _inventory(api, route["session_id"]) + assert code == 200 and len(default["rows"]) == 1 + assert _inventory(api, route["session_id"], limit=12)[1]["rows"] == default["rows"] + + +@pytest.mark.parametrize("limit", [12, 48]) +def test_inventory_withholds_all_content_when_closing_source_scope_changes(api, monkeypatch, limit): from loopx.presentation.request_control_api import RequestControlMixin _, _, route = _source(api) @@ -224,7 +289,7 @@ def observe_scope(reader, session_id): verified.append(session_id) return (observed, [], core) if len(verified) == 2 else (observed, goals, core) monkeypatch.setattr(RequestControlMixin, "_request_inventory_scope", observe_scope) - code, denied = _inventory(api, route["session_id"]) + code, denied = _inventory(api, route["session_id"], limit=limit) assert code == 400 and denied["error_code"] == "request_inventory_unavailable" assert "rows" not in denied and "Check the current contract" not in str(denied) From 89e4e791c3f7ec3aa368dc9e806356ea6eb3adb1 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:50:08 +0800 Subject: [PATCH 3/6] refactor(collaboration): separate request observation from projection Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../collaboration/conversation_inventory.py | 73 +++++++++++-------- 1 file changed, 41 insertions(+), 32 deletions(-) diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index 4640aec84d..724d47acf9 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/loopx/control_plane/collaboration/conversation_inventory.py @@ -32,6 +32,43 @@ def _last_event(values): return max(dated, default=(None, None), key=lambda pair: pair[0])[1] +def _project_request_observations(root, registry, observations, *, core_goal_ids, history): + """Combine captured receipt facts with current linked Core observations.""" + rows, unreadable, cache = [], 0, {} + for row, source_session_id, turn, state, agenda, controlled, read, read_error, decision, decision_error, replies in observations: + try: + gid, aid = row["goal_id"], row["agent_id"] + work = read_linked_work(root, registry, row, cache, + include_core_details=core_goal_ids is None or gid in core_goal_ids) + warnings = list(state.get("warnings", [])) + work["warnings"] + warnings += [error for error in (read_error, decision_error) if error] + if not history and agenda is None: + rows.append({"request_id": row["request_id"], "unfinished": False}) + continue + control_unavailable = "request_control" in row and controlled is None + brief = controlled["brief"] if controlled else None if control_unavailable else row.get("brief") + assessment = controlled["decision"] if controlled else { + "status": "unavailable" if control_unavailable else decision.get("decision", "not_recorded"), + "at": None if control_unavailable else decision.get("decided_at"), + "reason": None if control_unavailable else decision.get("reason")} + read_at = controlled["read"].get("at") if controlled else read.get("read_at") + rows.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": aid, + "unfinished": agenda is not None, "agenda": agenda, + "control_state": (state.get("request_control_view") or {}).get("control_state"), + "brief": brief, "source_question": turn.get("message", ""), + "request_revision": controlled["request_revision"] if controlled else None if control_unavailable else 1, + "read_status": controlled["card_read_status"] if controlled else + "unavailable" if read_error or control_unavailable else "supplied" if read else "pending", + "decision": assessment["status"], "decision_reason": assessment.get("reason"), + "returns": replies, "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"], + "warnings": warnings, "source_session_id": source_session_id, "source_turn_id": turn["turn_id"], + "last_event_at": _last_event([row.get("delivered_at"), read_at, assessment.get("at"), + *(reply.get(key) for reply in replies for key in ("created_at", "delivered_at"))])}) + except (OSError, ValueError, KeyError, TypeError): + unreadable += 1 + return rows, unreadable + + def read_conversation_requests(root, registry, store, session, *, goal_ids, core_goal_ids, validate_source_session, history=False, offset=0, limit=12, search=""): """Read only this host's exact source Session, with fresh caller read scope. @@ -45,7 +82,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core raise ValueError("request search must be at most 200 characters") query = search.strip().casefold() paths = sorted((_root(root) / "entries").glob("*/*.json")) - rows, unreadable, cache, turn_readers, diagnostics = [], 0, {}, {}, [] + unreadable, turn_readers, diagnostics = 0, {}, [] groups, work_reads = {}, [] for path in paths[:2000]: try: @@ -135,37 +172,9 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core unreadable += failed # Request/return observations are version-consistent. Core work is a # separate observation: release all lifetime/request guards before its IO. - for row, source_session_id, turn, state, agenda, controlled, read, read_error, decision, decision_error, replies in work_reads: - try: - gid, aid = row["goal_id"], row["agent_id"] - work = read_linked_work(root, registry, row, cache, - include_core_details=core_goal_ids is None or gid in core_goal_ids) - warnings = list(state.get("warnings", [])) + work["warnings"] - warnings += [error for error in (read_error, decision_error) if error] - if not history and agenda is None: - rows.append({"request_id": row["request_id"], "unfinished": False}) - continue - control_unavailable = "request_control" in row and controlled is None - brief = controlled["brief"] if controlled else None if control_unavailable else row.get("brief") - assessment = controlled["decision"] if controlled else { - "status": "unavailable" if control_unavailable else decision.get("decision", "not_recorded"), - "at": None if control_unavailable else decision.get("decided_at"), - "reason": None if control_unavailable else decision.get("reason")} - read_at = controlled["read"].get("at") if controlled else read.get("read_at") - rows.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": aid, - "unfinished": agenda is not None, "agenda": agenda, - "control_state": (state.get("request_control_view") or {}).get("control_state"), - "brief": brief, "source_question": turn.get("message", ""), - "request_revision": controlled["request_revision"] if controlled else None if control_unavailable else 1, - "read_status": controlled["card_read_status"] if controlled else - "unavailable" if read_error or control_unavailable else "supplied" if read else "pending", - "decision": assessment["status"], "decision_reason": assessment.get("reason"), - "returns": replies, "linked_todos": work["linked_todos"], "evidence_refs": work["evidence_refs"], - "warnings": warnings, "source_session_id": source_session_id, "source_turn_id": turn["turn_id"], - "last_event_at": _last_event([row.get("delivered_at"), read_at, assessment.get("at"), - *(reply.get(key) for reply in replies for key in ("created_at", "delivered_at"))])}) - except (OSError, ValueError, KeyError, TypeError): - unreadable += 1 + rows, projection_failures = _project_request_observations(root, registry, work_reads, + core_goal_ids=core_goal_ids, history=history) + unreadable += projection_failures rows = list({row["request_id"]: row for row in rows}.values()) pending_count = sum(row["unfinished"] for row in rows) visible = [row for row in rows if history or row["unfinished"]] From cdb5f3fbecb6ead202c410e1914758347f6e2a70 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 06:39:28 +0800 Subject: [PATCH 4/6] perf(collaboration): batch guarded inventory receipt observations Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../collaboration/conversation_inventory.py | 138 ++++++++++++------ .../test_chat_request_inventory.py | 73 ++++++++- 2 files changed, 169 insertions(+), 42 deletions(-) diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index 724d47acf9..40d7f1bd8d 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/loopx/control_plane/collaboration/conversation_inventory.py @@ -6,6 +6,7 @@ """ from __future__ import annotations +from contextlib import ExitStack, contextmanager from datetime import datetime import hashlib import json @@ -32,6 +33,41 @@ def _last_event(values): return max(dated, default=(None, None), key=lambda pair: pair[0])[1] +@contextmanager +def _locked_request_batch(root, scope, requests): + """Retain each request version for one page-sized typed observation.""" + available, lock_keys = set(), set() + with ExitStack() as locks: + for path, _, request_id in sorted(requests, key=lambda item: (item[2], str(item[0]))): + # Exact requests share their guard across legacy entry locations. + # Legacy records still use their individual entry lock. + key = (request_id, None if scope.exact else path) + if key not in lock_keys: + try: + locks.enter_context(_request_lock(root, request_id, scope, path.with_suffix(".lock"))) + except (OSError, ValueError, KeyError, TypeError): + continue + lock_keys.add(key) + available.add(path) + yield available + + +def _read_receipt_batch(root, rows): + if not rows: + return [] + try: + observed = read_inbox_transaction(root, {"operation": "receipt_states", "rows": rows, "include_agenda": True}) + states = list(zip(observed["items"], observed["agenda_items"], strict=True)) + if len(states) != len(rows): + raise ValueError("Inbox receipt count does not match requests") + return states + except (OSError, ValueError, KeyError, TypeError): + # A failed batch is only a read. Retry its rows independently while + # all their guards remain held; one unavailable record must not hide + # its readable neighbors or become an empty successful observation. + return [None] if len(rows) == 1 else [_read_receipt_batch(root, [row])[0] for row in rows] + + def _project_request_observations(root, registry, observations, *, core_goal_ids, history): """Combine captured receipt facts with current linked Core observations.""" rows, unreadable, cache = [], 0, {} @@ -69,21 +105,10 @@ def _project_request_observations(root, registry, observations, *, core_goal_ids return rows, unreadable -def read_conversation_requests(root, registry, store, session, *, goal_ids, core_goal_ids, - validate_source_session, history=False, offset=0, limit=12, search=""): - """Read only this host's exact source Session, with fresh caller read scope. - - The caller supplies provider-validated Goal visibility and rechecks it after - this read. A model/caller-supplied channel, path or source id is never used. - """ - if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 12: - raise ValueError("invalid request page") - if not isinstance(search, str) or len(search) > 200: - raise ValueError("request search must be at most 200 characters") - query = search.strip().casefold() +def _request_groups(root, store, session, goal_ids, validate_source_session): + """Discover source-validated routes before acquiring any Goal guard.""" paths = sorted((_root(root) / "entries").glob("*/*.json")) - unreadable, turn_readers, diagnostics = 0, {}, [] - groups, work_reads = {}, [] + unreadable, groups = 0, {} for path in paths[:2000]: try: candidate = _read(path) @@ -111,6 +136,23 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core groups.setdefault(key, []).append((path, route, candidate["request_id"])) except (OSError, ValueError, KeyError, TypeError): unreadable += 1 + return groups, unreadable, len(paths) <= 2000 + + +def read_conversation_requests(root, registry, store, session, *, goal_ids, core_goal_ids, + validate_source_session, history=False, offset=0, limit=12, search=""): + """Read only this host's exact source Session, with fresh caller read scope. + + The caller supplies provider-validated Goal visibility and rechecks it after + this read. A model/caller-supplied channel, path or source id is never used. + """ + if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 12: + raise ValueError("invalid request page") + if not isinstance(search, str) or len(search) > 200: + raise ValueError("request search must be at most 200 characters") + query = search.strip().casefold() + groups, unreadable, scan_complete = _request_groups(root, store, session, goal_ids, validate_source_session) + turn_readers, diagnostics, work_reads = {}, [], [] # Reuse only the held lifetime guard, never an authorization or request # snapshot. Resolve sources outside it first: their provider checks can # read another authority. Each source/Turn and request is reread below. @@ -121,9 +163,16 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core for (gid, _), requests in batches: start, failed = len(work_reads), 0 try: - with collaboration_goal_scope(registry, goal_id=gid, agents=(), - caller_goal_ref=requests[0][1].get("goal_ref")) as scope: + with ( + collaboration_goal_scope(registry, goal_id=gid, agents=(), + caller_goal_ref=requests[0][1].get("goal_ref")) as scope, + _locked_request_batch(root, scope, requests) as available, + ): + prepared = [] for path, route, request_id in requests: + if path not in available: + failed += 1 + continue try: source_session = store.load_session(route["session_id"]) if not source_session or source_session.get("channel_id") != session["channel_id"]: @@ -135,33 +184,40 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core turn = turn_readers[sid](route["client_turn_id"]) if turn is None: raise ValueError("original source Turn unavailable") - with _request_lock(root, request_id, scope, path.with_suffix(".lock")): - row = _entry(root, gid, route["agent_id"], request_id, scope=scope) - if row.get("source_kind") == "peer": - continue - grant = source_context_target_authority(root, source_session, turn, - {"goal_id": gid, "agent_id": route["agent_id"]}) - try: - decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route, - initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"])) - except CollaborationLifecycleRejected as exc: - # Only this typed refusal proves the route and source - # matched before the conflicting receipt was observed. - # Withhold the brief, work and completion state as before. - if exc.decision.get("delivery_proof_reason") == "handoff_receipt_conflict" and len(diagnostics) < 12: - diagnostics.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": row["agent_id"], - "code": exc.decision["code"], "reason": exc.decision["delivery_proof_reason"]}) - raise - observed = read_inbox_transaction(root, {"operation": "receipt_states", "rows": [row], "include_agenda": True}) - state, agenda = observed["items"][0], observed["agenda_items"][0] - controlled = state.get("readback") - read, read_error = _receipt(root, "reads", row) - decision, decision_error = _receipt(root, "decisions", row) - replies = reply_status(root, row) + row = _entry(root, gid, route["agent_id"], request_id, scope=scope) + if row.get("source_kind") == "peer": + continue + grant = source_context_target_authority(root, source_session, turn, + {"goal_id": gid, "agent_id": route["agent_id"]}) + try: + decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route, + initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"])) + except CollaborationLifecycleRejected as exc: + # Only this typed refusal proves the route and source + # matched before the conflicting receipt was observed. + # Withhold the brief, work and completion state as before. + if exc.decision.get("delivery_proof_reason") == "handoff_receipt_conflict" and len(diagnostics) < 12: + diagnostics.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": row["agent_id"], + "code": exc.decision["code"], "reason": exc.decision["delivery_proof_reason"]}) + raise # Do not retain full model responses while waiting for # linked-work IO; only the displayed question is used. source_turn = {"turn_id": turn["turn_id"], "message": turn.get("message", "")} - work_reads.append((row, source_session["session_id"], source_turn, state, agenda, controlled, + prepared.append((row, source_session["session_id"], source_turn)) + except (OSError, ValueError, KeyError, TypeError): + failed += 1 + observations = _read_receipt_batch(root, [row for row, _, _ in prepared]) + for (row, source_session_id, source_turn), observed in zip(prepared, observations, strict=True): + if observed is None: + failed += 1 + continue + try: + state, agenda = observed + controlled = state.get("readback") + read, read_error = _receipt(root, "reads", row) + decision, decision_error = _receipt(root, "decisions", row) + replies = reply_status(root, row) + work_reads.append((row, source_session_id, source_turn, state, agenda, controlled, read, read_error, decision, decision_error, replies)) except (OSError, ValueError, KeyError, TypeError): failed += 1 @@ -186,7 +242,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core (row.get("brief") or {}).get("purpose"), (row.get("brief") or {}).get("context")))] visible.sort(key=lambda row: (row.get("last_event_at") or "", row["request_id"]), reverse=True) page = visible[offset:offset + limit] - coverage = {"scan_complete": len(paths) <= 2000, "unreadable": unreadable, "diagnostics": diagnostics} + coverage = {"scan_complete": scan_complete, "unreadable": unreadable, "diagnostics": diagnostics} # A page fence over observations, not an atomic snapshot or a new ledger. snapshot_digest = hashlib.sha256(json.dumps([visible, coverage, query], sort_keys=True, ensure_ascii=False).encode()).hexdigest() return {"ok": True, "session_id": session["session_id"], "rows": page, diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index 77637d9379..0938f6cb1d 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -1,6 +1,8 @@ """Original steward requests, not inferred worker activity or a second ledger.""" from urllib.parse import urlencode +import pytest + from test_chat_goal_amendment import api as api from test_goal_amendment_authority import source_goal as source_goal from test_chat_request_control import _source @@ -246,6 +248,7 @@ def test_inventory_bounds_lifetime_guards_and_keeps_core_io_outside_them(api, mo response={"context_handoff_receipt": delivered}) server.chat_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) original_scope, original_work = inventory.collaboration_goal_scope, inventory.read_linked_work + original_lock, original_receipts = inventory._request_lock, inventory.read_inbox_transaction original_classify = request_control_api.conversation_scope classifications = [] @@ -254,7 +257,7 @@ def classify(source): return original_classify(source) monkeypatch.setattr(request_control_api, "conversation_scope", classify) - held, batches = [], [] + held, batches, locked, receipt_batches = [], [], set(), [] @contextmanager def scope(*args, **kwargs): @@ -276,13 +279,32 @@ def work(*args, **kwargs): assert not held, "Core IO must not nest under a Goal lifetime guard" return original_work(*args, **kwargs) + @contextmanager + def lock(root, request_id, *args, **kwargs): + with original_lock(root, request_id, *args, **kwargs): + locked.add(request_id) + try: + yield + finally: + locked.remove(request_id) + + def receipts(root, request, **kwargs): + ids = {row["request_id"] for row in request["rows"]} + assert held and ids <= locked, "Receipt versions need every corresponding request lock" + receipt_batches.append(len(ids)) + return original_receipts(root, request, **kwargs) + monkeypatch.setattr(inventory, "collaboration_goal_scope", scope) monkeypatch.setattr(inventory, "_entry", entry) monkeypatch.setattr(inventory, "read_linked_work", work) + monkeypatch.setattr(inventory, "_request_lock", lock) + monkeypatch.setattr(inventory, "read_inbox_transaction", receipts) code, value = _inventory(api, route["session_id"], view="history") assert code == 200 and value["matched"] == value["pending_count"] == 13 assert value["coverage"] == {"scan_complete": True, "unreadable": 0, "diagnostics": []} assert sum(batches) == 13 and len(batches) == 2 and max(batches) <= 12 + assert sum(receipt_batches) == 13 and len(receipt_batches) == 2 and max(receipt_batches) <= 12 + assert not locked, "A completed read must release every request lock" # Two fresh boundaries and at most one additional pure projection for the # unchanged source, independently of row count and on each HTTP read. assert set(classifications) == {session["session_id"]} and 2 <= len(classifications) <= 3 @@ -291,6 +313,55 @@ def work(*args, **kwargs): assert 2 <= len(classifications) - first_count <= 3 +@pytest.mark.parametrize("failure", ["request_lock", "receipt"]) +def test_inventory_isolates_one_failed_request_and_rereads_on_recovery(api, monkeypatch, failure): + from contextlib import contextmanager + from loopx.control_plane.collaboration import conversation_inventory as inventory + + _, _, route = _source(api) + _, f, server = api + store = server.chat_store + session = store.load_session(route["session_id"]) + expected = {route["request_id"]: store.load_turn(route["session_id"], route["turn_id"])["message"]} + for number in range(2): + message = f"Review artifact {number}\n" + "Retain the complete source question.\n" * 80 + turn, _ = store.create_turn(session["session_id"], client_turn_id=f"isolated-{number}", message=message, origin="web") + receipt = deliver(f["runtime"], f["registry"], session=session, turn=turn, + request={"goal_id": f["ref"]["goal_id"], "agent_id": "agent-b"}, source_store=store) + store.update_turn(session["session_id"], turn["turn_id"], status="completed", response={"context_handoff_receipt": receipt}) + store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) + expected[receipt["request_id"]] = message + broken = route["request_id"] + original_lock, original_receipts = inventory._request_lock, inventory.read_inbox_transaction + + @contextmanager + def lock(root, request_id, *args, **kwargs): + if request_id == broken: + raise OSError("request lock unavailable") + with original_lock(root, request_id, *args, **kwargs): + yield + + def receipts(root, request, **kwargs): + if any(row["request_id"] == broken for row in request["rows"]): + raise OSError("request receipt unavailable") + return original_receipts(root, request, **kwargs) + + monkeypatch.setattr(inventory, "_request_lock" if failure == "request_lock" else "read_inbox_transaction", + lock if failure == "request_lock" else receipts) + saved = {path: path.read_bytes() for path in _root(f["runtime"]).rglob("*.json")} + code, partial = _inventory(api, session["session_id"], view="history") + assert code == 200 and partial["matched"] == partial["pending_count"] == 2 + assert partial["coverage"] == {"scan_complete": True, "unreadable": 1, "diagnostics": []} + assert {row["request_id"]: row["source_question"] for row in partial["rows"]} == { + key: value for key, value in expected.items() if key != broken} + assert {path: path.read_bytes() for path in saved} == saved + monkeypatch.setattr(inventory, "_request_lock", original_lock) + monkeypatch.setattr(inventory, "read_inbox_transaction", original_receipts) + recovered = _inventory(api, session["session_id"], view="history")[1] + assert recovered["coverage"]["unreadable"] == 0 and recovered["matched"] == 3 + assert {row["request_id"]: row["source_question"] for row in recovered["rows"]} == expected + + def test_inventory_reclassifies_changed_facts_for_the_same_source_session(api, monkeypatch): _, _, route = _source(api) _, _, server = api From 2fa0671a8df319754e512074ee4ecd3676b9e1ee Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 10:11:52 +0800 Subject: [PATCH 5/6] fix(workspace): retain source failure across delegation filters Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../smoke/delegation-search-browser-smoke.mjs | 16 ++++++++++++++++ .../personal-workspace/conversation-requests.tsx | 12 +++++++++--- .../steward_requests_fixture.py | 9 ++++++++- 3 files changed, 33 insertions(+), 4 deletions(-) diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs index 972a4d76db..efe7103c3f 100644 --- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs @@ -86,6 +86,22 @@ try { await page.screenshot({path: resolve(output, 'read-failure.png')}); await fixture('recover'); await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled(); await rows(4); + await fixture('local-source-unavailable'); + const denied = await fetch(`${ready.origin}/api/chat/request-inventory?session_id=${ready.session_id}`); + assert.equal(denied.status, 400); + assert.equal((await denied.json()).error_code, 'request_inventory_unavailable'); + await dialog.getByRole('button', {name: '刷新委派', exact: true}).click(); await settled(); + await dialog.getByRole('alert').waitFor(); await rows(0); + assert.equal(await trigger.getAttribute('aria-label'), '委派，— 条待跟进'); + assert.equal(await dialog.locator('.conversation-requests-pagination').count(), 0); + // Changing the filter must not turn the same missing source into a new + // empty workspace, nor may an unsuccessful retry lose that distinction. + await search('builder'); await settled(); await dialog.getByRole('alert').waitFor(); + await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled(); + await dialog.getByRole('alert').waitFor(); await rows(0); + await fixture('local-source-recover'); + await dialog.getByRole('button', {name: '重试', exact: true}).click(); await settled(); await rows(9); + await search('reviewer'); await rows(4); assert.equal((await fixture('inspect')).source_unchanged, true); await fixture('deliver'); await search('复核迁移说明中的示例'); await settled(); await dialog.getByText('未找到匹配的委派。', {exact: true}).waitFor(); 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 65a50a995e..3ef5a906c2 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -216,7 +216,8 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const restorePagingFocus = useRef(false); useEffect(() => { setOpen(false); setHistory(false); setPages([]); setError(false); setSearch(""); setSearchDraft(""); - }, [sessionId]); + loadedScope.current = null; loadedSource.current = null; + }, [sessionId, ownerFallback]); useEffect(() => { if (open) dialog.current?.showModal(); else dialog.current?.close(); }, [open]); @@ -235,7 +236,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const refreshRows = sameScope ? Math.max(12, Math.ceil(loadedRows / 12) * 12) : 12; // Keep verified rows during a same-source refresh, but never across scope // changes or a failed read. Publish refreshed pages together below. - if (!sameScope) {setPages([]); loadedSource.current = null;} + if (!sameScope) setPages([]); loadedScope.current = scope; try { let source = sessionId && sessionId !== "new-session-pending" ? sessionId : undefined; @@ -244,7 +245,12 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se source = listed.sessions.find(row => row.channel_id === "manager")?.session_id; } if (current !== generation.current) return; - if (!source) {setPages([]); setNoSession(true); return;} + if (!source) { + // A vanished source is a failed read, not a new empty workspace. Keep + // its lineage across filters so retry continues to report the loss. + if (loadedSource.current) throw new Error("Previous steward source is unavailable."); + setPages([]); setNoSession(true); return; + } const sameSource = loadedSource.current === source; if (!sameSource) setPages([]); loadedSource.current = source; diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index 7d4535411c..6156e12939 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/examples/personal-workspace-browser/steward_requests_fixture.py @@ -4,7 +4,8 @@ availability is display-only fixture metadata; POSTs cannot execute it. Requests, decisions and return delivery below use the production owners in a temp runtime. Commands on stdin: inspect, fault, fault-window, deliver-window, recover, -deliver, source-unavailable, source-recover, stop. Window faults/returns apply +deliver, source-unavailable, source-recover, local-source-unavailable, +local-source-recover, stop. Window faults/returns apply after the first page to challenge multi-window refresh readback. """ from __future__ import annotations @@ -144,6 +145,8 @@ def reject_provider(_profile): raise AssertionError("An unselected steward must not read an external provider") install_conversation_transports(server, factories=(), observe_default=reject_provider) bindings = server.runtime_controller.project_contexts.conversation_bindings + local_session_path = store._session_path(session["session_id"]) + local_session_bytes = local_session_path.read_bytes() thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() def source_hashes(): @@ -170,6 +173,10 @@ def source_hashes(): bindings.path.write_text("{", encoding="utf-8") elif command == "source-recover": bindings.path.unlink() + elif command == "local-source-unavailable": + local_session_path.write_text("{invalid synthetic session", encoding="utf-8") + elif command == "local-source-recover": + local_session_path.write_bytes(local_session_bytes) elif command != "inspect": raise ValueError("Unknown fixture command") print(json.dumps({"event": command, "posts": server.posts, "inventory_reads": server.inventory_reads, From 61bfc3815f7a9fe207b90e0059ec155a90393288 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 30 Oct 2026 11:34:24 +0800 Subject: [PATCH 6/6] fix(workspace): advance delegation paging after snapshot changes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../smoke/delegation-search-browser-smoke.mjs | 13 ++++++++++++- .../personal-workspace/conversation-requests.tsx | 15 +++++++++++---- .../steward_requests_fixture.py | 11 +++++++++-- 3 files changed, 32 insertions(+), 7 deletions(-) diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs index efe7103c3f..3628865771 100644 --- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs @@ -142,9 +142,20 @@ try { await fixture('source-recover'); await page.getByRole('button', {name: 'Retry history', exact: true}).click(); await page.getByRole('button', {name: /Delegations,.*awaiting follow-up/}).click(); await page.getByRole('searchbox', {name: 'Search delegations'}).waitFor(); + const english = page.getByRole('dialog', {name: 'Delegations', exact: true}); + await english.getByRole('button', {name: 'All records', exact: true}).click(); await rows(12); + // A real return changes ordering and the snapshot while More is reading. + // Rebuild from the beginning, but fulfill this click's requested extent; + // the parent message refresh must not replace it with the old page count. + await fixture('report-window'); + const nextPage = english.getByRole('button', {name: 'More records', exact: true}); + await nextPage.focus(); await page.keyboard.press('Enter'); await rows(13); + assert.equal(await english.getByRole('button', {name: 'More records', exact: true}).count(), 0); + assert.equal(await page.evaluate(() => document.activeElement?.textContent), '13 / 13 shown'); + await english.getByRole('button', {name: 'Awaiting follow-up', exact: true}).click(); await rows(11); assert.equal(writes.some(path => /request-(?:inventory|control)|team-work/.test(path)), false); assert.deepEqual(errors, []); - console.log('PASS: packaged search, pre-pagination match, count, original conversation, clear/paging, failure/retry, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery and read-only receipts.'); + console.log('PASS: packaged search, pre-pagination match, count, original conversation, clear/paging, failure/retry, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery, changed-snapshot More and read-only receipts.'); } } catch (error) { if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());} 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 3ef5a906c2..c11a76bc9c 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -207,6 +207,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const [originalRevision, setOriginalRevision] = useState(0); const loadedScope = useRef(null); const loadedSource = useRef(null); + const requestedRows = useRef(12); const generation = useRef(0); const dialog = useRef(null); const trigger = useRef(null); @@ -217,6 +218,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se useEffect(() => { setOpen(false); setHistory(false); setPages([]); setError(false); setSearch(""); setSearchDraft(""); loadedScope.current = null; loadedSource.current = null; + requestedRows.current = 12; }, [sessionId, ownerFallback]); useEffect(() => { if (open) dialog.current?.showModal(); else dialog.current?.close(); @@ -233,7 +235,11 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se const scope = JSON.stringify([sessionId, ownerFallback, history, search]); const sameScope = loadedScope.current === scope; const loadedRows = pages.reduce((count, page) => count + page.rows.length, 0); - const refreshRows = sameScope ? Math.max(12, Math.ceil(loadedRows / 12) * 12) : 12; + // A fresh snapshot or parent refresh must retain the page requested by + // More, including while that read has not published its rows yet. + requestedRows.current = sameScope + ? Math.max(requestedRows.current, offset ? offset + 12 : Math.ceil(loadedRows / 12) * 12) : 12; + const refreshRows = requestedRows.current; // Keep verified rows during a same-source refresh, but never across scope // changes or a failed read. Publish refreshed pages together below. if (!sameScope) setPages([]); @@ -252,7 +258,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se setPages([]); setNoSession(true); return; } const sameSource = loadedSource.current === source; - if (!sameSource) setPages([]); + if (!sameSource) {setPages([]); requestedRows.current = 12;} loadedSource.current = source; const refreshed: ConversationRequestInventory[] = []; let cursor: number | null = offset; @@ -272,7 +278,7 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se setPages(previous => offset ? [...previous, ...refreshed] : refreshed); if (!offset) setOriginalRevision(value => value + 1); } catch { - if (current === generation.current && !abort.signal.aborted) {setError(true); setPages([]);} + if (current === generation.current && !abort.signal.aborted) {requestedRows.current = 12; setError(true); setPages([]);} } finally {if (current === generation.current) setBusy(false);} } const page = pages[0], last = pages.at(-1); @@ -292,10 +298,11 @@ export function ConversationRequests({sessionId, ownerFallback, refreshKey}: {se 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()); setRevision(value => value + 1); }; - const close = () => {generation.current++; requestAbort.current?.abort(); setBusy(false); setOpen(false); trigger.current?.focus();}; + const close = () => {generation.current++; requestAbort.current?.abort(); requestedRows.current = 12; setBusy(false); setOpen(false); trigger.current?.focus();}; return