From bbaccd46cb5d0b38488943270660af1bef2e2436 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Wed, 9 Sep 2026 08:09:21 +0800 Subject: [PATCH 1/2] perf(chat): seek cached events by sequence Signed-off-by: duanjialing.777 --- loopx/chat_store.py | 12 +++++++++--- tests/test_chat_event_cursor.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 3 deletions(-) create mode 100644 tests/test_chat_event_cursor.py diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 39602e1847..56d6823594 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -2,6 +2,7 @@ from __future__ import annotations +from bisect import bisect_right from datetime import datetime, timedelta, timezone import json import os @@ -1335,14 +1336,19 @@ def events_after(self, session_id: str, turn_id: str, event_id: str | None) -> l with self._event_lock: cached = self._event_cache.get(key) rows = ( - list(cached) + cached if cached is not None and self._event_cache_revision.get(key) == revision else None ) if rows is None: with exclusive_file_lock(path, agent_id="loopx-chat", operation="read_chat_events"): - rows = list(self._event_rows_locked(session_id, turn_id)) - return [row for row in rows if int(row.get("sequence") or 0) > after] + rows = self._event_rows_locked(session_id, turn_id) + start = bisect_right( + rows, + after, + key=lambda row: int(row.get("sequence") or 0), + ) + return rows[start:] def compact_completed_events(self, *, older_than_hours: float = 24.0) -> int: """Drop replay-only deltas after the durable final message is old enough.""" diff --git a/tests/test_chat_event_cursor.py b/tests/test_chat_event_cursor.py new file mode 100644 index 0000000000..4d8229f037 --- /dev/null +++ b/tests/test_chat_event_cursor.py @@ -0,0 +1,31 @@ +from __future__ import annotations + +from pathlib import Path + +from loopx.chat_store import ChatSessionStore + + +def test_events_after_does_not_scan_the_cached_prefix(tmp_path: Path) -> None: + class NonIterableRows(list[dict[str, object]]): + def __iter__(self): + raise AssertionError("events_after scanned the cached prefix") + + store = ChatSessionStore(tmp_path) + session_id = "session" + turn_id = "turn" + event_path = store._event_path(session_id, turn_id) + event_path.parent.mkdir(parents=True) + event_path.touch() + key = (session_id, turn_id) + store._event_cache[key] = NonIterableRows( + [ + {"sequence": 1, "event_id": "1"}, + {"sequence": 4, "event_id": "4"}, + {"sequence": 7, "event_id": "7"}, + ] + ) + store._event_cache_revision[key] = store._event_revision(event_path) + + assert store.events_after(session_id, turn_id, "4") == [ + {"sequence": 7, "event_id": "7"} + ] From 2fb24a891325b53f4191669f18e2240c5706dce5 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Thu, 10 Sep 2026 13:13:11 +0800 Subject: [PATCH 2/2] fix(coordination): stabilize fence read errors Signed-off-by: duanjialing.777 --- loopx/control_plane/coordination/legacy_writer_fence.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/loopx/control_plane/coordination/legacy_writer_fence.ts b/loopx/control_plane/coordination/legacy_writer_fence.ts index 8a8f4507e1..81af31fbf1 100644 --- a/loopx/control_plane/coordination/legacy_writer_fence.ts +++ b/loopx/control_plane/coordination/legacy_writer_fence.ts @@ -142,10 +142,12 @@ export async function loadLegacyCoordinationWriterFence( return { status: "loaded", fence }; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return { status: "missing" }; + const path = (error as NodeJS.ErrnoException).path; + const reason = error instanceof Error ? error.message : "legacy writer fence read failed"; return { status: "failed", reason_code: "legacy_writer_fence_read_failed", - reason: error instanceof Error ? error.message : "legacy writer fence read failed", + reason: typeof path === "string" ? reason.replace(` '${path}'`, "") : reason, }; } }