From 379b5a2c615adac003c7f84ee5d2b014c2a0dd0d Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Tue, 8 Sep 2026 20:07:48 +0800 Subject: [PATCH 1/3] perf(state): batch append event log writes Signed-off-by: duanjialing.777 --- loopx/event_sourced_state.py | 58 ++++++++++----- tests/test_event_sourced_state_store.py | 95 +++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 16 deletions(-) diff --git a/loopx/event_sourced_state.py b/loopx/event_sourced_state.py index 1a6483b8b4..25a4e5d7eb 100644 --- a/loopx/event_sourced_state.py +++ b/loopx/event_sourced_state.py @@ -581,24 +581,50 @@ 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 not isinstance(events, (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" + ) + 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]]: diff --git a/tests/test_event_sourced_state_store.py b/tests/test_event_sourced_state_store.py index e4b322d6b9..6764e2c95a 100644 --- a/tests/test_event_sourced_state_store.py +++ b/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,99 @@ 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", + ] + + @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" From 8a153664bae384287cc488838c60820eec4298f3 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Thu, 10 Sep 2026 12:09:47 +0800 Subject: [PATCH 2/3] fix(state): preserve batch append boundaries Signed-off-by: duanjialing.777 --- loopx/event_sourced_state.py | 3 +- tests/test_event_sourced_state_store.py | 81 +++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 1 deletion(-) diff --git a/loopx/event_sourced_state.py b/loopx/event_sourced_state.py index 25a4e5d7eb..e3283c7b74 100644 --- a/loopx/event_sourced_state.py +++ b/loopx/event_sourced_state.py @@ -584,7 +584,7 @@ def append(self, event: dict[str, Any]) -> dict[str, Any]: return self.append_many((event,))[0] def append_many(self, events: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: - if not isinstance(events, (list, tuple)): + if type(events) not in (list, tuple): return [self.append(event) for event in events] if not events: return [] @@ -618,6 +618,7 @@ def append_many(self, events: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: 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 diff --git a/tests/test_event_sourced_state_store.py b/tests/test_event_sourced_state_store.py index 6764e2c95a..062b516f53 100644 --- a/tests/test_event_sourced_state_store.py +++ b/tests/test_event_sourced_state_store.py @@ -127,6 +127,87 @@ def lazy_events(): ] +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" From 532492747577317f2f71f53d1694a3fd2df67967 Mon Sep 17 00:00:00 2001 From: "duanjialing.777" Date: Thu, 10 Sep 2026 13:13:11 +0800 Subject: [PATCH 3/3] 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, }; } }