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/3] 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 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 2/3] 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 3/3] 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