perf(collaboration): batch guarded inventory observations

loopx/capabilities/manager_context/README.md

@@ -182,6 +182,16 @@ 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.
+The pure conversation-identity projection is reused only while the complete
+host Session facts are identical within that read. Changed facts are classified
+again; live provider, grant, receipt and closing-source checks are not cached.
+
 When the typed lifecycle proves a matching original source/route but finds a
 conflicting handoff receipt, coverage includes at most 12 diagnostic
 request/Goal/Agent identities. This is a refusal readback, not a readable brief,
@@ -197,6 +207,12 @@ checks or requests. Custom scope resolvers keep their existing validation path.
 管家标题栏提供常驻委派入口，按接收方查看待跟进请求，并可展开当前交办内容、关联
 工作和原对话。同一来源的较早会话不会因重开会话而丢失；其它用户、来源、群聊和
 项目会话不混入。结论已回传与成果已验收不是同一事实，历史记录仍可回看。
+同一次读取中，相同 Goal 实例按至多一页大小复用持有中的生命周期锁；逐请求锁、
+来源与 Turn 读回、授权及 typed 回执检查保留，每批重新观察当前实例。关联 Core 工作
+在释放这些锁后读取，不跨请求或分页缓存授权；分页与搜索仍扫描有界历史。
+同一次读取中，完整宿主 Session 字段相同时复用纯身份分类；字段变化即重新分类，
+provider、授权、回执及结束时的来源检查仍实时执行。
+
 来源与原路匹配但发送回执冲突时，保留不完整计数，以警示小符号按需提供至多 12 条
 具体请求身份和原因；不展示未获读取资格的交办正文，不推定已取消或完成。刷新重新
 核验原 owner，不修改历史回执，也不解除执行或返回权限；其它读取缺口继续保持未知。

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,73 +33,47 @@ def _last_event(values):
     return max(dated, default=(None, None), key=lambda pair: pair[0])[1]
 
 
-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.
+@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
 
-    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()
-    paths = sorted((_root(root) / "entries").glob("*/*.json"))
-    rows, unreadable, cache, turn_readers, diagnostics = [], 0, {}, {}, []
-    for path in paths[:2000]:
+
+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, {}
+    for row, source_session_id, turn, state, agenda, controlled, read, read_error, decision, decision_error, replies in observations:
         try:
-            candidate = _read(path)
-            if candidate.get("source_kind") == "peer":
-                continue
-            if candidate.get("source_channel") not in {None, session["channel_id"]}:
-                continue
-            if not isinstance(candidate.get("request_id"), str) or not BARE_SHA256_PATTERN.fullmatch(candidate["request_id"]):
-                raise ValueError("invalid request identity")
-            route = _read(_root(root) / "roundtrips" / (candidate["request_id"] + ".json"))
-            if route.get("channel_id") != session["channel_id"]:
-                continue
-            if not is_source_chat_host(root, route, store):
-                raise ValueError("original source host changed")
-            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)
-            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})
-                    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": aid,
-                                "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)
-            # Request/return observations are version-consistent. Core work is
-            # a separate current observation; never hold its request lock over
-            # another authority's IO.
+            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"]
@@ -122,11 +97,140 @@ 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):
             unreadable += 1
+    return rows, unreadable
+
+
+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, groups = 0, {}
+    for path in paths[:2000]:
+        try:
+            candidate = _read(path)
+            if candidate.get("source_kind") == "peer":
+                continue
+            if candidate.get("source_channel") not in {None, session["channel_id"]}:
+                continue
+            if not isinstance(candidate.get("request_id"), str) or not BARE_SHA256_PATTERN.fullmatch(candidate["request_id"]):
+                raise ValueError("invalid request identity")
+            route = _read(_root(root) / "roundtrips" / (candidate["request_id"] + ".json"))
+            if route.get("channel_id") != session["channel_id"]:
+                continue
+            if not is_source_chat_host(root, route, store):
+                raise ValueError("original source host changed")
+            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)
+            gid, aid = route["goal_id"], route["agent_id"]
+            if goal_ids is not None and gid not in goal_ids:
+                continue
+            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
+    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.
+    # 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,
+                _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"]:
+                            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")
+                        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", "")}
+                        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
+        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.
+    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"]]
@@ -138,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,