From 0d470c8201d0661b7c8612d8e9589d9472759fe1 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 20 Oct 2026 02:16:20 +0800 Subject: [PATCH] fix(chat): keep recovery discovery free of provider identity IO Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 5 +++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 4 +++ .../native_chat/external_conversations.py | 18 +++++++--- tests/test_native_steward_private.py | 36 +++++++++++++++++++ 4 files changed, 59 insertions(+), 4 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 1ea36de886..085aca0fce 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md @@ -1313,6 +1313,11 @@ the complete visible user/assistant history once in native persisted history; the old upstream identity remains recorded, and native Goal/attached executions are not replaced by this migration. +Discovery uses the local enabled binding and the same typed candidate planner, +without provider identity IO for every historical failure. This is a queue hint; +admission still re-resolves current App/owner/grant and exact source/native proof +under the existing fences before reserving or enqueueing any continuation. + Core may admit one correlated continuation per request for this migration only after current App/owner/grant and unchanged source verification, an exact failed native Turn readback, and a complete inventory containing only its original user 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 a72264cca6..0b9fe629f3 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 @@ -832,6 +832,10 @@ message，仅以该 App 身份下载属于这条消息的资源，再将 PNG/JPE 历史一次性恢复到原生持久历史中，保留旧上游身份；本次迁移不替换原生 Goal 或 attached 执行。 +发现阶段使用本地已启用绑定和同一 typed 候选规划器，不为每条历史失败做网络身份 +核验。这只是队列提示；受理仍须在原 fence 内重新解析当前 App、本人、grant， +核验确切原消息与原生凭证后，才能预留或排队任何续办。 + 只有重新核验当前 App、本人、grant 和未改变的原消息，并独立读回确切失败的原生 Turn，确认完整条目清单仅包含原用户输入后，Core 才能为本次格式迁移给每条请求 受理一次关联续办。策略或额度拒绝、追加纠正、工具、未知效果、关闭的 Session、 diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py index f9a8d02a65..23720e068c 100644 --- a/loopx/capabilities/native_chat/external_conversations.py +++ b/loopx/capabilities/native_chat/external_conversations.py @@ -46,7 +46,8 @@ def __init__(self, controller: Any) -> None: def _context_recovery_plan(self, row: dict[str, Any], *, observation: dict[str, Any] | None = None, - source_verified: bool = False) -> dict[str, Any]: + source_verified: bool = False, + discovery_only: bool = False) -> dict[str, Any]: from ...chat_manager import MANAGER_CONTEXT_VERSION from ...control_plane.effect_runtime import effect_runtime_result @@ -55,10 +56,19 @@ def _context_recovery_plan(self, row: dict[str, Any], *, turn = store.load_turn(str(row.get("session_id") or ""), str(row.get("turn_id") or "")) if not session or not turn: return {} - selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"]) + if discovery_only: + # A queue hint must not serialize provider identity IO across all + # historical failures. Reuse the same typed planner with the local + # enabled binding; admission independently resolves fresh authority. + selected_binding = next((item for item in self.bindings.read()["bindings"] + if item["binding_id"] == row["binding_id"] and item.get("enabled") is True), None) + if selected_binding is None: + return {} + else: + selected_binding = self.bindings.resolve(binding_id=row["binding_id"], **row["source"])["binding"] return effect_runtime_result("collaboration.conversation.request", { "request": {key: value for key, value in row.items() if key != "attachments"}, - "current_session": session, "binding": selected["binding"], + "current_session": session, "binding": selected_binding, "context_recovery": {"version": MANAGER_CONTEXT_VERSION, "failed_turn": turn, "steering_observed": any(event.get("kind") == "turn.steered" for event in store.events_after(session["session_id"], turn["turn_id"], None)), @@ -80,7 +90,7 @@ def context_recovery_candidate(self, request_ref: str) -> bool: turn = self.controller.store.load_turn(row["session_id"], row["turn_id"]) if not turn or turn.get("status") != "failed": return False - return self._context_recovery_plan(row).get("recovery_candidate") is True + return self._context_recovery_plan(row, discovery_only=True).get("recovery_candidate") is True except (KeyError, ValueError, RuntimeError, OSError): return False diff --git a/tests/test_native_steward_private.py b/tests/test_native_steward_private.py index 556c3e4ba9..fcc04c2bb5 100644 --- a/tests/test_native_steward_private.py +++ b/tests/test_native_steward_private.py @@ -74,6 +74,42 @@ def finish(runtime, row): return runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15) +@pytest.mark.parametrize("case", ["enabled", "disabled", "missing", "authority-unavailable"]) +def test_recovery_discovery_never_observes_provider_but_admission_still_does(steward, monkeypatch, case): + store, runtime, provider, transport, binding, capture, _ = steward + fake = Path(runtime.codex_bin) + source = fake.read_text().replace('"codexErrorInfo": "cyberPolicy"', '"codexErrorInfo": "contextWindowExceeded"') + source = source.replace(' print(json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": "durable-thread", "turnId": active_turn, "delta": "Partial answer."}}), flush=True)\n', '') + fake.write_text(source) + event = provider.event("steward-app", "discovery-only", "typed terminal failure") + transport.admit("steward-app", event) + ref = identity_ref(binding["provider_ref"], event["message_id"]) + row = transport.core.read_request(ref) + assert finish(runtime, row)["status"] == "failed" + store.update_session(row["session_id"], manager_context_version=20) + assert transport.reconcile() == 1 + if case == "disabled": + transport.bindings.disconnect(binding["binding_id"], expected_revision=transport.bindings.read()["revision"]) + elif case == "missing": + current = transport.bindings.read() + current["bindings"] = [] + transport.bindings.path.write_text(json.dumps(current)) + before_calls, before_requests, before_writes = len(provider.calls), capture.read_bytes(), list(provider.writes) + with monkeypatch.context() as discovery: + discovery.setattr(transport.bindings, "observe", lambda *_a, **_kw: pytest.fail("provider IO in discovery")) + assert transport.core.context_recovery_candidate(ref) is (case not in {"disabled", "missing"}) + assert len(provider.calls) == before_calls # Includes no hidden auth/status probe. + if case == "authority-unavailable": + provider.owner_ready = False + # A local hint is not permission: the normal admission owner must + # freshly reject unavailable authority before native observation/work. + with pytest.raises(ValueError): + transport.core.recover_failed_request(ref, source_observation=lambda _selected: pytest.fail("source IO after rejection")) + assert not transport.core.read_request(ref).get("context_recovery") + assert transport.core.read_request(ref)["turn_id"] == row["turn_id"] + assert capture.read_bytes() == before_requests and provider.writes == before_writes + + @pytest.mark.parametrize("fault", [None, "revoked", "source-changed", "native-effect", "unknown-native", "reservation-crash", "post-enqueue-crash", "transport-sync-crash", "existing-return", "foreign-native-thread", "wrong-native-turn", "policy-native-proof", "steered", "native-goal", "current-context"])