From 6437299456be03ba4890c762a082526f22134a4c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 26 Oct 2026 02:22:20 +0800 Subject: [PATCH] fix(chat): bound inventory verification by original sources Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 12 +++++++--- ...p-conversation-and-async-inbox-v0.zh-CN.md | 10 +++++--- loopx/presentation/request_control_api.py | 23 +++++++++++++----- .../test_chat_request_inventory.py | 24 +++++++++++++++++-- tests/test_native_steward_ingress.py | 20 ++++++++++++++-- 5 files changed, 73 insertions(+), 16 deletions(-) diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index e3a83cb0c5..c2ecb0ff53 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index 3c3a178fb2..d2dc292240 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/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 路由因此不会误入默认管家发现。 diff --git a/loopx/presentation/request_control_api.py b/loopx/presentation/request_control_api.py index d1d144d4f6..ee51b0ef3a 100644 --- a/loopx/presentation/request_control_api.py +++ b/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") diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index e32f3168c0..f80706f231 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/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"]} diff --git a/tests/test_native_steward_ingress.py b/tests/test_native_steward_ingress.py index 68b77f597c..a500a3860a 100644 --- a/tests/test_native_steward_ingress.py +++ b/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