fix(chat): keep recovery discovery free of provider identity IO

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

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、

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
 

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"])