Keep scanning private replies when a delivery journal is unreadable

loopx/extensions/lark/private_delivery_discovery.py

@@ -43,23 +43,34 @@ def rows(self) -> Iterator[tuple[Path, dict[str, Any], bool]]:
 
         for path in sorted(self.root.glob("*.json")):
             seen.add(path)
-            before = file_change_fingerprint(path)
-            saved = self._completed.get(path)
-            if before is not None and saved is not None and saved[0] == before:
-                yield path, saved[1], saved[2]
+            try:
+                row, repairs = self._observe(path)
+            except (OSError, ValueError, KeyError, TypeError, AttributeError):
+                # Keep unreadable or disappearing journals intact and evict
+                # their stale hints. The next scan can observe a readable
+                # receipt; unrelated replies stay discoverable now.
+                self._completed.pop(path, None)
                 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 before is not None and file_change_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]
+
+    def _observe(self, path: Path) -> tuple[dict[str, Any], bool]:
+        before = file_change_fingerprint(path)
+        saved = self._completed.get(path)
+        if before is not None and saved is not None and saved[0] == before:
+            return saved[1], saved[2]
+
+        self._completed.pop(path, None)
+        row = _read_json(path)
+        if row["status"] != "delivered":
+            return row, False
+
+        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 before is not None and file_change_fingerprint(path) == before:
+            self._completed[path] = (before, view, repairs)
+        return view, repairs

tests/test_lark_private_conversations.py

@@ -2469,6 +2469,64 @@ def _drain_app_projection(transport, predicate):
         time.sleep(.02)
 
 
+@pytest.mark.parametrize("fault", ["truncated_json", "invalid_utf8", "vanished_during_scan"])
+def test_unreadable_private_delivery_does_not_hold_other_original_replies(ordinary, monkeypatch, fault):  # noqa: F811
+    from loopx.extensions.lark import private_delivery_discovery
+
+    _store, runtime, provider, transport = connect(ordinary)
+    originals = {}
+    try:
+        for profile in ("notes-app", "steward-app"):
+            event = provider.event(profile, f"independent-{profile}", f"Complete the original {profile} request")
+            transport.admit(profile, event)
+            request = identity_ref(transport._binding(profile)["provider_ref"], event["message_id"])
+            native = transport.core.read_request(request)
+            result = runtime.wait_for_turn(session_id=native["session_id"], turn_id=native["turn_id"], timeout_sec=10)
+            assert result["status"] == "completed"
+            originals[profile] = (transport.root / f"{request}.json", event)
+
+        held, held_event = originals["notes-app"]
+        healthy, healthy_event = originals["steward-app"]
+        saved = held.read_bytes()
+        with monkeypatch.context() as damage:
+            if fault == "vanished_during_scan":
+                fingerprint = private_delivery_discovery.file_change_fingerprint
+
+                def vanished(path):
+                    if path == held:
+                        raise FileNotFoundError(path)
+                    return fingerprint(path)
+
+                damage.setattr(private_delivery_discovery, "file_change_fingerprint", vanished)
+            else:
+                held.write_bytes(b'{"status":' if fault == "truncated_json" else b"\xff")
+            unchanged = held.read_bytes()
+            for discover in (transport.pending_body_paths, transport.pending_delivery_paths):
+                assert discover() == [healthy]
+
+            requests = ordinary[4].read_bytes()
+            writes_before = len(provider.writes)
+            transport.reconcile()
+            final = json.loads(healthy.read_text())["deliveries"]["terminal"]
+            assert final["verified"]
+            assert provider.messages[final["attempt"]["message_ref"]]["chat_id"] == healthy_event["chat_id"]
+            assert held.read_bytes() == unchanged
+            assert all(profile != "notes-app" for profile, _ in provider.writes[writes_before:])
+
+        if fault != "vanished_during_scan":
+            held.write_bytes(saved)
+        transport.reconcile()
+        final = json.loads(held.read_text())["deliveries"]["terminal"]
+        assert final["verified"]
+        assert provider.messages[final["attempt"]["message_ref"]]["chat_id"] == held_event["chat_id"]
+        sent = list(provider.writes)
+        transport.reconcile()
+        assert provider.writes == sent
+        assert ordinary[4].read_bytes() == requests
+    finally:
+        runtime.close()
+
+
 def test_app_messages_and_answers_reach_the_original_chat_once_without_new_execution(ordinary):  # noqa: F811
     store, runtime, provider, transport, original, turns, before = _app_projection_journey(ordinary)
     sid = original["session_id"]