perf(lark): batch source reads for completed stop receipts

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:

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