fix(chat): verify inventory scope per original source

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

@@ -180,9 +180,15 @@ Chat storage overrides do not select another registered-work authority. The
 workspace Status and delivery task map read the coordinator's existing registered
 runtime root, while private Chat history and its settings keep their own store.
 The original-conversation request inventory pins one qualified code revision per
-HTTP read through the existing Effect request scope. This does not cache source
-authorization or receipt decisions: each source and the final caller scope are
-still rechecked, and the next read qualifies its own revision. Unreadable sources
+HTTP read through the existing Effect request scope. Within that read, provider
+scope is verified once per distinct original Session, including older Sessions
+behind a reopened conversation, then every observed source is freshly verified
+again before any content is returned. Each row still checks its current source
+basis, host, Turn, target grant, Goal lifetime and receipt state. A revoked binding
+or provider identity at the closing check refuses the whole response. No proof
+survives the HTTP read; the next page independently qualifies source scope and
+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.
 
 Cold Goal links wait for the current status directory before invalidating a

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

@@ -124,9 +124,13 @@ Goal 范围、项目、明确原会话、受管 Agent 以及远端/只读视图
 
 Chat 存储覆盖不另选注册工作的权威根目录。工作区状态与交付任务图沿用协调器
 既有的注册 runtime root，私有 Chat 历史与设置仍留在原存储。原会话委派列表通过
-既有 Effect 请求范围为每次 HTTP 读取固定一个已核验代码修订；这不缓存来源授权或
-回执决策，各来源与最终调用者范围仍重新核验，下次读取独立核验代码修订。
-来源不可读继续显示覆盖不完整，不能当作成功的空列表。
+既有 Effect 请求范围为每次 HTTP 读取固定一个已核验代码修订。在一次读取内，
+每个不同的原 Session（包括重开会话背后的旧 Session）先独立核验 provider 范围，
+返回任何内容前再对所有已观察来源做一次新核验。每条记录仍检查当前来源依据、
+宿主、Turn、目标授权、Goal 生命周期和回执状态；结束核验发现 binding 或 provider
+身份被撤销时，整份响应拒绝返回。证明不跨 HTTP 读取保存，下一页独立核验来源
+范围和代码修订。provider IO 因而随来源数而非历史记录数增长，不缓存回执决策、
+不缩小扫描覆盖。来源不可读继续显示覆盖不完整，不能当作成功的空列表。
 
 冷启动 Goal 深链接时，先等待当前状态目录，再判断所选 Goal 是否已失效；内置
 占位数据不能证明有效 Goal 已被删除，初始 Goal 路由因此不会误入默认管家发现。

loopx/presentation/request_control_api.py

@@ -89,18 +89,29 @@ def _request_inventory_read(self) -> None:
                 raise ValueError("invalid request inventory view")
             session, goals, core = self._request_inventory_scope(query["session_id"][0])
             controller = self.server.runtime_controller
+            def source_basis(source):
+                return conversation_scope(source), source.get("steward_context")
+            # Provider identity belongs to a source, not to every historical row.
+            # Keep this proof within one read and fence every observed source
+            # again before returning any content. Receipt/grant IO stays live.
+            sources = {session["session_id"]: (source_basis(session), goals, core)}
             def validate_source(source):
-                observed, permitted, _ = self._request_inventory_scope(source["session_id"])
-                if observed["channel_id"] != session["channel_id"] or permitted != goals:
+                sid = source["session_id"]
+                if sid not in sources:
+                    observed, permitted, evidence = self._request_inventory_scope(sid)
+                    if observed["channel_id"] != session["channel_id"] or permitted != goals:
+                        raise ValueError("original conversation authorization changed")
+                    sources[sid] = source_basis(observed), permitted, evidence
+                if source_basis(source) != sources[sid][0]:
                     raise ValueError("original conversation authorization changed")
             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])
-            current, current_goals, current_core = self._request_inventory_scope(session["session_id"])
-            if (current["channel_id"], current.get("steward_context"), current_goals, current_core) != (
-                    session["channel_id"], session.get("steward_context"), goals, core):
-                raise ValueError("conversation authorization changed during read")
+            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:
+                    raise ValueError("conversation authorization changed during read")
             self._send_json(_operator_payload(result))
         except (OSError, ValueError, TypeError, KeyError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
             self._send_error(redact_local_paths(str(exc)), status=400, error_code="request_inventory_unavailable")

tests/control_plane/test_chat_request_inventory.py

@@ -87,15 +87,25 @@ def test_revision_and_cancellation_reuse_current_control_and_preserve_history(ap
     assert _inventory(api, route["session_id"], view="history", search="Corrected source")[1]["matched"] == 1
 
 
-def test_earlier_source_session_survives_reopen_and_other_audiences_are_excluded(api):
+def test_earlier_source_session_survives_reopen_and_other_audiences_are_excluded(api, monkeypatch):
+    from loopx.presentation.request_control_api import RequestControlMixin
+
     _, _, route = _source(api)
     _, f, server = api
     store = server.chat_store
     fresh = store.create_session(goal_id="loopx-manager", agent_id="codex", adapter_kind="codex_app_server",
         upstream_thread_id="new-steward-session", channel_id="manager")
+    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)
     code, value = _inventory(api, fresh["session_id"])
     assert code == 200 and value["pending_count"] == 1
     assert value["rows"][0]["source_session_id"] == route["session_id"]
+    assert verified.count(fresh["session_id"]) == 2
+    assert verified.count(route["session_id"]) == 2, "Older sources need their own closing authorization check"
     project = store.create_session(goal_id=f["ref"]["goal_id"], agent_id="codex", adapter_kind="codex_app_server",
         upstream_thread_id="project-owner", channel_id="goal." + f["ref"]["goal_id"])
     turn, _ = store.create_turn(project["session_id"], client_turn_id="other-conversation", message="Private project request", origin="web")
@@ -111,7 +121,9 @@ def test_earlier_source_session_survives_reopen_and_other_audiences_are_excluded
     assert _inventory(api, "missing-session")[0] == 400
 
 
-def test_paging_and_unreadable_source_do_not_report_empty_or_mutate_receipts(api):
+def test_paging_and_unreadable_source_do_not_report_empty_or_mutate_receipts(api, monkeypatch):
+    from loopx.presentation.request_control_api import RequestControlMixin
+
     _, _, route = _source(api)
     _, f, server = api
     session = server.chat_store.load_session(route["session_id"])
@@ -122,9 +134,17 @@ def test_paging_and_unreadable_source_do_not_report_empty_or_mutate_receipts(api
             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"])
+    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)
     code, first = _inventory(api, route["session_id"])
     assert code == 200 and first["pending_count"] == 13 and len(first["rows"]) == 12, first
+    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"
     assert len(second["rows"]) == 1 and second["next_offset"] is None
     assert second["snapshot_digest"] == first["snapshot_digest"]
     assert not {row["request_id"] for row in first["rows"]} & {row["request_id"] for row in second["rows"]}

tests/test_native_steward_ingress.py

@@ -15,7 +15,9 @@ def ingress_path(store, row):
         _hash([row["session_id"], "external-" + row["request_ref"]]) + ".json")
 
 
-def test_request_inventory_keeps_native_source_scope_and_holds_revoked_binding(steward, monkeypatch):  # noqa: F811
+@pytest.mark.parametrize("revoke_during_read", [False, True])
+@pytest.mark.parametrize("revoke", ["binding", "provider"])
+def test_request_inventory_keeps_native_source_scope_and_holds_revoked_binding(steward, monkeypatch, revoke_during_read, revoke):  # noqa: F811
     from types import SimpleNamespace
     from urllib.parse import urlencode
     from loopx.presentation.request_control_api import RequestControlMixin
@@ -57,7 +59,21 @@ def read(self):
     assert value["rows"][0]["request_id"] == receipt["request_id"]
     # A fresh source read cannot turn a revoked personal binding into local-owner
     # scope, nor use cached brief content after the authorization disappears.
-    transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+    def withdraw_scope():
+        if revoke == "binding":
+            transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
+        else:
+            provider.owner_ready = False
+    if revoke_during_read:
+        from loopx.control_plane.collaboration import conversation_inventory
+        read_requests = conversation_inventory.read_conversation_requests
+        def read_then_revoke(*args, **kwargs):
+            result = read_requests(*args, **kwargs)
+            withdraw_scope()
+            return result
+        monkeypatch.setattr(conversation_inventory, "read_conversation_requests", read_then_revoke)
+    else:
+        withdraw_scope()
     code, refused = reader.read()
     assert code == 400 and "rows" not in refused
     assert (capture.read_bytes() if capture.exists() else b"") == calls