From 8ad13ff703aac08d1981d3ec165be9c08b910d01 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:32:05 +0800 Subject: [PATCH] fix(lark): isolate unreadable private delivery journals Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../lark/private_delivery_discovery.py | 43 +++++++++----- tests/test_lark_private_conversations.py | 58 +++++++++++++++++++ 2 files changed, 85 insertions(+), 16 deletions(-) diff --git a/loopx/extensions/lark/private_delivery_discovery.py b/loopx/extensions/lark/private_delivery_discovery.py index 24794af71e..673409f5bd 100644 --- a/loopx/extensions/lark/private_delivery_discovery.py +++ b/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 diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py index d9630e8153..8d83e75a44 100644 --- a/tests/test_lark_private_conversations.py +++ b/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"]