Avoid repeated private delivery body scans in reply discovery

loopx/extensions/lark/private_conversations.py

@@ -33,6 +33,7 @@
                              mark_lark_event_inbox_processing, mark_lark_event_inbox_received)
 from .outbound import present_lark_conversation_text as _presentation_text
 from .private_images import private_message_caption, private_message_images
+from .private_delivery_discovery import PrivateDeliveryDiscovery
 from .return_files import uploaded_chat_images, verify_chat_images
 from .private_progress import (ACTIVITY_UPDATE_INTERVAL_SEC, UPDATE_INTERVAL_SEC, legacy_recovery_repairs,
                                presentation_phase, presentation_stream, project_progress)
@@ -171,6 +172,7 @@ def __init__(self, *, controller: Any, runtime_root: Path, runner: Any, cli_bin:
         self.reaction_feedback = reaction_feedback
         self.root = controller.store.root / "lark-private-deliveries"
         self._context_recovery_probe_at: dict[Path, float] = {}
+        self._delivery_discovery = PrivateDeliveryDiscovery(self.root)
         self._web_projection_scan_at = 0.0
         self._web_projection_snapshots: dict[str, tuple[int, int]] = {}
         from .owner_attention import LarkOwnerAttention
@@ -1309,8 +1311,7 @@ def pending_delivery_paths(self) -> list[Path]:
         """The durable transport store is the queue; no in-memory admission."""
         paths = []
         recovery_candidate = self.core.context_recovery_observer()
-        for path in sorted(self.root.glob("*.json")):
-            row = _read_json(path)
+        for path, row, repairs in self._delivery_discovery.rows():
             if row["status"] != "delivered":
                 paths.append(path)
             elif (row.get("session_id") and row.get("turn_id")
@@ -1319,7 +1320,7 @@ def pending_delivery_paths(self) -> list[Path]:
                 # The canonical request, provider effect inventory and source
                 # are independently re-read by recover_failed_request.
                 self._context_recovery_probe_at[path] = time.monotonic() + 60
-                if legacy_recovery_repairs(row) or recovery_candidate(row["request_ref"]):
+                if repairs or recovery_candidate(row["request_ref"]):
                     paths.append(path)
         return [*paths, *self._web_projection_paths()]
 
@@ -1344,22 +1345,26 @@ def _web_projection_paths(self) -> list[Path]:
             self._web_projection_scan_at = time.monotonic() + 5
             enabled = {row["binding_id"] for row in self.bindings.read()["bindings"]
                        if row.get("enabled") is True and row.get("audience", "owner") == "owner"}
-            sources: dict[str, dict[str, Any]] = {}
-            for path in sorted(self.root.glob("*.json")):
-                row = _read_json(path)
+            sources: dict[str, Path] = {}
+            for path, row, _ in self._delivery_discovery.rows():
                 if (row.get("binding_id") in enabled and row.get("session_id")
                         and row.get("event", {}).get("chat_type") == "p2p"
                         and not row.get("agent_target") and row.get("status") in {"accepted", "delivered"}):
-                    sources.setdefault(row["session_id"], row)
+                    sources.setdefault(row["session_id"], path)
             store = self.core.controller.store
-            for sid, source in sources.items():
+            for sid, source_path in sources.items():
                 try:
                     stat = (store.sessions_root / sid / "messages.jsonl").stat()
                     snapshot = (stat.st_mtime_ns, stat.st_size)
-                    key = identity_ref(source["binding_id"], sid)
-                    path = root / f"{key}.json"
                     if self._web_projection_snapshots.get(sid) == snapshot:
                         continue
+                    source = _read_json(source_path)
+                    if (source.get("binding_id") not in enabled or source.get("session_id") != sid
+                            or source.get("event", {}).get("chat_type") != "p2p"
+                            or source.get("agent_target") or source.get("status") not in {"accepted", "delivered"}):
+                        continue
+                    key = identity_ref(source["binding_id"], sid)
+                    path = root / f"{key}.json"
                     messages = store.messages(sid)
                     web_turns = {message["turn_id"] for message in messages
                                  if message.get("role") == "user" and message.get("origin") == "web"}

loopx/extensions/lark/private_delivery_discovery.py

@@ -0,0 +1,50 @@
+"""Read-only scheduling hints for the existing private-message reply worker."""
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+from typing import Any
+
+from ...chat_store import _read_json
+from .private_progress import legacy_recovery_repairs
+
+
+class PrivateDeliveryDiscovery:
+    def __init__(self, root: Path) -> None:
+        self.root = root
+        self._completed: dict[Path, tuple[tuple[int, ...], dict[str, Any], bool]] = {}
+
+    def rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]:
+        """Reuse only unchanged delivered-file hints, never bodies or authority.
+
+        Pending files stay fresh. Reconciliation still reads the complete
+        journal and current Core/provider state before any delivery or recovery.
+        """
+        seen = set()
+
+        def fingerprint(path: Path) -> tuple[int, ...]:
+            stat = path.stat()
+            return (stat.st_dev, stat.st_ino, stat.st_mtime_ns, stat.st_ctime_ns, stat.st_size)
+
+        for path in sorted(self.root.glob("*.json")):
+            seen.add(path)
+            before = fingerprint(path)
+            saved = self._completed.get(path)
+            if saved is not None and saved[0] == before:
+                yield path, saved[1], saved[2]
+                continue
+            self._completed.pop(path, None)
+            row = _read_json(path)
+            repairs = False
+            if row["status"] == "delivered":
+                repairs = bool(row.get("session_id") and row.get("turn_id") and legacy_recovery_repairs(row))
+                view = {key: row.get(key) for key in
+                        ("status", "request_ref", "binding_id", "session_id", "turn_id")}
+                view.update(agent_target=bool(row.get("agent_target")),
+                            event={"chat_type": row.get("event", {}).get("chat_type")})
+                if fingerprint(path) == before:
+                    self._completed[path] = (before, view, repairs)
+                row = view
+            yield path, row, repairs
+        for path in self._completed.keys() - seen:
+            del self._completed[path]

tests/test_lark_private_conversations.py

@@ -16,6 +16,7 @@
 from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
 from loopx.extensions.lark.conversation_identity import identity_ref, lark_private_source, observe_lark_conversation_identity
 from loopx.extensions.lark.private_conversations import LarkPrivateConversations
+from loopx.extensions.lark.private_delivery_discovery import PrivateDeliveryDiscovery
 from loopx.extensions.lark.goal_topic_runtime import poll_lark_goal_topic_profile_once
 
 
@@ -4961,6 +4962,111 @@ def held_reaction(self, args, cwd=None, timeout=None):
     assert all(phase.get("verified") for phase in projection["deliveries"].values())
 
 
+def test_private_discovery_skips_unchanged_bodies_but_not_new_work_or_core_checks(tmp_path, monkeypatch):
+    from loopx.chat_store import _atomic_write_json
+    from loopx.extensions.lark import private_conversations as private
+
+    clock, checks, reads = [100.0], [], []
+    eligible = {"old": False}
+    transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
+    transport.root = tmp_path
+    transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path)
+    transport._context_recovery_probe_at = {}
+    transport._web_projection_paths = lambda: []
+    transport.core = SimpleNamespace(context_recovery_observer=lambda: lambda ref:
+        checks.append(ref) or eligible[ref])
+    monkeypatch.setattr(private, "time", SimpleNamespace(monotonic=lambda: clock[0]))
+    read_json = private._read_json
+
+    def observed(path):
+        reads.append(path)
+        return read_json(path)
+
+    monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", observed)
+    old, pending, new = (tmp_path / f"{name}.json" for name in ("old", "pending", "new"))
+    _atomic_write_json(old, {"status": "delivered", "request_ref": "old", "session_id": "session",
+        "turn_id": "failed", "event": {"chat_type": "p2p", "text": "private-body-marker" * 10000},
+        "deliveries": {"terminal": {"text": "private-body-marker" * 10000}}})
+    _atomic_write_json(pending, {"status": "pending"})
+    assert transport.pending_delivery_paths() == [pending]
+    reads.clear()
+    assert transport.pending_delivery_paths() == [pending] and reads == [pending]
+    assert "private-body-marker" not in repr(transport._delivery_discovery._completed)
+    _atomic_write_json(new, {"status": "pending"})
+    assert transport.pending_delivery_paths() == [new, pending]
+    clock[0] += 61
+    eligible["old"] = True
+    assert transport.pending_delivery_paths() == [new, old, pending]
+    clock[0] += 61
+    eligible["old"] = False  # A fresh Core decision wins over unchanged provider hints.
+    assert transport.pending_delivery_paths() == [new, pending]
+    assert checks == ["old", "old", "old"]
+    old.unlink()
+    assert transport.pending_delivery_paths() == [new, pending]
+    assert old not in transport._delivery_discovery._completed
+
+
+@pytest.mark.parametrize("replace", [False, True])
+def test_private_discovery_detects_same_size_updates_with_preserved_mtime(tmp_path, monkeypatch, replace):
+    import os
+    from loopx.chat_store import _atomic_write_json
+
+    transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
+    transport.root = tmp_path
+    transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path)
+    transport._context_recovery_probe_at = {}
+    transport._web_projection_paths = lambda: []
+    transport.core = SimpleNamespace(context_recovery_observer=lambda: lambda _: False)
+    path = tmp_path / "original.json"
+    _atomic_write_json(path, {"status": "delivered"})
+    assert transport.pending_delivery_paths() == []
+    before = path.stat()
+    body = path.read_bytes().replace(b'delivered', b'uncertain')
+    if replace:
+        other = tmp_path / "replacement.tmp"
+        other.write_bytes(body)
+        os.utime(other, ns=(before.st_atime_ns, before.st_mtime_ns))
+        os.replace(other, path)
+    else:
+        path.write_bytes(body)
+        os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
+    assert path.stat().st_size == before.st_size and path.stat().st_mtime_ns == before.st_mtime_ns
+    assert transport.pending_delivery_paths() == [path]
+    # Unknown read failures must remain visible rather than reuse an old
+    # delivered view or turn the request into an empty/successful queue.
+    def unavailable(_):
+        raise OSError("read unavailable")
+
+    monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", unavailable)
+    with pytest.raises(OSError, match="read unavailable"):
+        transport.pending_delivery_paths()
+
+
+def test_private_discovery_does_not_cache_a_file_replaced_during_read(tmp_path, monkeypatch):
+    from loopx.chat_store import _atomic_write_json
+    from loopx.extensions.lark import private_conversations as private
+
+    transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
+    transport.root = tmp_path
+    transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path)
+    transport._context_recovery_probe_at = {}
+    transport._web_projection_paths = lambda: []
+    transport.core = SimpleNamespace(context_recovery_observer=lambda: lambda _: False)
+    path = tmp_path / "original.json"
+    _atomic_write_json(path, {"status": "delivered"})
+    read_json = private._read_json
+
+    def replaced(path):
+        row = read_json(path)
+        _atomic_write_json(path, {"status": "pending"})
+        return row
+
+    monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", replaced)
+    assert transport.pending_delivery_paths() == []
+    monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", read_json)
+    assert transport.pending_delivery_paths() == [path]
+
+
 @pytest.mark.parametrize("background_slots", [0, 2])
 def test_private_recovery_backlog_uses_released_capacity_before_retry_delay(tmp_path, monkeypatch, background_slots):
     from loopx.extensions.lark import goal_topic_runtime_service as worker
@@ -4978,6 +5084,7 @@ def test_private_recovery_backlog_uses_released_capacity_before_retry_delay(tmp_
     native_root.mkdir()
     transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
     transport.root = tmp_path
+    transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path)
     transport._context_recovery_probe_at = {}
     transport._web_projection_paths = lambda: []
     transport.reconcile_app_feedback = lambda: 0
@@ -5057,6 +5164,7 @@ def test_current_body_uses_next_slot_before_unattempted_recovery_backlog(tmp_pat
     slots = 4 - background_slots
     transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
     transport.root = tmp_path
+    transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path)
     transport._context_recovery_probe_at = {}
     transport.core = SimpleNamespace(context_recovery_observer=lambda: lambda ref: True)
     transport.reconcile_app_feedback = lambda: 0
@@ -5129,6 +5237,7 @@ def test_private_recovery_discovery_keeps_negative_and_attempt_backoff(tmp_path,
     eligible = {"candidate": True, "ineligible": False}
     transport = LarkPrivateConversations.__new__(LarkPrivateConversations)
     transport.root = tmp_path
+    transport._delivery_discovery = PrivateDeliveryDiscovery(tmp_path)
     transport._context_recovery_probe_at = {}
     transport._web_projection_paths = lambda: []
     monkeypatch.setattr(private, "time", SimpleNamespace(monotonic=lambda: clock[0]))