fix(chat): avoid duplicate steward inventory identity probes

loopx/capabilities/manager_context/README.md

@@ -189,13 +189,22 @@ cancellation or completion. The App's warning glyph exposes that cause on demand
 other unreadable sources remain unknown. Refresh rechecks the original owners;
 the read neither repairs historical receipts nor grants execution/return authority.
 
+For native-bound steward inventory, one fresh provider selection supplies both
+Session validation and Goal visibility at each opening/closing check. The read
+still revalidates before returning content; no observation is cached between
+checks or requests. Custom scope resolvers keep their existing validation path.
+
 管家标题栏提供常驻委派入口，按接收方查看待跟进请求，并可展开当前交办内容、关联
 工作和原对话。同一来源的较早会话不会因重开会话而丢失；其它用户、来源、群聊和
 项目会话不混入。结论已回传与成果已验收不是同一事实，历史记录仍可回看。
 来源与原路匹配但发送回执冲突时，保留不完整计数，以警示小符号按需提供至多 12 条
 具体请求身份和原因；不展示未获读取资格的交办正文，不推定已取消或完成。刷新重新
 核验原 owner，不修改历史回执，也不解除执行或返回权限；其它读取缺口继续保持未知。
 
+原生绑定管家的每次读取，在开始和结束时各验证一次 provider 身份，同一检查点的
+会话和 Goal 范围复用该次选择；检查点及请求之间不缓存权限。自定义范围解析器沿用
+原验证路径。
+
 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
 brief and independently assess its authority, evidence and constraints. The

loopx/chat_server.py

@@ -95,7 +95,6 @@
 )
 from .extensions.lark.goal_topic_connections import list_lark_apps
 from .extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
-from .extensions.lark.manager_routing import authorized_manager_goal_ids
 from .extensions.lark.cli_transport import CommandRunner
 from .extensions.runtime import (
     default_extension_state_file,
@@ -1909,18 +1908,9 @@ def serve_chat(
     )
     server.chat_store = ChatSessionStore(runtime_root)
     server.action_store = ChatActionStore(runtime_root / "chat" / "actions")
-    def manager_scope(session: dict[str, object]) -> list[str] | None:
-        if isinstance(session.get("steward_context"), dict):
-            bindings = server.runtime_controller.project_contexts.conversation_bindings
-            if bindings is None:
-                raise ValueError("Steward conversation bindings are unavailable")
-            return cast(list[str] | None, bindings.steward_scope(session))
-        return authorized_manager_goal_ids(
-            build_lark_goal_topic_runtime_snapshot(
-                registry_path=server.registry_path, runtime_root_override=server.runtime_root_override,
-            ), session, runtime_root=server.runtime_controller.coordination_runtime_root,
-            registry_path=resolved_registry_path, executor_runtime_root=runtime_root,
-        )
+    from .presentation.request_control_api import bound_steward_inventory_scope, manager_source_scope_resolver
+    manager_scope = manager_source_scope_resolver(server,
+        registry_path=resolved_registry_path, executor_runtime_root=runtime_root)
 
     server.runtime_controller = ChatRuntimeController(
         store=server.chat_store,
@@ -1935,6 +1925,8 @@ def manager_scope(session: dict[str, object]) -> list[str] | None:
         idle_timeout_sec=idle_timeout_sec,
         hard_timeout_sec=hard_timeout_sec,
     )
+    server.request_inventory_scope_resolver = bound_steward_inventory_scope(
+        server.runtime_controller, manager_scope_resolver=manager_scope)
     from .capabilities.native_chat.transports import install_conversation_transports
     install_conversation_transports(server, factories=external_conversation_factories,
         observe_group=lambda profile: observe_lark_conversation_identity(profile=profile, runner=server.lark_runner,

loopx/presentation/request_control_api.py

@@ -1,6 +1,7 @@
 """Source-owner request controls on the conversation that created the request."""
 from __future__ import annotations
 
+from collections.abc import Callable
 from pathlib import Path
 from typing import Any
 from urllib.parse import parse_qs, urlparse
@@ -35,6 +36,49 @@ def _operator_payload(value: Any) -> Any:
     return value
 
 
+def bound_steward_inventory_scope(
+    controller: Any, *, manager_scope_resolver: Callable[[dict[str, Any]], list[str] | None],
+) -> Callable[[dict[str, Any]], list[str] | None]:
+    """Reuse one fresh native selection for Session validation and Goal scope.
+
+    The server supplies its ordinary binding resolver. A later custom resolver
+    keeps the compatibility path; no provider observation survives this call.
+    """
+    def read(session):
+        if controller.manager_scope_resolver is not manager_scope_resolver:
+            return None
+        bindings = controller.project_contexts.conversation_bindings
+        if bindings is None:
+            raise ValueError("bound steward authority is unavailable")
+        selected = bindings.session_context(session["steward_context"], session=session)
+        controller.project_contexts.runtime_context(session, selected)
+        return list(selected["context"]["goal_ids"])
+    return read
+
+
+def manager_source_scope_resolver(
+    server: Any, *, registry_path: Path, executor_runtime_root: Path,
+) -> Callable[[dict[str, Any]], list[str] | None]:
+    """Keep the default manager's native and legacy source routing together."""
+    from ..chat_lark_api import build_lark_goal_topic_runtime_snapshot
+    from ..extensions.lark.manager_routing import authorized_manager_goal_ids
+
+    def read(session):
+        controller = server.runtime_controller
+        if isinstance(session.get("steward_context"), dict):
+            bindings = controller.project_contexts.conversation_bindings
+            if bindings is None:
+                raise ValueError("Steward conversation bindings are unavailable")
+            return bindings.steward_scope(session)
+        return authorized_manager_goal_ids(
+            build_lark_goal_topic_runtime_snapshot(
+                registry_path=server.registry_path, runtime_root_override=server.runtime_root_override,
+            ), session, runtime_root=controller.coordination_runtime_root,
+            registry_path=registry_path, executor_runtime_root=executor_runtime_root,
+        )
+    return read
+
+
 class RequestControlMixin:
     server: Any
     path: str
@@ -58,8 +102,12 @@ def _request_inventory_scope(self, session_id):
         scope = conversation_scope(session)
         if scope["kind"] not in {"owner_portfolio", "owner_goal"} and not scope.get("bound_steward"):
             raise ValueError("request inventory requires an authorized steward conversation")
+        verified_goals = None
         if session.get("steward_context") is not None:
-            controller.project_contexts.session_context(session)
+            read_scope = getattr(self.server, "request_inventory_scope_resolver", None)
+            verified_goals = read_scope(session) if read_scope is not None else None
+            if verified_goals is None:
+                controller.project_contexts.session_context(session)
         selected = getattr(self.server, "selected_goal_id", None)
         if selected and scope.get("goal_ids") != [selected]:
             raise ValueError("request inventory is outside this Goal workspace")
@@ -68,7 +116,7 @@ def _request_inventory_scope(self, session_id):
         else:
             if controller.manager_scope_resolver is None:
                 raise ValueError("current steward read scope is unavailable")
-            goals = controller.manager_scope_resolver(session)
+            goals = verified_goals if verified_goals is not None else controller.manager_scope_resolver(session)
             if not isinstance(goals, list):
                 raise ValueError("current steward read scope is unavailable")
             core = evidence_goal_scope(controller.coordination_runtime_root, session["channel_id"],

tests/test_native_steward_ingress.py

@@ -16,11 +16,12 @@ def ingress_path(store, row):
 
 
 @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
+@pytest.mark.parametrize("revoke", ["binding", "provider", "scope"])
+@pytest.mark.parametrize("scoped_reader", [False, True])
+def test_request_inventory_keeps_native_source_scope_and_holds_revoked_binding(steward, monkeypatch, revoke_during_read, revoke, scoped_reader):  # noqa: F811
     from types import SimpleNamespace
     from urllib.parse import urlencode
-    from loopx.presentation.request_control_api import RequestControlMixin
+    from loopx.presentation.request_control_api import RequestControlMixin, bound_steward_inventory_scope
 
     store, runtime, provider, transport, binding, capture, workspace = steward
     target = {"goal_id": "registered-work", "agent_id": "worker"}
@@ -54,16 +55,28 @@ def read(self):
             return self.result
 
     reader = Reader()
+    if scoped_reader:
+        reader.server.request_inventory_scope_resolver = bound_steward_inventory_scope(
+            runtime, manager_scope_resolver=runtime.manager_scope_resolver)
+    observations = []
+    observe = transport.bindings._observe_owner
+    def observe_owner(ref):
+        observations.append(ref)
+        return observe(ref)
+    monkeypatch.setattr(transport.bindings, "_observe_owner", observe_owner)
     code, value = reader.read()
     assert code == 200 and value["pending_count"] == 1, value
     assert value["rows"][0]["request_id"] == receipt["request_id"]
+    assert len(observations) == (2 if scoped_reader else 4)
     # A fresh source read cannot turn a revoked personal binding into local-owner
     # scope, nor use cached brief content after the authorization disappears.
     def withdraw_scope():
         if revoke == "binding":
             transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"])
-        else:
+        elif revoke == "provider":
             provider.owner_ready = False
+        else:
+            runtime.manager_scope_resolver = lambda _: []
     if revoke_during_read:
         from loopx.control_plane.collaboration import conversation_inventory
         read_requests = conversation_inventory.read_conversation_requests
@@ -75,10 +88,146 @@ def read_then_revoke(*args, **kwargs):
     else:
         withdraw_scope()
     code, refused = reader.read()
-    assert code == 400 and "rows" not in refused
+    if revoke == "scope" and not revoke_during_read:
+        assert code == 200 and refused["rows"] == [] and refused["pending_count"] == 0
+    else:
+        assert code == 400 and "rows" not in refused
     assert (capture.read_bytes() if capture.exists() else b"") == calls
 
 
+@pytest.mark.parametrize("backend", ["file", "sqlite"])
+def test_native_inventory_http_keeps_long_history_and_fresh_provider_checks(steward, monkeypatch, backend):  # noqa: F811
+    import hashlib
+    import http.client
+    import threading
+    from urllib.parse import urlencode
+    from tests.control_plane.canonical_authority_fixture import initialize_canonical_authority
+    from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
+    from loopx.control_plane.collaboration.links import link
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.todos.mutation_api import add_goal_todo
+    from loopx.extensions.lark.conversation_identity import identity_ref
+    from loopx.presentation.request_control_api import bound_steward_inventory_scope, manager_source_scope_resolver
+
+    store, runtime, provider, transport, binding, capture, workspace = steward
+    goal = "registered-work"
+    state = workspace / "ACTIVE_GOAL_STATE.md"
+    state.write_text("# Goal\n\n## User Todo\n\n## Agent Todo\n\n## Completed Work Archive\n")
+    root = runtime.coordination_runtime_root
+    registry = json.loads(runtime.registry_path.read_text())
+    registry["common_runtime_root"] = str(root)
+    registry["goals"] = [{"id": goal, "repo": str(workspace), "state_file": str(state),
+        "coordination": {"registered_agents": ["worker", "reviewer"]}}]
+    runtime.registry_path.write_text(json.dumps(registry))
+    initialize_canonical_authority(root, goal, build_todo_runtime_shadow_projection(goal_id=goal, todos=[]),
+        state_path=state, provider=backend)
+    tasks = {}
+    for agent in ("worker", "reviewer"):
+        created = add_goal_todo(registry_path=runtime.registry_path, runtime_root_arg=str(root), goal_id=goal,
+            role="agent", text=f"Inspect the retained artifact for {agent}", claimed_by=agent)
+        assert created["ok"] is True
+        tasks[agent] = created["todo_id"]
+    for number in range(43):
+        event = provider.event("steward-app", f"history-{number}",
+            f"Record {number}: " + "保留来源身份、当前版本和核验结果。" * 100)
+        transport.admit("steward-app", event)
+        accepted = transport.core.read_request(identity_ref(binding["provider_ref"], event["message_id"]))
+        finish(runtime, accepted)
+        session = store.load_session(accepted["session_id"])
+        turn = store.load_turn(accepted["session_id"], accepted["turn_id"])
+        if number == 0:
+            configure_evidence_scope(root, runtime.registry_path, channel=session["channel_id"], goal_ids=[goal], execute=True)
+            for agent in ("worker", "reviewer"):
+                configure_delivery_target(root, runtime.registry_path, channel=session["channel_id"],
+                    goal_id=goal, agent_id=agent, grant=True, execute=True)
+        agent = "reviewer" if number % 2 else "worker"
+        delivered = deliver(root, runtime.registry_path, session=session, turn=turn,
+            request={"goal_id": goal, "agent_id": agent}, source_store=store)
+        if number < 2:
+            link(root, runtime.registry_path, goal, agent, delivered["request_id"], [tasks[agent]], [])
+
+    sources = {p: hashlib.sha256(p.read_bytes()).hexdigest() for p in _root(root).rglob("*.json")}
+    model_calls = capture.read_bytes() if capture.exists() else b""
+    server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
+    server.registry_path, server.runtime_root, server.chat_store = runtime.registry_path, store.root.parent, store
+    server.runtime_root_override, server.runtime_controller = str(store.root.parent), runtime
+    server.selected_goal_id, server.verbose = None, False
+    runtime.manager_scope_resolver = manager_source_scope_resolver(server,
+        registry_path=runtime.registry_path, executor_runtime_root=store.root.parent)
+    server.request_inventory_scope_resolver = bound_steward_inventory_scope(
+        runtime, manager_scope_resolver=runtime.manager_scope_resolver)
+    observations = []
+    observe = transport.bindings._observe_owner
+    def observe_owner(ref):
+        observations.append(ref)
+        return observe(ref)
+    monkeypatch.setattr(transport.bindings, "_observe_owner", observe_owner)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    def read(**query):
+        connection = http.client.HTTPConnection("127.0.0.1", server.server_port, timeout=30)
+        before = len(observations)
+        connection.request("GET", "/api/chat/request-inventory?" + urlencode({"session_id": session["session_id"], **query}))
+        response = connection.getresponse()
+        body = json.loads(response.read())
+        connection.close()
+        return response.status, body, len(observations) - before
+    try:
+        offset, ids, snapshot, linked = 0, [], None, []
+        while offset is not None:
+            code, page, checks = read(view="history", offset=offset)
+            assert code == 200 and checks == 2 and page["matched"] == page["pending_count"] == 43
+            assert page["coverage"] == {"scan_complete": True, "unreadable": 0, "diagnostics": []}
+            assert all(len(row["source_question"]) > 1000 for row in page["rows"])
+            snapshot = snapshot or page["snapshot_digest"]
+            assert page["snapshot_digest"] == snapshot
+            ids.extend(row["request_id"] for row in page["rows"])
+            linked.extend(todo for row in page["rows"] for todo in row["linked_todos"])
+            offset = page["next_offset"]
+        assert len(ids) == len(set(ids)) == 43
+        assert {todo["todo_id"] for todo in linked} == set(tasks.values())
+        assert all(todo["status"] == "open" and "Inspect the retained artifact" in todo["title"] for todo in linked)
+        assert {row["agent_id"] for row in page["rows"]} <= {"worker", "reviewer"}
+        code, searched, checks = read(view="history", search=ids[-1])
+        assert code == 200 and checks == 2 and searched["matched"] == 1
+        provider.owner_ready = False
+        code, refused, _ = read(view="history")
+        assert code == 400 and "rows" not in refused and "保留来源身份" not in str(refused)
+        provider.owner_ready = True
+        assert read(view="history")[1]["matched"] == 43
+        assert {p: hashlib.sha256(p.read_bytes()).hexdigest() for p in sources} == sources
+        assert (capture.read_bytes() if capture.exists() else b"") == model_calls
+    finally:
+        server.shutdown()
+        thread.join(5)
+        server.server_close()
+
+
+def test_manager_source_scope_keeps_source_and_executor_routes_distinct(monkeypatch, tmp_path):
+    from types import SimpleNamespace
+    from loopx import chat_lark_api
+    from loopx.extensions.lark import manager_routing
+    from loopx.presentation.request_control_api import manager_source_scope_resolver
+
+    source, frozen_registry, executor = tmp_path / "source", tmp_path / "registry", tmp_path / "executor"
+    server = SimpleNamespace(registry_path=source, runtime_root_override="current-source-override",
+        runtime_controller=SimpleNamespace(coordination_runtime_root=tmp_path / "coordination"))
+    observed = []
+    def snapshot(**kwargs):
+        observed.append(kwargs)
+        return {"source": "current"}
+    def authorize(value, session, **kwargs):
+        assert value == {"source": "current"} and session == {"channel_id": "legacy-manager"}
+        observed.append(kwargs)
+        return ["permitted-goal"]
+    monkeypatch.setattr(chat_lark_api, "build_lark_goal_topic_runtime_snapshot", snapshot)
+    monkeypatch.setattr(manager_routing, "authorized_manager_goal_ids", authorize)
+    read = manager_source_scope_resolver(server, registry_path=frozen_registry, executor_runtime_root=executor)
+    assert read({"channel_id": "legacy-manager"}) == ["permitted-goal"]
+    assert observed == [{"registry_path": source, "runtime_root_override": "current-source-override"}, {
+        "runtime_root": tmp_path / "coordination", "registry_path": frozen_registry, "executor_runtime_root": executor}]
+
+
 @pytest.mark.parametrize("separate_coordination_root", [False, True])
 def test_native_ingress_uses_coordination_owner_with_legacy_fallback(steward, monkeypatch, tmp_path, separate_coordination_root):  # noqa: F811
     store, runtime, provider, transport, binding, _, workspace = steward