perf(chat): seek cached events by sequence

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."""

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,
     };
   }
 }

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"}
+    ]