From 2bd87a002d6cf8cf65fd67a1c3071df79d7991a0 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Wed, 23 Sep 2026 13:03:31 +0800 Subject: [PATCH 1/3] fix(run-journal): allocate writer sequence within append transaction --- api/run_journal.py | 9 +-- .../webui-run-state-consistency-contract.md | 10 +++ tests/test_run_journal_writer_ordering.py | 74 +++++++++++++++++++ 3 files changed, 87 insertions(+), 6 deletions(-) create mode 100644 tests/test_run_journal_writer_ordering.py diff --git a/api/run_journal.py b/api/run_journal.py index 087a08ba..f4185268 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -464,18 +464,15 @@ class RunJournalWriter: # the offline-gap coverage and replay-cursor contiguity checks rely on. if str(event_name or "").strip() in REPLAY_SKIPPED_SSE_EVENTS: return None - # Draw from the shared module-level seq cache under the per-path lock so - # this writer and any direct append_run_event() call on the same path - # agree on one monotonic, gapless sequence. - with self._lock: - seq = _reserve_next_seq(self._path) + # Allocate the sequence inside the same per-path transaction that writes + # the row. Reserving here, then releasing the lock before append, lets a + # concurrent writer put a higher sequence on disk first. return append_run_event( self.session_id, self.run_id, event_name, payload or {}, session_dir=self.session_dir, - seq=seq, ) diff --git a/docs/rfcs/webui-run-state-consistency-contract.md b/docs/rfcs/webui-run-state-consistency-contract.md index d1362548..9128e910 100644 --- a/docs/rfcs/webui-run-state-consistency-contract.md +++ b/docs/rfcs/webui-run-state-consistency-contract.md @@ -30,6 +30,16 @@ This RFC defines a consistency contract for those layers. It complements the larger run adapter direction in #1925 by documenting what must remain coherent while WebUI still has multiple overlapping state stores. +## Run-journal sequence publication + +Within one WebUI process, auto-numbered appends to the same journal allocate +sequence numbers and write their rows under the same per-path lock. +`RunJournalWriter` delegates both operations to `append_run_event`; it must not +reserve a sequence and release the lock before the physical append. Otherwise +individually valid rows can reach disk out of order and the session replay +reader must reject them as noncontiguous. This does not change caller-supplied +sequence semantics, cross-process ownership, or failed-write recovery. + ## Inactive compression continuation recovery The Agent profile's SQLite compression lineage owns the canonical continuation, diff --git a/tests/test_run_journal_writer_ordering.py b/tests/test_run_journal_writer_ordering.py new file mode 100644 index 00000000..432ea067 --- /dev/null +++ b/tests/test_run_journal_writer_ordering.py @@ -0,0 +1,74 @@ +"""Sequence allocation and physical append must share one journal lock. + +Only scheduling at the real append boundary is controlled. Writers, sequence +allocation, on-disk JSONL and the session replay reader are production code. +""" +from concurrent.futures import ThreadPoolExecutor +import threading + +import pytest + +from api import run_journal + + +@pytest.mark.parametrize("competitor", ["same_writer", "other_writer", "free_function"]) +def test_writer_sequence_follows_physical_append_order(tmp_path, monkeypatch, competitor): + sid, rid = "writer_order_session", "writer_order_run" + writer = run_journal.RunJournalWriter(sid, rid, session_dir=tmp_path) + other = run_journal.RunJournalWriter(sid, rid, session_dir=tmp_path) + reached = threading.Event() + release = threading.Event() + real_append = run_journal.append_run_event + + def pause_first_append(session_id, run_id, event_name, payload=None, **kwargs): + if payload == {"text": "paused"}: + reached.set() + assert release.wait(5), "test did not release the paused append" + return real_append(session_id, run_id, event_name, payload, **kwargs) + + monkeypatch.setattr(run_journal, "append_run_event", pause_first_append) + + def append_competitor(): + if competitor == "free_function": + return run_journal.append_run_event( + sid, rid, "token", {"text": "overtaking"}, session_dir=tmp_path + ) + target = writer if competitor == "same_writer" else other + return target.append_sse_event("token", {"text": "overtaking"}) + + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(writer.append_sse_event, "token", {"text": "paused"}) + try: + assert reached.wait(5), "writer did not reach the append boundary" + second = pool.submit(append_competitor).result(timeout=5) + finally: + release.set() + first_result = first.result(timeout=5) + + journal = run_journal.read_run_events(sid, rid, session_dir=tmp_path) + # Deliberately do not sort: the reader requires physical contiguous order. + assert [row["seq"] for row in journal["events"]] == [1, 2] + assert [row["payload"]["text"] for row in journal["events"]] == ["overtaking", "paused"] + assert second["event_id"] == f"{rid}:1" + assert first_result["event_id"] == f"{rid}:2" + replay = run_journal.read_session_run_events( + sid, after_event_id=second["event_id"], session_dir=tmp_path + ) + assert replay["status"] == "ok" + assert [row["event_id"] for row in replay["events"]] == [first_result["event_id"]] + + +def test_writer_rejected_empty_name_does_not_reserve_a_sequence(tmp_path): + writer = run_journal.RunJournalWriter("name_session", "name_run", session_dir=tmp_path) + with pytest.raises(ValueError, match="event_name is required"): + writer.append_sse_event(" ", {"text": "rejected"}) + event = writer.append_sse_event("token", {"text": "accepted"}) + assert event["seq"] == 1 + assert event["event_id"] == "name_run:1" + + +def test_metering_skip_still_does_not_reserve_or_write(tmp_path): + writer = run_journal.RunJournalWriter("meter_session", "meter_run", session_dir=tmp_path) + assert writer.append_sse_event("metering", {"tps": 10}) is None + assert not (tmp_path / "_run_journal" / "meter_session" / "meter_run.jsonl").exists() + assert writer.append_sse_event("token", {"text": "kept"})["seq"] == 1 From 68a068c2e5345be1e4c12f68485809f234419497 Mon Sep 17 00:00:00 2001 From: Frank Song Date: Wed, 23 Sep 2026 13:10:19 +0800 Subject: [PATCH 2/3] fix(run-journal): drop obsolete writer lock state --- api/run_journal.py | 2 -- tests/test_run_journal_writer_ordering.py | 7 +++++++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/api/run_journal.py b/api/run_journal.py index f4185268..48cd8ef7 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -451,8 +451,6 @@ class RunJournalWriter: self.session_id = _validate_id(session_id, "session_id") self.run_id = _validate_id(run_id, "run_id") self.session_dir = Path(session_dir) if session_dir is not None else None - self._path = _run_path(self.session_id, self.run_id, session_dir=self.session_dir) - self._lock = _lock_for(self._path) def append_sse_event(self, event_name: str, payload=None) -> dict | None: # Live-UI-only telemetry (metering) has no recovery value in the journal: diff --git a/tests/test_run_journal_writer_ordering.py b/tests/test_run_journal_writer_ordering.py index 432ea067..1956fc0e 100644 --- a/tests/test_run_journal_writer_ordering.py +++ b/tests/test_run_journal_writer_ordering.py @@ -72,3 +72,10 @@ def test_metering_skip_still_does_not_reserve_or_write(tmp_path): assert writer.append_sse_event("metering", {"tps": 10}) is None assert not (tmp_path / "_run_journal" / "meter_session" / "meter_run.jsonl").exists() assert writer.append_sse_event("token", {"text": "kept"})["seq"] == 1 + + +def test_unused_writer_does_not_allocate_a_registry_lock(tmp_path): + parent = str(tmp_path / run_journal.RUN_JOURNAL_DIR_NAME / "unused_session") + writer = run_journal.RunJournalWriter("unused_session", "unused_run", session_dir=tmp_path) + assert writer.append_sse_event("metering", {"tps": 10}) is None + assert not any(key[0] == parent for key in run_journal._WRITER_LOCKS) From 99e48f11c202e516bc7c2e9a262c54abb167e1cb Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Wed, 23 Sep 2026 06:23:42 +0000 Subject: [PATCH 3/3] docs(changelog): #7751 --- CHANGELOG.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5602de1a..f02b6d07 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,11 @@ ### Fixed +- **A session's run journal can no longer be written out of order.** The journal writer reserved a + sequence number under the per-path lock, released it, and then appended, so two concurrent writers + could land sequence N+1 on disk before N and the replay reader would stop at the gap + (`replay_noncontiguous`). Sequence allocation and the physical append now happen under the same + existing lock. (#7751 by @franksong2702) - **Reconnecting to a running session no longer redraws every tool card over and over.** After a reconnect restored the live activity scene from the run journal, the client also replayed its older cached in-flight tool list on top, so a turn with N tool cards redrew them N times. When the