perf(lark): overlap completed stop source verification

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

@@ -112,6 +112,14 @@ target that topic. Admission feedback, streamed posts, terminal edits and receip
 recovery use the existing delivery path with `source_thread` placement. A revoked
 binding or unreadable source prevents later result delivery.
 
+A completed, target-frozen stop on a locally enabled group binding reads its two
+original provider messages while Core freshly resolves the binding and App
+identity. Both readonly observations must finish successfully before source,
+ancestry and root-mention validation can authorize delivery. Only that invocation
+reuses the verified identity; no observation is persisted or reused by a later
+verification. Changed bindings discard the buffered message observation; ordinary
+input, private conversations and unknown stop scope keep their existing paths.
+
 When an ordinary Codex topic has no active Turn, `/stop` also covers background
 commands left by completed Turns in that same canonical Session, even after a
 later answer. Core freezes the original Session and canonical candidate Turn

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

@@ -81,6 +81,11 @@ listener、队列或模型 runner。一个 App 仍只有一个 binding owner，
 编辑和回执恢复复用现有投递路径，使用 `source_thread` 落点。解绑或原消息不可读后，
 即使执行完成，也不得继续发送结果。
 
+本机群绑定仍启用且 Core 已完成、固定原目标的停止请求，可在 Core 重新核验绑定和
+App 身份时，并行读取两条原 provider 消息。两项只读观测全部成功后，仍须核对来源、
+关联关系和根消息提及，才能投递；仅该次校验复用身份观测，不持久化或跨次复用。
+绑定发生变化时丢弃预读消息；普通输入、本人私聊和未知停止范围保持原路径。
+
 普通 Codex 话题空闲时，`/stop` 也覆盖同一原生 Session 中已完成 Turn 遗留的后台命令。
 Core 先在原请求日志固定 Session 与候选 Turn，原 listener 随即继续受理消息；现有
 回复工作池再恢复这条 prepared 请求，读取原生命令清单并取消原目标。恢复只能缩小

loopx/extensions/lark/private_conversations.py

@@ -121,6 +121,25 @@ def _completed_stop_sources(self, profile: str, message_id: str,
         messages = {item["message_id"]: item for item in items}
         return messages[message_id], messages[root_id]
 
+    def _completed_stop_root(self, record: Mapping[str, Any], binding: Mapping[str, Any]) -> str | None:
+        """Read scheduling only; fresh Core resolution still owns authorization."""
+        event = record["event"]
+        root = str(event.get("root_id") or event["message_id"])
+        path = self.core.root / f"{record['request_ref']}.json"
+        native = _read_json(path) if path.exists() else {}
+        if (binding.get("enabled") is True and binding.get("audience") == "group"
+                and binding.get("binding_id") == record["binding_id"]
+                and binding.get("transport_ref") == record["profile"]
+                and root != event["message_id"] and native.get("command") == "stop"
+                and native.get("status") == "command_completed" and native.get("target_recorded") is True
+                and native.get("binding_id") == record["binding_id"]
+                and native.get("source") == record["source"] == lark_private_source(
+                    provider_ref=binding["provider_ref"], event=event)
+                and identity_ref(binding["provider_ref"], event["message_id"])
+                == record["request_ref"] == native.get("request_ref")):
+            return root
+        return None
+
     def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] | None = None,
                         render_content: bool = True) -> Mapping[str, Any] | None:
         """Read the exact source under this App, then recheck the Core audience.
@@ -132,8 +151,30 @@ def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] |
         result: Mapping[str, Any]
         message: Mapping[str, Any] | None
         try:
+            prefetched = None
+            read_binding: Mapping[str, Any] | None = None
             if selected is None:
-                selected = self.bindings.resolve(binding_id=record["binding_id"], **record["source"])
+                # Only an existing, locally enabled binding and a completed,
+                # target-frozen Core stop can schedule these original two IDs.
+                # Buffer readonly IO while Core freshly verifies identity and
+                # authority; neither result is usable until both have joined.
+                if not render_content:
+                    read_binding = next((row for row in self.bindings.read()["bindings"]
+                                         if row.get("binding_id") == record["binding_id"]), {})
+                root = self._completed_stop_root(record, read_binding) if read_binding else None
+                if root:
+                    with ThreadPoolExecutor(max_workers=1) as reads:
+                        envelopes = reads.submit(self._completed_stop_sources,
+                                                 record["profile"], event["message_id"], root)
+                        selected = self.bindings.resolve(binding_id=record["binding_id"],
+                                                         include_observation=True, **record["source"])
+                        prefetched = envelopes.result()
+                    current = next((row for row in self.bindings.read()["bindings"]
+                                    if row.get("binding_id") == record["binding_id"]), None)
+                    if current != read_binding or current != selected["binding"]:
+                        return None  # Revocation/replacement while reads ran is not a delivery grant.
+                else:
+                    selected = self.bindings.resolve(binding_id=record["binding_id"], **record["source"])
             native_path = self.core.root / f"{record['request_ref']}.json"
             native = _read_json(native_path) if native_path.exists() else {}
             if native.get("agent_target"):
@@ -144,7 +185,9 @@ def rendered() -> Mapping[str, Any]:
                           "--as", "bot", "--no-reactions", "--format", "json"]))
 
             group = selected["binding"].get("audience") == "group"
-            observation = None
+            # Resolve has just verified this App for this invocation. Reuse
+            # that observation for mention checks, never across verifications.
+            observation = selected.get("observation")
             stop_root: Mapping[str, Any] | None = None
             completed_stop = False
             if group:
@@ -155,7 +198,6 @@ def rendered() -> Mapping[str, Any]:
                     # Admission already verified this exact App. Reuse only
                     # that call's ephemeral observation; Core still rechecks
                     # under its fences, and outbound verification stays fresh.
-                    observation = selected.get("observation")
                     # An untrusted event hint only changes read scheduling.
                     # Only authenticated provider text can select a control;
                     # ordinary input still needs its display projection.
@@ -188,25 +230,24 @@ def rendered() -> Mapping[str, Any]:
                     # Only batch IDs already fixed by a target-frozen Core stop.
                     # The current binding above still authorizes this read.
                     # Event hints alone keep the source-first lookup.
-                    completed_stop = (root_id != event["message_id"]
-                        and native.get("command") == "stop" and native.get("status") == "command_completed"
-                        and native.get("target_recorded") is True
-                        and native.get("binding_id") == record["binding_id"]
-                        and native.get("source") == record["source"] == lark_private_source(
-                            provider_ref=selected["binding"]["provider_ref"], event=event)
-                        and identity_ref(selected["binding"]["provider_ref"], event["message_id"])
-                        == record["request_ref"] == native.get("request_ref"))
-                    # Delivery still needs both fresh observations. Their IO
-                    # is independent; join it before checking ancestry and @.
+                    completed_stop = self._completed_stop_root(record, selected["binding"]) == root_id
+                    # Delivery reuses only its own binding observation. An
+                    # explicitly supplied selection without one still needs a
+                    # fresh identity; join that read before ancestry and @.
                     def sources() -> tuple[Mapping[str, Any] | None, Mapping[str, Any] | None]:
                         if completed_stop:
+                            if prefetched is not None and read_binding == selected["binding"]:
+                                return prefetched
                             return self._completed_stop_sources(record["profile"], event["message_id"], root_id)
                         return self._group_source_message(record["profile"], event["message_id"]), None
 
-                    with ThreadPoolExecutor(max_workers=2) as reads:
-                        envelopes = reads.submit(sources)
-                        identity = reads.submit(self.bindings.observe, record["profile"])
-                        (canonical, stop_root), observation = envelopes.result(), identity.result()
+                    if observation is not None:
+                        canonical, stop_root = sources()
+                    else:
+                        with ThreadPoolExecutor(max_workers=2) as reads:
+                            envelopes = reads.submit(sources)
+                            identity = reads.submit(self.bindings.observe, record["profile"])
+                            (canonical, stop_root), observation = envelopes.result(), identity.result()
                     result, message = {"returncode": 0 if canonical else 1}, canonical
             else:
                 result = rendered()

tests/test_lark_private_conversations.py

@@ -1171,14 +1171,14 @@ def completed_group_stop(ordinary):  # noqa: F811
         runtime.close()
 
 
-def test_completed_stop_batches_only_its_core_bound_source_and_root(completed_group_stop):
+def test_completed_stop_batches_its_core_bound_source_and_root_with_one_current_identity(completed_group_stop):
     _, provider, transport, root, event, record = completed_group_stop
     assert transport._source_verified(record)
     calls = [a for a in provider.calls if "GET" in a]
     assert len(calls) == 1 and calls[0][calls[0].index("GET") + 1].endswith("/mget")
     assert json.loads(calls[0][calls[0].index("--params") + 1]) == {
         "message_ids": [event["message_id"], root["message_id"]]}
-    assert sum("auth" in a for a in provider.calls) == 2
+    assert sum("auth" in a for a in provider.calls) == 1
     provider.messages[root["message_id"]]["mentions"] = []
     assert not transport._source_verified(record)  # Re-read, never cache the old root.
     assert not provider.writes and not provider.reaction_creates
@@ -1353,8 +1353,32 @@ def runner(args, cwd=None, timeout=None):
     assert not provider.writes and not provider.reaction_creates
 
 
+@pytest.mark.parametrize("change", ["unverified_app", "other_app", "root_mention", "source_parent"])
+def test_completed_stop_source_verification_reuses_one_current_observation_only_within_the_call(
+    completed_group_stop, change,
+):
+    _, provider, transport, _, _, record = completed_group_stop
+    provider.calls.clear()
+    assert transport._source_verified(record)
+    # The binding check already verified this App. Its observation also checks
+    # the root mention; a second network identity probe adds no new authority.
+    assert sum("auth" in args for args in provider.calls) == 1
+    if change == "unverified_app":
+        provider.bot_ready = False
+    elif change == "other_app":
+        provider.profile_apps[record["profile"]] = "cli_other_app"
+    elif change == "root_mention":
+        provider.messages[record["event"]["root_id"]]["mentions"] = []
+    else:
+        provider.messages[record["event"]["message_id"]]["parent_id"] = "om_other_parent"
+    provider.calls.clear()
+    assert not transport._source_verified(record)
+    assert sum("auth" in args for args in provider.calls) == 1
+    assert not provider.writes and not provider.reaction_creates
+
+
 @pytest.mark.parametrize("identity_fails", [False, True])
-def test_completed_stop_batch_joins_a_separate_fresh_identity(completed_group_stop, monkeypatch, identity_fails):
+def test_completed_stop_preselected_without_observation_joins_a_fresh_identity(completed_group_stop, monkeypatch, identity_fails):
     _, provider, transport, _, _, record = completed_group_stop
     started, observed, release = threading.Event(), threading.Event(), threading.Event()
     original_sources, original_observe = transport._completed_stop_sources, transport.bindings.observe
@@ -1376,18 +1400,107 @@ def observe(*args, **kwargs):
 
     monkeypatch.setattr(transport, "_completed_stop_sources", sources)
     monkeypatch.setattr(transport.bindings, "observe", observe)
+    selected = transport.bindings.resolve(binding_id=record["binding_id"], **record["source"])
     with ThreadPoolExecutor(max_workers=1) as caller:
-        pending = caller.submit(transport._source_verified, record)
+        pending = caller.submit(transport._source_message, record, selected=selected, render_content=False)
         try:
             assert started.wait(2) and observed.wait(2)
             assert not pending.done()
         finally:
             release.set()
-        assert pending.result(timeout=3) is not identity_fails
+        assert bool(pending.result(timeout=3)) is not identity_fails
     assert len(observations) == 2
     assert not provider.writes and not provider.reaction_creates
 
 
+@pytest.mark.parametrize("failure", [None, "identity", "source", "revoked"])
+def test_completed_stop_default_overlaps_readonly_observations_but_joins_and_rechecks_authority(
+    completed_group_stop, monkeypatch, failure,
+):
+    _, provider, transport, _, _, record = completed_group_stop
+    source_started, identity_started, release = threading.Event(), threading.Event(), threading.Event()
+    original_sources, original_observe = transport._completed_stop_sources, transport.bindings.observe
+
+    def sources(*args):
+        source_started.set()
+        assert release.wait(3)
+        return (None, None) if failure == "source" else original_sources(*args)
+
+    def observe(*args, **kwargs):
+        observation = original_observe(*args, **kwargs)
+        identity_started.set()
+        assert release.wait(3)
+        if failure == "identity":
+            raise ValueError("current App observation unavailable")
+        return observation
+
+    monkeypatch.setattr(transport, "_completed_stop_sources", sources)
+    monkeypatch.setattr(transport.bindings, "observe", observe)
+    with ThreadPoolExecutor(max_workers=1) as caller:
+        pending = caller.submit(transport._source_verified, record)
+        try:
+            assert source_started.wait(2) and identity_started.wait(2)
+            assert not pending.done() and not provider.writes
+            if failure == "revoked":
+                state = transport.bindings.read()
+                transport.bindings.disconnect(binding_id=record["binding_id"], expected_revision=state["revision"])
+        finally:
+            release.set()
+        assert pending.result(timeout=3) is (failure is None)
+    assert not provider.writes and not provider.reaction_creates
+
+
+@pytest.mark.parametrize("revoke", [False, True])
+def test_completed_stop_reply_worker_waits_for_both_reads_and_does_not_send_after_revocation(
+    completed_group_stop, monkeypatch, revoke,
+):
+    _, provider, transport, _, event, record = completed_group_stop
+    checking_source, source_started, identity_started, release = (
+        threading.Event() for _ in range(4)
+    )
+    original_verify = transport._source_verified
+    original_sources, original_observe = transport._completed_stop_sources, transport.bindings.observe
+
+    def verify(value):
+        checking_source.set()
+        return original_verify(value)
+
+    def sources(*args):
+        source_started.set()
+        assert release.wait(3)
+        return original_sources(*args)
+
+    def observe(*args, **kwargs):
+        observation = original_observe(*args, **kwargs)
+        if checking_source.is_set():
+            identity_started.set()
+            assert release.wait(3)
+        return observation
+
+    monkeypatch.setattr(transport, "_source_verified", verify)
+    monkeypatch.setattr(transport, "_completed_stop_sources", sources)
+    monkeypatch.setattr(transport.bindings, "observe", observe)
+    path = transport.root / f"{record['request_ref']}.json"
+    with ThreadPoolExecutor(max_workers=1) as caller:
+        pending = caller.submit(transport.reconcile_request, path)
+        try:
+            assert source_started.wait(2) and identity_started.wait(2)
+            assert not pending.done() and not provider.writes
+            if revoke:
+                state = transport.bindings.read()
+                transport.bindings.disconnect(binding_id=record["binding_id"], expected_revision=state["revision"])
+        finally:
+            release.set()
+        assert pending.result(timeout=3) == (0 if revoke else 1)
+    delivered = json.loads(path.read_text())
+    if revoke:
+        assert not provider.writes and not delivered["deliveries"].get("terminal", {}).get("verified")
+    else:
+        assert delivered["deliveries"]["terminal"]["verified"]
+        assert len(provider.writes) == 1
+        assert provider.messages[event["message_id"]]["root_id"] == event["root_id"]
+
+
 @pytest.mark.parametrize("followup", [False, True])
 @pytest.mark.parametrize("identity_fails", [False, True])
 def test_group_outbound_joins_current_source_and_identity_before_root_read(ordinary, monkeypatch, followup, identity_fails):  # noqa: F811