From 7c601a2844a842adde522e558ee4ba95da9a9d9e Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 17 Oct 2026 00:50:05 +0800 Subject: [PATCH] perf(lark): batch verified completed-stop source reads Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../extensions/lark/private_conversations.py | 58 ++++++- tests/test_lark_private_conversations.py | 151 +++++++++++++++++- 2 files changed, 197 insertions(+), 12 deletions(-) diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 2d9fcbe0c9..054df53612 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -21,7 +21,8 @@ from ...presentation.renderers.conversation_status_markdown import render_conversation_status from .conversation_identity import identity_ref, lark_private_source from .event_inbox import acknowledge_lark_event_inbox, ingest_lark_event_inbox -from .goal_channel_transport import APP_ID_PATTERN, call, json_payload, lark_args, lark_provider_mention_identities +from .goal_channel_transport import call, json_payload, lark_args, lark_provider_mention_identities +from .identity_shapes import LARK_APP_ID_PATTERN as APP_ID_PATTERN from .inbox_reply import _message, reply_lark_event_inbox, update_lark_inbox_reply, verify_lark_inbox_reply from .manager_context import manager_failure_reply from .inbox_reactions import mark_lark_event_inbox_processing, mark_lark_event_inbox_received @@ -101,6 +102,25 @@ def _group_source_message(self, profile: str, message_id: str) -> Mapping[str, A return None return _message(payload, message_id) + def _completed_stop_sources(self, profile: str, message_id: str, + root_id: str) -> tuple[Mapping[str, Any] | None, Mapping[str, Any] | None]: + # Both IDs must already belong to the persisted Core stop below. Raw + # mget retains body/ancestry/mentions; the display shortcut does not. + result = call(self.runner, lark_args(cli_bin=self.cli_bin, profile=profile, + tail=["api", "GET", "/open-apis/im/v1/messages/mget", "--params", + json.dumps({"message_ids": [message_id, root_id]}), "--as", "bot"])) + payload = json_payload(result) + if result.get("returncode") != 0 or payload.get("code", 0) != 0 or payload.get("ok") is False: + return None, None + data = payload.get("data") + items = data.get("items") if isinstance(data, Mapping) else None + if (not isinstance(items, list) or len(items) != 2 + or any(not isinstance(item, Mapping) or not isinstance(item.get("message_id"), str) for item in items) + or {item.get("message_id") for item in items} != {message_id, root_id}): + return None, None + messages = {item["message_id"]: item for item in items} + return messages[message_id], messages[root_id] + 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. @@ -109,6 +129,8 @@ def _source_message(self, record: dict[str, Any], *, selected: dict[str, Any] | scopes. A caller cannot opt into this path with an HTTP boolean. """ event = record["event"] + result: Mapping[str, Any] + message: Mapping[str, Any] | None try: if selected is None: selected = self.bindings.resolve(binding_id=record["binding_id"], **record["source"]) @@ -123,11 +145,12 @@ def rendered() -> Mapping[str, Any]: group = selected["binding"].get("audience") == "group" observation = None + stop_root: Mapping[str, Any] | None = None + completed_stop = False if group: - # Both projections read this exact message under the same App. - # The root is read only after canonical ancestry checks below. - # Outbound provenance needs no display rendering. Ordinary - # admission retains that projection for text and media inputs. + # Canonical envelopes alone supply ancestry and @. Outbound + # provenance needs no display rendering; ordinary admission + # retains that projection for text and media inputs. if render_content: # Admission already verified this exact App. Reuse only # that call's ephemeral observation; Core still rechecks @@ -149,8 +172,10 @@ def rendered() -> Mapping[str, Any]: observation = identity.result() command_input = (self._group_text_command_input(canonical, observation) if control_hint and canonical is not None + and observation is not None and canonical.get("msg_type", canonical.get("message_type")) == "text" else None) if (command_input in CHAT_CONTROL_COMMANDS + and canonical is not None and (not native or native.get("message") == command_input)): result, message = {"returncode": 0}, {**canonical, "content": command_input} else: @@ -159,12 +184,29 @@ def rendered() -> Mapping[str, Any]: result = display.result() if display is not None else rendered() message = _message(json_payload(result), event["message_id"]) else: + root_id = str(event.get("root_id") or event["message_id"]) + # 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 @. + def sources() -> tuple[Mapping[str, Any] | None, Mapping[str, Any] | None]: + if completed_stop: + 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: - current = reads.submit(self._group_source_message, record["profile"], event["message_id"]) + envelopes = reads.submit(sources) identity = reads.submit(self.bindings.observe, record["profile"]) - canonical, observation = current.result(), identity.result() + (canonical, stop_root), observation = envelopes.result(), identity.result() result, message = {"returncode": 0 if canonical else 1}, canonical else: result = rendered() @@ -201,7 +243,7 @@ def rendered() -> Mapping[str, Any]: or parent != event_parent or (event_thread and event_thread != str(message.get("thread_id") or ""))): return None - original = (message if root == event["message_id"] else + original = (message if root == event["message_id"] else stop_root if completed_stop else self._group_source_message(record["profile"], root)) from .event_inbox import lark_event_mentions_bot if observation is None: diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index ed08df0ded..10135a7690 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -71,10 +71,13 @@ def __call__(self, args, cwd=None, timeout=None): elif "api" in args and "GET" in args: path = args[args.index("GET") + 1] assert path.startswith("/open-apis/im/v1/messages/") - message = dict(self.messages[path.rsplit("/", 1)[-1]]) - if message.get("msg_type") == "text" and "content" in message and "body" not in message: - message["body"] = {"content": json.dumps({"text": message.pop("content")})} - data = {"code": 0, "data": {"items": [message]}} + ids = (json.loads(args[args.index("--params") + 1])["message_ids"] + if path.endswith("/mget") else [path.rsplit("/", 1)[-1]]) + messages = [dict(self.messages[ref]) for ref in ids] + for message in messages: + if message.get("msg_type") == "text" and "content" in message and "body" not in message: + message["body"] = {"content": json.dumps({"text": message.pop("content")})} + data = {"code": 0, "data": {"items": messages}} elif "+messages-mget" in args: ref = args[args.index("--message-ids") + 1] message = self.messages[ref] @@ -1144,6 +1147,146 @@ def test_group_outbound_source_verification_reads_fresh_envelopes_without_displa runtime.close() +@pytest.fixture +def completed_group_stop(ordinary): # noqa: F811 + _, runtime, provider, transport = connect_group(ordinary) + try: + root = provider.topic("batch-root", "wait for interrupt") + assert transport.admit("notes-app", root)["status"] == "durably_accepted" + request = transport.core.pending()[0] + deadline = time.monotonic() + 5 + while runtime.store.load_turn(request["session_id"], request["turn_id"])["status"] != "running": + assert time.monotonic() < deadline + time.sleep(.01) + event = provider.topic("batch-stop", "/stop", root=root["message_id"]) + assert transport.admit("notes-app", event)["status"] == "command_recorded" + assert runtime.wait_for_turn(session_id=request["session_id"], + turn_id=request["turn_id"], timeout_sec=5)["status"] == "interrupted" + record = next(json.loads(p.read_text()) for p in transport.root.glob("*.json") + if json.loads(p.read_text())["event"]["message_id"] == event["message_id"]) + assert transport.core.read_request(record["request_ref"])["target_recorded"] + provider.calls.clear() + yield runtime, provider, transport, root, event, record + finally: + runtime.close() + + +def test_completed_stop_batches_only_its_core_bound_source_and_root(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 + 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 + + +@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): + _, provider, transport, _, _, record = completed_group_stop + path = transport.core.root / f"{record['request_ref']}.json" + native = json.loads(path.read_text()) + path.write_text(json.dumps({**native, field: value})) + transport._source_verified(record) + assert not any("GET" in a and a[a.index("GET") + 1].endswith("/mget") for a in provider.calls) + assert not provider.writes and not provider.reaction_creates + + +@pytest.mark.parametrize("change", ["sender", "chat", "type", "root", "parent", "thread", + "mention", "root_sender", "root_parent", "root_chat", "app", "bot", "revoke"]) +def test_completed_stop_batch_retains_source_and_grant_fences(completed_group_stop, change): + _, provider, transport, root, event, record = completed_group_stop + message, original = provider.messages[event["message_id"]], provider.messages[root["message_id"]] + if change in {"sender", "chat", "type", "root", "parent", "thread"}: + field, value = {"sender": ("sender", {"id": "ou_other", "sender_type": "user"}), + "chat": ("chat_id", "oc_other"), "type": ("msg_type", "image"), + "root": ("root_id", "om_other"), "parent": ("parent_id", "om_other"), + "thread": ("thread_id", "omt_other")}[change] + message[field] = value + elif change == "mention": + original["mentions"] = [] + elif change == "root_sender": + original["sender"] = {"id": "ou_other", "sender_type": "app"} + elif change == "root_parent": + original["parent_id"] = "om_other" + elif change == "root_chat": + original["chat_id"] = "oc_other" + elif change == "app": + provider.profile_apps["notes-app"] = "cli_other" + elif change == "bot": + provider.bot_ready = False + else: + transport.bindings.disconnect(record["binding_id"], expected_revision=transport.bindings.read()["revision"]) + assert not transport._source_verified(record) + assert not provider.writes and not provider.reaction_creates + + +@pytest.mark.parametrize("failure", ["missing", "duplicate", "extra", "error", "unavailable", "bad_id"]) +def test_completed_stop_batch_rejects_partial_or_ambiguous_results(completed_group_stop, failure): + _, provider, transport, _, _, record = completed_group_stop + def runner(args, cwd=None, timeout=None): + result = provider(args, cwd, timeout) + if "GET" in args and args[args.index("GET") + 1].endswith("/mget"): + data = json.loads(result["stdout"]) + items = data["data"]["items"] + if failure == "missing": + items.pop() + elif failure == "duplicate": + items[1] = items[0] + elif failure == "extra": + items.append({"message_id": "om_unrequested"}) + elif failure == "error": + data["code"] = 999 + elif failure == "bad_id": + items[0]["message_id"] = [] + else: + result["returncode"] = 1 + result["stdout"] = json.dumps(data) + return result + transport.runner = runner + assert not transport._source_verified(record) + 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): + _, 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 + observations = [] + + def sources(*args): + started.set() + assert release.wait(3) + return original_sources(*args) + + def observe(*args, **kwargs): + observations.append(True) + if len(observations) == 2: + assert started.wait(2) + observed.set() + if identity_fails: + raise ValueError("fresh App observation unavailable") + return original_observe(*args, **kwargs) + + 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 started.wait(2) and observed.wait(2) + assert not pending.done() + finally: + release.set() + assert pending.result(timeout=3) is not identity_fails + assert len(observations) == 2 + assert not provider.writes and not provider.reaction_creates + + @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