diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py index c7b2723467..de63f200fd 100644 --- a/loopx/extensions/lark/private_conversations.py +++ b/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"} 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 08995cace5..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 @@ -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]))