From 66660e505ba5902a9220a5d68aa3db750a65c489 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:31:03 +0800 Subject: [PATCH 1/2] Avoid reparsing unchanged private delivery journals during discovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../extensions/lark/private_conversations.py | 65 +++++++++-- tests/test_lark_private_conversations.py | 103 ++++++++++++++++++ 2 files changed, 157 insertions(+), 11 deletions(-) diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index c7b2723467..4af7701846 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -10,7 +10,7 @@ import re import subprocess import time -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any @@ -171,6 +171,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_cache: dict[Path, tuple[tuple[int, ...], dict[str, Any], bool]] = {} self._web_projection_scan_at = 0.0 self._web_projection_snapshots: dict[str, tuple[int, int]] = {} from .owner_attention import LarkOwnerAttention @@ -1304,13 +1305,51 @@ def inbox() -> Path: except (KeyError, ValueError, OSError): return # Unknown authority or provider effects stay pending. + def _delivery_discovery_rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]: + """Reuse only unchanged delivered-file hints, never bodies or authority. + + Pending files stay fresh. A completed file's inode, timestamps and size + must match before its small routing view can be reused; reconciliation + still reads the complete journal and current Core/provider state. + """ + cache = getattr(self, "_delivery_discovery_cache", None) + if cache is None: + self._delivery_discovery_cache = cache = {} + 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 = cache.get(path) + if saved is not None and saved[0] == before: + yield path, saved[1], saved[2] + continue + cache.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: + cache[path] = (before, view, repairs) + row = view + yield path, row, repairs + for path in cache.keys() - seen: + del cache[path] + @effect_runtime_request_scope() 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 +1358,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 +1383,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"} diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 08995cace5..8a6fbf0ffe 100644 --- a/tests/test_lark_private_conversations.py +++ b/tests/test_lark_private_conversations.py @@ -4961,6 +4961,109 @@ 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._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(private, "_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_cache) + _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_cache + + +@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 + from loopx.extensions.lark import private_conversations as private + + transport = LarkPrivateConversations.__new__(LarkPrivateConversations) + transport.root = 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(private, "_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._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(private, "_read_json", replaced) + assert transport.pending_delivery_paths() == [] + monkeypatch.setattr(private, "_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 From 56b10a0c7ef88268a95e2b89cc8f40d624c4f11c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 14:41:11 +0800 Subject: [PATCH 2/2] Keep delivery discovery hints in their bounded provider module Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../extensions/lark/private_conversations.py | 48 ++---------------- .../lark/private_delivery_discovery.py | 50 +++++++++++++++++++ tests/test_lark_private_conversations.py | 20 +++++--- 3 files changed, 68 insertions(+), 50 deletions(-) create mode 100644 loopx/extensions/lark/private_delivery_discovery.py diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index 4af7701846..de63f200fd 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/loopx/extensions/lark/private_conversations.py @@ -10,7 +10,7 @@ import re import subprocess import time -from collections.abc import Callable, Iterator, Mapping, Sequence +from collections.abc import Callable, Mapping, Sequence from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any @@ -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,7 +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_cache: dict[Path, tuple[tuple[int, ...], dict[str, Any], bool]] = {} + 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 @@ -1305,51 +1306,12 @@ def inbox() -> Path: except (KeyError, ValueError, OSError): return # Unknown authority or provider effects stay pending. - def _delivery_discovery_rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]: - """Reuse only unchanged delivered-file hints, never bodies or authority. - - Pending files stay fresh. A completed file's inode, timestamps and size - must match before its small routing view can be reused; reconciliation - still reads the complete journal and current Core/provider state. - """ - cache = getattr(self, "_delivery_discovery_cache", None) - if cache is None: - self._delivery_discovery_cache = cache = {} - 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 = cache.get(path) - if saved is not None and saved[0] == before: - yield path, saved[1], saved[2] - continue - cache.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: - cache[path] = (before, view, repairs) - row = view - yield path, row, repairs - for path in cache.keys() - seen: - del cache[path] - @effect_runtime_request_scope() 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, row, repairs in self._delivery_discovery_rows(): + 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") @@ -1384,7 +1346,7 @@ def _web_projection_paths(self) -> list[Path]: 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, Path] = {} - for path, row, _ in self._delivery_discovery_rows(): + 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"}): diff --git a/loopx/extensions/lark/private_delivery_discovery.py b/loopx/extensions/lark/private_delivery_discovery.py new file mode 100644 index 0000000000..1c61a895c9 --- /dev/null +++ b/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] diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index 8a6fbf0ffe..322d4d3cb2 100644 --- a/tests/test_lark_private_conversations.py +++ b/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 @@ -4969,6 +4970,7 @@ def test_private_discovery_skips_unchanged_bodies_but_not_new_work_or_core_check 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: @@ -4980,7 +4982,7 @@ def observed(path): reads.append(path) return read_json(path) - monkeypatch.setattr(private, "_read_json", observed) + 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}, @@ -4989,7 +4991,7 @@ def observed(path): 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_cache) + 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 @@ -5001,17 +5003,17 @@ def observed(path): assert checks == ["old", "old", "old"] old.unlink() assert transport.pending_delivery_paths() == [new, pending] - assert old not in transport._delivery_discovery_cache + 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 - 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) @@ -5035,7 +5037,7 @@ def test_private_discovery_detects_same_size_updates_with_preserved_mtime(tmp_pa def unavailable(_): raise OSError("read unavailable") - monkeypatch.setattr(private, "_read_json", unavailable) + monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", unavailable) with pytest.raises(OSError, match="read unavailable"): transport.pending_delivery_paths() @@ -5046,6 +5048,7 @@ def test_private_discovery_does_not_cache_a_file_replaced_during_read(tmp_path, 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) @@ -5058,9 +5061,9 @@ def replaced(path): _atomic_write_json(path, {"status": "pending"}) return row - monkeypatch.setattr(private, "_read_json", replaced) + monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", replaced) assert transport.pending_delivery_paths() == [] - monkeypatch.setattr(private, "_read_json", read_json) + monkeypatch.setattr("loopx.extensions.lark.private_delivery_discovery._read_json", read_json) assert transport.pending_delivery_paths() == [path] @@ -5081,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 @@ -5160,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 @@ -5232,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]))