fix(chat): read original workbench state from its registered authority

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

@@ -176,6 +176,15 @@ discarding it. Developer isolation remains explicit. Browser acceptance against
 one Bot service does not qualify the native App's actual service route. This
 checkpoint remains open until that native path is implemented and read back.
 
+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
+remain incomplete coverage rather than an empty successful inventory.
+
 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

@@ -122,6 +122,12 @@ Goal 范围、项目、明确原会话、受管 Agent 以及远端/只读视图
 用于明确恢复，不擅自混入或删除；开发隔离须显式选择。对某个 Bot 服务做浏览器
 验收不能证明实际原生 App 路由正确。此项保持未闭，直至原生路径实现并独立读回。
 
+Chat 存储覆盖不另选注册工作的权威根目录。工作区状态与交付任务图沿用协调器
+既有的注册 runtime root，私有 Chat 历史与设置仍留在原存储。原会话委派列表通过
+既有 Effect 请求范围为每次 HTTP 读取固定一个已核验代码修订；这不缓存来源授权或
+回执决策，各来源与最终调用者范围仍重新核验，下次读取独立核验代码修订。
+来源不可读继续显示覆盖不完整，不能当作成功的空列表。
+
 冷启动 Goal 深链接时，先等待当前状态目录，再判断所选 Goal 是否已失效；内置
 占位数据不能证明有效 Goal 已被删除，初始 Goal 路由因此不会误入默认管家发现。
 

loopx/chat_status_api.py

@@ -53,8 +53,15 @@
 from .control_plane.todos.list_readback import list_goal_todos
 
 
+def _status_runtime_root(server: Any) -> str | None:
+    """Registered work follows the same authority as the Chat coordinator."""
+    controller = getattr(server, "runtime_controller", None)
+    root = getattr(controller, "coordination_runtime_root", None)
+    return str(root) if root is not None else server.runtime_root_override
+
+
 def _goal_task_map(server: Any, goal_id: str) -> dict[str, Any] | None:
-    runtime_root = server.runtime_root_override
+    runtime_root = _status_runtime_root(server)
     source_read_at = now_utc_iso()
     try:
         listed = list_goal_todos(
@@ -198,6 +205,7 @@ def _status(self, *, delivery_review: bool = False) -> None:
             )
             return
         try:
+            status_runtime_root = _status_runtime_root(self.server)
             registry = {}
             if views is not None or requested_goals is not None or delivery_review:
                 # Discovery reads only metadata. Full status/runtime reads keep
@@ -249,7 +257,7 @@ def _status(self, *, delivery_review: bool = False) -> None:
                 # empty workspace (serve_chat); its status reports the same
                 # fact as a warning, so a fresh install is not a failed source.
                 allow_missing_registry=True,
-                runtime_root_override=self.server.runtime_root_override,
+                runtime_root_override=status_runtime_root,
                 limit=self.server.limit,
                 goal_id=goal_id,
                 include_zcode_goal_eligibility=not delivery_review,
@@ -329,6 +337,7 @@ def _status(self, *, delivery_review: bool = False) -> None:
                 self.server.registry_path,
                 *self.server.scan_roots,
                 self.server.runtime_root,
+                *([status_runtime_root] if status_runtime_root else []),
             ]
             projection = json.loads(
                 redact_local_paths(

loopx/presentation/request_control_api.py

@@ -12,7 +12,7 @@
 from ..control_plane.collaboration.inbox import _entry, _read, _root
 from ..control_plane.collaboration.request_amendment import control_request
 from ..control_plane.collaboration.source_chat_observation import is_source_chat_host
-from ..control_plane.effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError
+from ..control_plane.effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError, effect_runtime_request_scope
 from ..control_plane.goals.goal_ref_validation import exact_goal_ref
 from ..control_plane.projects.registry_codec import load_project_registry
 from ..capabilities.manager_context import target_authority
@@ -75,6 +75,7 @@ def _request_inventory_scope(self, session_id):
                 registry_path=self.server.registry_path) or []
         return session, goals, core
 
+    @effect_runtime_request_scope()
     def _request_inventory_read(self) -> None:
         from ..control_plane.collaboration.conversation_inventory import read_conversation_requests
 

tests/control_plane/test_chat_request_inventory.py

@@ -15,6 +15,28 @@ def _inventory(api, session_id, **query):
     return call(path="/api/chat/request-inventory?" + urlencode({"session_id": session_id, **query}))
 
 
+def test_inventory_pins_code_revision_but_rereads_original_source(api, monkeypatch):
+    from loopx.control_plane import effect_runtime
+
+    _, _, route = _source(api)
+    _, fixture, _ = api
+    original = effect_runtime._runtime_fingerprint
+    scans = []
+    def fingerprint():
+        scans.append(True)
+        return original()
+    monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", fingerprint)
+    first = _inventory(api, route["session_id"])
+    assert first[0] == 200 and first[1]["pending_count"] == 1
+    assert len(scans) == 1
+    source = _root(fixture["runtime"]) / "roundtrips" / (route["request_id"] + ".json")
+    source.write_text("{damaged")
+    second = _inventory(api, route["session_id"])
+    assert second[0] == 200 and second[1]["coverage"]["unreadable"] == 1
+    assert second[1]["pending_count"] == 0
+    assert len(scans) == 2, "Each HTTP read must qualify its own runtime revision"
+
+
 def test_read_and_adoption_still_owe_conclusion_and_queued_return_stays_visible(api):
     _, _, route = _source(api)
     _, f, server = api

tests/control_plane/test_chat_workbench_roots.py

@@ -0,0 +1,59 @@
+"""Chat storage is not another registered work/status authority."""
+from types import SimpleNamespace
+import json
+
+import pytest
+
+from canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+from loopx.chat_runtime import ChatRuntimeController
+from loopx.chat_status_api import ChatStatusRequestMixin
+from loopx.chat_store import ChatSessionStore
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("registered_root", [True, False])
+def test_status_reads_registered_work_while_chat_keeps_its_own_store(tmp_path, monkeypatch, provider, registered_root):
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, common, _ = promoted_create_fixture(tmp_path, provider=provider)
+    if not registered_root:
+        configuration = json.loads(registry.read_text())
+        del configuration["common_runtime_root"]
+        registry.write_text(json.dumps(configuration))
+    chat_root = tmp_path / "private-chat" if registered_root else common
+    store = ChatSessionStore(chat_root)
+    runtime = ChatRuntimeController(store=store, registry_path=registry, codex_bin="must-not-run")
+    session = store.create_session(goal_id="loopx-manager", agent_id="codex",
+        adapter_kind="codex_app_server", upstream_thread_id="original", channel_id="manager")
+    store.append_message(session["session_id"], role="agent", text="Keep the original private history.")
+    before = {path: path.read_bytes() for path in store.root.rglob("*.json")}
+    # External host activity is an optional observer, not the status authority.
+    monkeypatch.setattr("loopx.chat_status_api.attach_host_thread_activity", lambda *a, **k: None)
+    monkeypatch.setattr("loopx.chat_status_api.attach_host_delivery_windows", lambda *a, **k: None)
+
+    class Reader(ChatStatusRequestMixin):
+        server = SimpleNamespace(registry_path=registry, runtime_root=store.root.parent,
+            runtime_root_override=str(store.root.parent), runtime_controller=runtime,
+            selected_goal_id=None, scan_roots=[tmp_path / "project"], limit=80,
+            goal_subagent_configuration_enabled=False)
+        path = "/status.json?goal_id=goal-a"
+
+        def _send_json(self, value, *, status=200):
+            self.result = status, value
+
+        def _send_error(self, message, **kwargs):
+            self.result = kwargs.get("status", 500), {"error": message, **kwargs}
+
+    try:
+        assert runtime.coordination_runtime_root == common
+        reader = Reader()
+        reader._status()
+        code, result = reader.result
+        assert code == 200, result
+        assert not any("canonical authority" in error.casefold() for error in result.get("contract_errors", []))
+        assert result["run_history"]["goals"][0]["id"] == "goal-a"
+        assert str(common) not in json.dumps(result)
+        assert {path: path.read_bytes() for path in before} == before
+        assert runtime.store.root == chat_root / "chat"
+    finally:
+        runtime.close()