diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 054df53612..86d4d74cd0 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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 diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 10135a7690..6f7b7025c9 100644 --- a/tests/test_lark_private_conversations.py +++ b/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):