perf(state): batch append event log writes

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

loopx/event_sourced_state.py

@@ -581,24 +581,51 @@ def load(self) -> list[dict[str, Any]]:
         return _dedupe_events(events)
 
     def append(self, event: dict[str, Any]) -> dict[str, Any]:
-        with exclusive_file_lock(self.path):
-            events = self.load()
-            existing = {item["event_id"]: item for item in events}
-            next_sequence = max((int(item["append_sequence"]) for item in events), default=0) + 1
-            normalized = normalize_state_event(event, append_sequence=next_sequence)
-            prior = existing.get(normalized["event_id"])
-            if prior is not None:
-                if event_fingerprint(prior) != event_fingerprint(normalized):
-                    raise StateEventConflictError(f"conflicting event_id: {normalized['event_id']}")
-                return prior
-
-            self.path.parent.mkdir(parents=True, exist_ok=True)
-            with self.path.open("a", encoding="utf-8") as stream:
-                stream.write(json.dumps(normalized, sort_keys=True, ensure_ascii=False) + "\n")
-            return normalized
+        return self.append_many((event,))[0]
 
     def append_many(self, events: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:
-        return [self.append(event) for event in events]
+        if type(events) not in (list, tuple):
+            return [self.append(event) for event in events]
+        if not events:
+            return []
+
+        with exclusive_file_lock(self.path):
+            stored = self.load()
+            existing = {item["event_id"]: item for item in stored}
+            next_sequence = max(
+                (int(item["append_sequence"]) for item in stored), default=0
+            ) + 1
+            appended: list[dict[str, Any]] = []
+            stream = None
+            try:
+                for event in events:
+                    normalized = normalize_state_event(
+                        event,
+                        append_sequence=next_sequence,
+                    )
+                    prior = existing.get(normalized["event_id"])
+                    if prior is not None:
+                        if event_fingerprint(prior) != event_fingerprint(normalized):
+                            raise StateEventConflictError(
+                                f"conflicting event_id: {normalized['event_id']}"
+                            )
+                        appended.append(prior)
+                        continue
+
+                    if stream is None:
+                        self.path.parent.mkdir(parents=True, exist_ok=True)
+                        stream = self.path.open("a", encoding="utf-8")
+                    stream.write(
+                        json.dumps(normalized, sort_keys=True, ensure_ascii=False) + "\n"
+                    )
+                    stream.flush()
+                    existing[normalized["event_id"]] = normalized
+                    appended.append(normalized)
+                    next_sequence += 1
+            finally:
+                if stream is not None:
+                    stream.close()
+            return appended
 
 
 def _dedupe_events(events: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:

tests/test_event_sourced_state_store.py

@@ -1,8 +1,10 @@
 import json
+from contextlib import contextmanager
 from pathlib import Path
 
 import pytest
 
+import loopx.event_sourced_state as event_sourced_state
 from loopx.event_sourced_state import (
     TODO_ADDED,
     AppendOnlyStateEventStore,
@@ -32,6 +34,180 @@ def test_load_observes_events_appended_by_another_store(tmp_path: Path) -> None:
     assert reader.load() == [appended]
 
 
+def test_append_many_loads_once_and_preserves_idempotent_sequence(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store = AppendOnlyStateEventStore(tmp_path / "events.jsonl")
+    events = [
+        make_state_event(
+            event_id=f"event-{index}",
+            goal_id="goal-a",
+            event_type=TODO_ADDED,
+            refs={"todo_id": f"todo_event_{index}"},
+            payload={"role": "agent", "title": f"Event {index}"},
+            recorded_at="2026-09-08T00:00:00Z",
+        )
+        for index in range(3)
+    ]
+    first = store.append(events[0])
+    load = store.load
+    load_count = 0
+
+    def counted_load() -> list[dict[str, object]]:
+        nonlocal load_count
+        load_count += 1
+        return load()
+
+    monkeypatch.setattr(store, "load", counted_load)
+    appended = store.append_many([events[1], events[1], events[0], events[2]])
+
+    assert load_count == 1
+    assert [event["append_sequence"] for event in appended] == [2, 2, 1, 3]
+    assert [event["event_id"] for event in load()] == [
+        "event-0",
+        "event-1",
+        "event-2",
+    ]
+    assert appended[2] == first
+
+
+def test_append_many_preserves_lazy_iterable_visibility_and_reentrancy(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store = AppendOnlyStateEventStore(tmp_path / "events.jsonl")
+    events = [
+        make_state_event(
+            event_id=f"lazy-event-{index}",
+            goal_id="goal-a",
+            event_type=TODO_ADDED,
+            refs={"todo_id": f"todo_lazy_event_{index}"},
+            payload={"role": "agent", "title": f"Lazy event {index}"},
+            recorded_at="2026-09-09T00:00:00Z",
+        )
+        for index in range(3)
+    ]
+    lock_held = False
+
+    @contextmanager
+    def non_reentrant_lock(_path: Path):
+        nonlocal lock_held
+        assert not lock_held
+        lock_held = True
+        try:
+            yield
+        finally:
+            lock_held = False
+
+    monkeypatch.setattr(
+        event_sourced_state,
+        "exclusive_file_lock",
+        non_reentrant_lock,
+    )
+    observed_prefix: list[str] = []
+
+    def lazy_events():
+        yield events[0]
+        observed_prefix.extend(event["event_id"] for event in store.load())
+        store.append(events[1])
+        yield events[2]
+
+    appended = store.append_many(lazy_events())
+
+    assert observed_prefix == ["lazy-event-0"]
+    assert [event["event_id"] for event in appended] == [
+        "lazy-event-0",
+        "lazy-event-2",
+    ]
+    assert [event["event_id"] for event in store.load()] == [
+        "lazy-event-0",
+        "lazy-event-1",
+        "lazy-event-2",
+    ]
+
+
+def test_append_many_does_not_iterate_list_subclasses_under_lock(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    store = AppendOnlyStateEventStore(tmp_path / "events.jsonl")
+    events = [
+        make_state_event(
+            event_id=f"subclass-event-{index}",
+            goal_id="goal-a",
+            event_type=TODO_ADDED,
+            refs={"todo_id": f"todo_subclass_event_{index}"},
+            payload={"role": "agent", "title": f"Subclass event {index}"},
+            recorded_at="2026-09-10T00:00:00Z",
+        )
+        for index in range(2)
+    ]
+    lock_held = False
+
+    @contextmanager
+    def non_reentrant_lock(_path: Path):
+        nonlocal lock_held
+        assert not lock_held
+        lock_held = True
+        try:
+            yield
+        finally:
+            lock_held = False
+
+    monkeypatch.setattr(event_sourced_state, "exclusive_file_lock", non_reentrant_lock)
+
+    class ReentrantList(list):
+        def __iter__(self):
+            store.append(events[1])
+            return super().__iter__()
+
+    appended = store.append_many(ReentrantList([events[0]]))
+
+    assert [event["event_id"] for event in appended] == ["subclass-event-0"]
+    assert [event["event_id"] for event in store.load()] == [
+        "subclass-event-1",
+        "subclass-event-0",
+    ]
+
+
+def test_append_many_flushes_each_event_before_processing_the_next(
+    tmp_path: Path,
+) -> None:
+    event_log = tmp_path / "events.jsonl"
+    store = AppendOnlyStateEventStore(event_log)
+    first = make_state_event(
+        event_id="flush-event-0",
+        goal_id="goal-a",
+        event_type=TODO_ADDED,
+        refs={"todo_id": "todo_flush_event_0"},
+        payload={"role": "agent", "title": "First event"},
+        recorded_at="2026-09-10T00:00:00Z",
+    )
+    observed_prefix: list[str] = []
+
+    class ObservingEvent(dict):
+        def get(self, key, default=None):
+            if not observed_prefix:
+                observed_prefix.append(event_log.read_text(encoding="utf-8"))
+            return super().get(key, default)
+
+    second = ObservingEvent(
+        make_state_event(
+            event_id="flush-event-1",
+            goal_id="goal-a",
+            event_type=TODO_ADDED,
+            refs={"todo_id": "todo_flush_event_1"},
+            payload={"role": "agent", "title": "Second event"},
+            recorded_at="2026-09-10T00:00:00Z",
+        )
+    )
+
+    store.append_many([first, second])
+
+    assert '"event_id": "flush-event-0"' in observed_prefix[0]
+
+
 @pytest.mark.parametrize("sequence", [True, False, 1.5, "2"])
 def test_load_rejects_non_integer_append_sequence(tmp_path: Path, sequence: object) -> None:
     event_log = tmp_path / "events.jsonl"