fix(lark): keep Inbox metadata probing off completed stop replies

loopx/extensions/lark/private_conversations.py

@@ -294,11 +294,26 @@ def _group_text_command_input(message: Mapping[str, Any], observation: Mapping[s
                 return text[len(leading[0]["key"]):].strip()
         return text
 
-    def _inbox(self, record: dict[str, Any], *, observation: dict[str, Any] | None = None) -> Path:
-        if observation is None:
-            observation = self.bindings.observe(record["profile"])
+    def _inbox(self, record: dict[str, Any], *, observation: dict[str, Any] | None = None,
+               reuse_display_name: bool = False) -> Path:
         scope = record["source"]["source_ref"]
+        if not isinstance(scope, str) or not re.fullmatch(r"[a-f0-9]{24}", scope):
+            raise ValueError("invalid private source scope")
         path = self.runtime_root / ".loopx" / "config" / "private-chat" / f"{scope}.json"
+        if observation is None and reuse_display_name and path.exists():
+            try:
+                previous = _read_json(path)
+                reply = previous.get("reply") or {}
+                name = reply.get("bot_display_name")
+                if (previous.get("schema_version") == "lark_event_inbox_config_v0"
+                        and reply.get("sender_profile") == record["profile"]
+                        and reply.get("chat_id") == record["event"]["chat_id"]
+                        and isinstance(name, str) and name):
+                    observation = {"bot_display_name": name}
+            except (OSError, ValueError, AttributeError):
+                pass
+        if observation is None:
+            observation = self.bindings.observe(record["profile"])
         _atomic_write_json(path, {
             "schema_version": "lark_event_inbox_config_v0", "enabled": True,
             "inbox_dir": f".loopx/inbox/private-chat/{scope}", "capture_scope": "configured_chat_all",
@@ -781,18 +796,19 @@ def reconcile_request(self, path: Path) -> int:
             def inbox() -> Path:
                 nonlocal config
                 if config is None:
-                    # The binding owner just verified this exact source. Only
-                    # reuse its presentation metadata; outbound provenance and
-                    # Core recovery/admission keep their own fresh checks.
-                    config = self._inbox(record, observation=selected["observation"] if selected else None)
+                    # A name supplies presentation metadata only; outbound
+                    # provenance and Core recovery/admission keep their own
+                    # fresh checks, including on the completed-stop path.
+                    config = self._inbox(record, observation=selected["observation"] if selected else None,
+                                         reuse_display_name=control_receipt_first and selected is None)
                 return config
 
             try:
-                selected = self.bindings.resolve(binding_id=record["binding_id"],
-                                                 include_observation=True, **record["source"])
                 if record["status"] in {"command_queued", "command_completed", "commission_running"}:
                     native = self.core.read_request(record["request_ref"])
                     if native["status"] == "command_queued":
+                        selected = self.bindings.resolve(binding_id=record["binding_id"],
+                                                         include_observation=True, **record["source"])
                         self._feedback(path, record, inbox=inbox)
                         self._deliver(path, record, "admission", native["response"], inbox=inbox)
                         return 0
@@ -801,6 +817,18 @@ def inbox() -> Path:
                                   status_snapshot=native.get("status_snapshot"))
                     control_receipt_first = (native["status"] == "command_completed"
                                              and native.get("command") in CHAT_CONTROL_COMMANDS.values())
+                # A completed group stop has no Core work left. Its existing
+                # Inbox name is presentation only; sending still runs both
+                # fresh binding/source observations in _source_verified. Do
+                # not put an extra metadata probe ahead of that receipt.
+                if not (control_receipt_first and native.get("command") == "stop"
+                        and native.get("target_recorded") is True
+                        and record["source"].get("group_human_message") is True
+                        and not record["deliveries"].get("terminal", {}).get("verified")
+                        and all(native.get(key) == record.get(key)
+                                for key in ("request_ref", "binding_id", "source"))):
+                    selected = self.bindings.resolve(binding_id=record["binding_id"],
+                                                     include_observation=True, **record["source"])
                 if record["status"] == "accepted":
                     native = self.core.read_request(record["request_ref"])
                     if native.get("agent_target"):
@@ -888,6 +916,13 @@ def inbox() -> Path:
                     record["status"] = "delivered"
                     _atomic_write_json(path, record)
                     return 1
+                if (control_receipt_first and selected is None
+                        and record["deliveries"].get("terminal", {}).get("blocker")
+                        == "lark_inbox_reply_sender_identity_mismatch"):
+                    # A renamed Bot can invalidate the presentation hint. No
+                    # send occurred; refresh only metadata for the next normal
+                    # reconcile, preserving any frozen provider attempt.
+                    self._inbox(record)
             except (KeyError, ValueError, OSError):
                 # Keep pending receipts available for actionable recovery.
                 return 0

tests/test_lark_private_conversations.py

@@ -1184,6 +1184,107 @@ def test_completed_stop_batches_only_its_core_bound_source_and_root(completed_gr
     assert not provider.writes and not provider.reaction_creates
 
 
+def test_completed_stop_inbox_name_cannot_replace_fresh_source_checks(completed_group_stop, monkeypatch):
+    _, provider, transport, root, event, record = completed_group_stop
+    transport._inbox(record)  # The preceding ordinary topic already has an Inbox.
+    path = transport.root / f"{record['request_ref']}.json"
+    entered, release = threading.Event(), threading.Event()
+    original = transport._source_verified
+
+    def verify(row):
+        entered.set()
+        assert release.wait(3)
+        return original(row)
+
+    monkeypatch.setattr(transport, "_source_verified", verify)
+    with ThreadPoolExecutor(max_workers=1) as workers:
+        pending = workers.submit(transport.reconcile_request, path)
+        try:
+            assert entered.wait(2)
+            assert not provider.writes and not provider.reaction_creates
+            # The hint cannot retain authority when the root loses its @.
+            provider.messages[root["message_id"]]["mentions"] = []
+        finally:
+            release.set()
+        assert pending.result(timeout=3) == 0
+    assert not provider.writes and not provider.reaction_creates
+    provider.messages[root["message_id"]]["mentions"] = [{"id": {"open_id": "ou_notes_app_bot"}}]
+    assert transport.reconcile_request(path) == 1
+    assert provider.topic_writes == [(event["chat_id"], root["message_id"], "已请求停止这条消息对应的执行。")]
+    assert transport.reconcile_request(path) == 0 and len(provider.writes) == 1
+
+
+@pytest.mark.parametrize("change", ["app", "bot", "revoke", "sender", "chat", "root", "parent", "thread"])
+def test_completed_stop_cached_name_does_not_authorize_a_changed_source(completed_group_stop, change):
+    _, provider, transport, _, event, record = completed_group_stop
+    transport._inbox(record)
+    if change == "app":
+        provider.profile_apps["notes-app"] = "cli_other"
+    elif change == "bot":
+        provider.bot_ready = False
+    elif change == "revoke":
+        transport.bindings.disconnect(record["binding_id"], expected_revision=transport.bindings.read()["revision"])
+    else:
+        field, value = {"sender": ("sender", {"id": "ou_other", "sender_type": "user"}),
+            "chat": ("chat_id", "oc_other"), "root": ("root_id", "om_other"),
+            "parent": ("parent_id", "om_other"), "thread": ("thread_id", "omt_other")}[change]
+        provider.messages[event["message_id"]][field] = value
+    assert transport.reconcile_request(transport.root / f"{record['request_ref']}.json") == 0
+    assert not provider.writes and not provider.reaction_creates
+
+
+def test_completed_stop_verified_reply_rechecks_grant_before_feedback(completed_group_stop):
+    _, provider, transport, _, _, record = completed_group_stop
+    config = transport._inbox(record)
+    path = transport.root / f"{record['request_ref']}.json"
+    # Model a crash after verified reply, before the original source's cleanup.
+    assert transport._deliver(path, record, "terminal", "已请求停止这条消息对应的执行。", inbox=lambda: config)
+    writes = list(provider.writes)
+    assert not provider.reaction_creates
+    transport.bindings.disconnect(record["binding_id"], expected_revision=transport.bindings.read()["revision"])
+    assert transport.reconcile_request(path) == 0
+    assert provider.writes == writes and not provider.reaction_creates
+    assert json.loads(path.read_text())["status"] == "command_completed"
+
+
+def test_completed_stop_refreshes_a_renamed_bot_without_resending_an_attempt(completed_group_stop, monkeypatch):
+    _, provider, transport, root, event, record = completed_group_stop
+    config = transport._inbox(record)
+    original = Provider.__call__
+    def renamed(self, args, cwd=None, timeout=None):
+        result = original(self, args, cwd, timeout)
+        if "auth" in args:
+            value = json.loads(result["stdout"])
+            value["identities"]["bot"]["appName"] = "Renamed Bot"
+            result["stdout"] = json.dumps(value)
+        return result
+    monkeypatch.setattr(Provider, "__call__", renamed)
+    path = transport.root / f"{record['request_ref']}.json"
+    assert transport.reconcile_request(path) == 0 and not provider.writes
+    assert json.loads(config.read_text())["reply"]["bot_display_name"] == "Renamed Bot"
+    assert not json.loads(path.read_text())["deliveries"]["terminal"].get("started")
+    assert transport.reconcile_request(path) == 1
+    assert provider.topic_writes == [(event["chat_id"], root["message_id"], "已请求停止这条消息对应的执行。")]
+    assert transport.reconcile_request(path) == 0 and len(provider.writes) == 1
+
+
+@pytest.mark.parametrize("invalid", ["missing", "malformed", "foreign_profile", "foreign_chat"])
+def test_completed_stop_rebuilds_unusable_inbox_metadata(completed_group_stop, invalid):
+    _, provider, transport, root, event, record = completed_group_stop
+    config = transport._inbox(record)
+    value = json.loads(config.read_text())
+    if invalid == "missing":
+        config.unlink()
+    elif invalid == "malformed":
+        config.write_text('{')
+    else:
+        value["reply"]["sender_profile" if invalid == "foreign_profile" else "chat_id"] = "foreign"
+        value["reply"]["bot_display_name"] = "Foreign Bot"
+        config.write_text(json.dumps(value))
+    assert transport.reconcile_request(transport.root / f"{record['request_ref']}.json") == 1
+    assert provider.topic_writes == [(event["chat_id"], root["message_id"], "已请求停止这条消息对应的执行。")]
+
+
 @pytest.mark.parametrize("field,value", [("target_recorded", False), ("command", "status"),
     ("status", "prepared"), ("binding_id", "other"), ("source", {}), ("request_ref", "other")])
 def test_completed_stop_batch_requires_the_persisted_core_proof(completed_group_stop, field, value):