fix(chat): bound original Turn lookup in delegation inventory

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -191,6 +191,14 @@ code revision. This bounds provider IO by sources instead of historical rows,
 without caching receipt decisions or reducing scan coverage. Unreadable sources
 remain incomplete coverage rather than an empty successful inventory.
 
+Within each HTTP read, the Chat source observer locates original client Turns once per
+verified source Session, then rereads the exact Turn file for every request.
+This Python IO locator retains only store paths: it caches no content, grants
+or receipt decisions and is discarded before the next page. A missing Turn or
+changed client identity remains unreadable source coverage. This removes repeated
+full-history scans without changing the typed collaboration owner or its checks;
+provider latency and full installed cross-channel acceptance remain separate.
+
 Cold Goal links wait for the current status directory before invalidating a
 missing selection; the bundled placeholder is not proof that a valid Goal was
 removed. This keeps an initial Goal route out of default-steward discovery.

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -132,6 +132,12 @@ Chat 存储覆盖不另选注册工作的权威根目录。工作区状态与交
 范围和代码修订。provider IO 因而随来源数而非历史记录数增长，不缓存回执决策、
 不缩小扫描覆盖。来源不可读继续显示覆盖不完整，不能当作成功的空列表。
 
+每次 HTTP 读取内，Chat 来源观测器对每个已核验来源 Session 只定位一次原 client Turn，
+随后每条请求仍重新读取确切 Turn 文件。此 Python IO 定位器只保留 store 路径，
+不缓存正文、授权或回执决策，下一页重新构建。Turn 丢失或 client 身份变化仍计为
+来源覆盖不可读。这样消除重复全历史扫描，不改变 typed 协作 owner 及其核验；
+provider 延迟与完整安装态跨端验收仍是独立边界。
+
 冷启动 Goal 深链接时，先等待当前状态目录，再判断所选 Goal 是否已失效；内置
 占位数据不能证明有效 Goal 已被删除，初始 Goal 路由因此不会误入默认管家发现。
 

loopx/control_plane/collaboration/conversation_inventory.py

@@ -14,7 +14,7 @@
 from .goal_instance_scope import collaboration_goal_scope, decide_collaboration_lifecycle
 from .inbox import _entry, _now, _read, _receipt, _request_lock, _root, read_inbox_transaction, reply_status
 from .links import read_linked_work
-from .source_chat_observation import is_source_chat_host, original_delivery_facts
+from .source_chat_observation import is_source_chat_host, original_delivery_facts, original_turn_reader
 from .source_grant_observation import source_context_target_authority
 
 
@@ -45,7 +45,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 = [], 0, {}
+    rows, unreadable, cache, turn_readers = [], 0, {}, {}
     for path in paths[:2000]:
         try:
             candidate = _read(path)
@@ -67,7 +67,10 @@ 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
-            turn = store.turn_for_client(source_session["session_id"], route["client_turn_id"])
+            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:

loopx/control_plane/collaboration/source_chat_observation.py

@@ -4,6 +4,7 @@
 with their existing typed owners. No model or peer supplies this address.
 """
 
+from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 
@@ -24,6 +25,31 @@ def is_source_chat_host(coordination_root: Path, route: dict[str, Any], store: C
     return _source_chat_root(coordination_root, route) == store.root.parent.resolve()
 
 
+def original_turn_reader(store: ChatSessionStore, session_id: str) -> Callable[[str], dict[str, Any] | None]:
+    """Locate a batch within one read; reread each original Turn on use.
+
+    Keep only trusted store paths, not Turn contents or authorization. The
+    caller discards this locator after its read, including between pages.
+    """
+    turns_dir = store._session_dir(session_id) / "turns"
+    paths: dict[str, Path] = {}
+    for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []:
+        if path.name.endswith(".events.json"):
+            continue
+        client_id = _read_json(path).get("client_turn_id")
+        if isinstance(client_id, str):
+            paths.setdefault(client_id, path)
+
+    def read(client_turn_id: str) -> dict[str, Any] | None:
+        path = paths.get(client_turn_id)
+        if path is None:
+            return None
+        payload = _read_json(path)
+        return payload if payload.get("client_turn_id") == client_turn_id else None
+
+    return read
+
+
 def observe_source_chat(coordination_root: Path, route: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
     source_root = _source_chat_root(coordination_root, route)
     session_id = _opaque_id(route.get("session_id"), field="session_id")

tests/control_plane/test_chat_request_inventory.py

@@ -122,6 +122,8 @@ def observe_scope(reader, session_id):
 
 
 def test_paging_and_unreadable_source_do_not_report_empty_or_mutate_receipts(api, monkeypatch):
+    import loopx.chat_store as chat_store
+    from loopx.control_plane.collaboration import source_chat_observation
     from loopx.presentation.request_control_api import RequestControlMixin
 
     _, _, route = _source(api)
@@ -140,8 +142,17 @@ def observe_scope(reader, session_id):
         verified.append(session_id)
         return scope_read(reader, session_id)
     monkeypatch.setattr(RequestControlMixin, "_request_inventory_scope", observe_scope)
+    read_json = source_chat_observation._read_json
+    turn_reads = []
+    def counted_read(path):
+        if path.parent.name == "turns" and not path.name.endswith(".events.json"):
+            turn_reads.append(path)
+        return read_json(path)
+    monkeypatch.setattr(source_chat_observation, "_read_json", counted_read)
+    monkeypatch.setattr(chat_store, "_read_json", counted_read)
     code, first = _inventory(api, route["session_id"])
     assert code == 200 and first["pending_count"] == 13 and len(first["rows"]) == 12, first
+    assert len(turn_reads) <= 3 * 13, "One inventory must not rescan full Turn history for every request"
     assert len(verified) == 2, "Provider scope verification must not grow with rows from one source"
     second = _inventory(api, route["session_id"], offset=first["next_offset"])[1]
     assert len(verified) == 4, "A new HTTP read must verify the source independently"
@@ -176,3 +187,43 @@ def observe_scope(reader, session_id):
     assert missing["coverage"]["unreadable"] == 1 and missing["pending_count"] == 12
     source.write_bytes(saved)
     assert _inventory(api, route["session_id"], offset=-1)[0] == 400
+
+
+def test_turn_locator_rereads_content_and_rejects_lost_identity(api):
+    import json
+    from loopx.control_plane.collaboration.source_chat_observation import original_turn_reader
+
+    _, _, route = _source(api)
+    _, _, server = api
+    store = server.chat_store
+    read = original_turn_reader(store, route["session_id"])
+    original = read("request-source")
+    assert original["turn_id"] == route["turn_id"]
+    response = {**original["response"], "message": "Current result"}
+    store.update_turn(route["session_id"], route["turn_id"], response=response)
+    assert read("request-source")["response"] == response
+    path = store._turn_path(route["session_id"], route["turn_id"])
+    current = path.read_bytes()
+    path.write_text(json.dumps({**original, "client_turn_id": "different-source"}))
+    assert read("request-source") is None
+    assert _inventory(api, route["session_id"])[1]["coverage"]["unreadable"] == 1
+    path.unlink()
+    assert read("request-source") is None
+    path.write_bytes(current)
+    assert _inventory(api, route["session_id"])[1]["pending_count"] == 1
+
+
+def test_inventory_withholds_all_content_when_closing_source_scope_changes(api, monkeypatch):
+    from loopx.presentation.request_control_api import RequestControlMixin
+
+    _, _, route = _source(api)
+    scope_read = RequestControlMixin._request_inventory_scope
+    verified = []
+    def observe_scope(reader, session_id):
+        observed, goals, core = scope_read(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"])
+    assert code == 400 and denied["error_code"] == "request_inventory_unavailable"
+    assert "rows" not in denied and "Check the current contract" not in str(denied)