fix(chat): fence prepared recovery and reject in-flight binding changes

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

@@ -72,6 +72,11 @@ without private bindings do not add provider authentication; configured aliases
 reuse one request-scoped verified identity observation. Lark HTTP composition
 resides in the extension, while the typed binding owner remains provider-neutral.
 
+Binding resolution rereads the exact grant after provider identity observation.
+Revoking or replacing that binding during the observation rejects the request;
+an unrelated binding edit does not invalidate it. The typed owner still checks
+the current context, with no identity cache shared across verification calls.
+
 The same Settings read model includes Core bindings for locally installed
 conversation transports. Listener status comes from the exact host-composed
 provider's optional `health_snapshot()` returning a content-free `status`, or
@@ -128,7 +133,10 @@ transport resume can fail. Native observation may only narrow that frozen set;
 a retry cannot add later completed/active Turns or switch to a newer Session.
 The existing Lark listener returns after this durable freeze; its existing reply
 workers recover the prepared request to observe and cancel native background
-commands. Slow or unavailable native control therefore does not hold inbound
+commands. Prepared recovery uses admission's fresh authority check inside both
+the request and source-route fences, without a duplicate provider probe before
+waiting for those fences. Revocation while waiting prevents control and delivery.
+Slow or unavailable native control therefore does not hold inbound
 admission. This is scheduling only: it introduces no queue or completion state,
 does not defer an active Turn's exact interrupt, and sends no successful stop
 reply until native control and the original delivery checks have succeeded.

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

@@ -51,6 +51,9 @@ provider 读回确认回复；发生没有 receipt 的不确定写入时不盲
 时，同一请求内的别名检查与连接共用一次已验证身份观测。飞书 HTTP 组合位于
 extension，typed binding owner 继续保持 provider-neutral。
 
+绑定解析在 provider 身份观测后重读确切授权。观测期间撤销或替换该绑定会拒绝请求；
+其它绑定的修改不使当前绑定失效。typed owner 仍检查当前上下文，不跨次缓存身份核验。
+
 同一设置读模型也展示本机已安装 transport 的 Core binding。监听状态取确切宿主组合
 provider 的可选 `health_snapshot()`，返回仅含状态的观测；默认 profile 使用已有
 Lark listener 观测。该本地 hook 只读取缓存，不发起网络请求。仅投影已有监听标签
@@ -89,7 +92,9 @@ App 身份时，并行读取两条原 provider 消息。两项只读观测全部
 普通 Codex 话题空闲时，`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。
 Core 先在原请求日志固定 Session 与候选 Turn，原 listener 随即继续受理消息；现有
 回复工作池再恢复这条 prepared 请求，读取原生命令清单并取消原目标。恢复只能缩小
-固定候选，不能追停后来的 Turn 或新 Session。慢速或失败的原生控制不阻塞入站，
+固定候选，不能追停后来的 Turn 或新 Session。prepared 恢复复用受理路径在原请求与
+来源路由锁内的新鲜授权核验，不在等待锁前重复探测 provider；等待期间撤权即阻止
+控制和投递。慢速或失败的原生控制不阻塞入站，
 也不提前回复“已停止”；没有固定目标的旧请求保持未知，活跃 Turn 仍沿原确切中断路径。
 这只调整现有 IO 调度，不新增队列、消费者或完成状态；实际安装、停止时延与完整社区
 准出仍需分别验收。

loopx/capabilities/native_chat/conversation_bindings.py

@@ -183,10 +183,17 @@ def resolve(self, *, binding_id: str, source_ref: str, sender_ref: str,
         row = next((item for item in current.get("bindings", []) if item.get("binding_id") == binding_id), None)
         if row is None:
             raise ValueError("conversation binding is no longer authorized")
+        observation = dict(self.observe(row["transport_ref"]))
+        # Provider verification can wait on IO. It cannot authorize a binding
+        # that was revoked or replaced while that observation was in flight.
+        # Unrelated binding changes do not invalidate this exact source.
+        current = self.read()
+        if next((item for item in current.get("bindings", [])
+                 if item.get("binding_id") == binding_id), None) != row:
+            raise ValueError("conversation binding changed during identity verification")
         registry_scope = {}
         if row.get("goal_scope") == "all_registered":
             registry_scope["available_goal_ids"] = self.goal_scope_ids(row)
-        observation = dict(self.observe(row["transport_ref"]))
         selected = self._core("collaboration.conversation.bound_context", {
             "current": current, "binding_id": binding_id, "source_ref": source_ref, "sender_ref": sender_ref,
             "private_human_message": private_human_message, "observation": observation,

loopx/capabilities/native_chat/external_conversations.py

@@ -610,12 +610,16 @@ def recover(self, *, request_ref: str | None = None) -> None:
                 # delivering their pending response; do not add an idle probe.
                 if row["status"] == "command_completed" and not row.get("commission_adoption_pending"):
                     continue
-                self.bindings.resolve(binding_id=row["binding_id"], **row["source"])
                 if row["status"] == "prepared":
+                    # Admission resolves current authority after both request
+                    # and route fences. A separate pre-fence network probe
+                    # neither authorizes recovery nor needs to precede it.
                     self.admit(binding_id=row["binding_id"], source=row["source"], request_ref=row["request_ref"],
                                message=row["message"], command=row["command"],
                                attachments=row.get("attachments"), origin=row.get("origin", "lark"))
-                elif row["status"] == "command_queued":
+                    continue
+                self.bindings.resolve(binding_id=row["binding_id"], **row["source"])
+                if row["status"] == "command_queued":
                     self._run_steward_action(row)
                 elif row["status"] == "accepted":
                     if row.get("agent_target"):

tests/test_chat_conversation_bindings.py

@@ -259,6 +259,113 @@ def observe(profile):
     runtime.close()
 
 
+@pytest.mark.parametrize("group", [False, True])
+@pytest.mark.parametrize("change", ["revoke", "replace", "unrelated"])
+def test_binding_change_while_identity_is_observed_uses_current_grant(ordinary, monkeypatch, group, change):  # noqa: F811
+    from test_lark_private_conversations import connect, connect_group
+    from loopx.extensions.lark.conversation_identity import lark_private_source
+
+    store, runtime, provider, transport = (connect_group if group else connect)(ordinary)
+    bindings = transport.bindings
+    binding = transport._binding("notes-app")
+    event = provider.topic("grant-observation", "Read project") if group else provider.event("notes-app", "grant-observation", "Read project")
+    source = lark_private_source(provider_ref=binding["provider_ref"], event=event)
+    observe, armed = bindings.observe, True
+
+    def changed(profile, **kwargs):
+        nonlocal armed
+        result = observe(profile, **kwargs)
+        if armed:
+            armed = False
+            if change == "revoke":
+                bindings.disconnect(binding["binding_id"], expected_revision=bindings.read()["revision"])
+            else:
+                target = "steward-app" if change == "unrelated" else "notes-app"
+                if change == "replace":
+                    provider.profile_apps[target] = "cli_replaced_app"
+                bindings.configure(transport_ref=target, project_ref=binding["project_ref"],
+                    executor_endpoint_id="codex", audience="group" if group else None,
+                    group_refs=binding.get("group_refs"), available_group_refs=binding.get("group_refs"))
+        return result
+
+    monkeypatch.setattr(bindings, "observe", changed)
+    try:
+        if change == "unrelated":
+            selected = bindings.resolve(binding_id=binding["binding_id"], **source)
+            assert selected["binding"] == binding
+        else:
+            with pytest.raises(ValueError, match="binding|identity|authorized"):
+                bindings.resolve(binding_id=binding["binding_id"], **source)
+            assert not any(row["enabled"] and row["binding_id"] == binding["binding_id"]
+                           for row in bindings.read()["bindings"])
+        assert not store.list_sessions() and not provider.writes
+    finally:
+        runtime.close()
+
+
+@pytest.mark.parametrize("revoke", [False, True])
+def test_prepared_stop_recovery_waits_for_route_fence_before_its_identity_probe(ordinary, monkeypatch, revoke):  # noqa: F811
+    import threading
+    from concurrent.futures import ThreadPoolExecutor
+    from test_chat_completed_background_stop import _completed_with_later_answer
+    from loopx.file_lock import exclusive_file_lock
+
+    store, runtime, provider, transport, root, first, wire = _completed_with_later_answer(ordinary, monkeypatch)
+    stop = provider.topic("fenced-recovery-stop", "/stop", root=root["message_id"])
+    transport.admit("notes-app", stop, defer_background_stop=True)
+    frozen = next(row for row in transport.core.pending() if row["command"] == "stop")
+    assert frozen["status"] == "prepared" and frozen["target_recorded"]
+    fence = transport.core.root / "source-fences" / f"{frozen['binding_id']}.{frozen['source']['source_ref']}.json"
+    started, observed = threading.Event(), threading.Event()
+    observe, finish = transport.bindings.observe, transport.core._finish_background_stop
+    before_native = []
+
+    def observing(*args, **kwargs):
+        observed.set()
+        return observe(*args, **kwargs)
+
+    def finishing(*args, **kwargs):
+        before_native.append(sum("auth" in call for call in provider.calls))
+        return finish(*args, **kwargs)
+
+    def recover():
+        started.set()
+        transport.core.recover(request_ref=frozen["request_ref"])
+
+    monkeypatch.setattr(transport.bindings, "observe", observing)
+    monkeypatch.setattr(transport.core, "_finish_background_stop", finishing)
+    provider.calls.clear()
+    writes_before = list(provider.writes)
+    receipts = {turn_id: store.load_turn(first["session_id"], turn_id)
+                for turn_id in frozen["background_turn_ids"]}
+    try:
+        with ThreadPoolExecutor(max_workers=1) as pool:
+            with exclusive_file_lock(fence, operation="hold_original_source_route"):
+                future = pool.submit(recover)
+                assert started.wait(5)
+                assert not observed.wait(.1), "A pre-fence probe cannot authorize prepared recovery"
+                if revoke:
+                    transport.bindings.disconnect(frozen["binding_id"], expected_revision=transport.bindings.read()["revision"])
+            future.result(timeout=10)
+        current = transport.core.read_request(frozen["request_ref"])
+        if revoke:
+            assert current["status"] == "prepared" and current["background_turn_ids"] == frozen["background_turn_ids"]
+            assert not before_native and wire.children["owned"].poll() is None
+            assert provider.writes == writes_before
+        else:
+            assert before_native == [1], "The fenced admission already owns the fresh identity proof"
+            assert current["status"] == "command_completed" and current["background_turn_ids"] == [first["turn_id"]]
+            assert wire.children["owned"].poll() is not None
+            transport.reconcile()
+            assert sum("停止" in text for _, text in provider.writes) == 1
+        assert wire.children["unowned"].poll() is None
+        assert all(store.load_turn(first["session_id"], turn_id) == receipt
+                   for turn_id, receipt in receipts.items())
+    finally:
+        wire.close()
+        runtime.close()
+
+
 @pytest.mark.parametrize("resumed", [False, True])
 def test_group_disconnect_stops_running_and_queued_turns_without_stopping_other_session(ordinary, resumed):  # noqa: F811
     from test_lark_private_conversations import connect_group