Cache next-seq per run journal path to avoid O(n^2) appends

append_run_event() without an explicit seq called _next_seq(path), which
re-read and re-parsed the entire journal file on every append — O(n) per
append, O(n^2) over a run. Direct callers therefore paid a full file
reparse per event; only RunJournalWriter avoided it via its own in-memory
counter.

Introduce a module-level _SEQ_CACHE keyed by run-journal file path. The
per-path _lock_for(path) serializes same-path reserve->append so seqs
stay monotonic and file order matches; both RunJournalWriter and the free
append_run_event now draw from this one cache (_reserve_next_seq), so
seqs stay gapless even when both write the same path — the writer's
private counter is removed to keep a single source of truth. An explicit
caller-supplied seq advances the cache (_note_assigned_seq) so it can
never be re-issued. delete_run_journal() now also evicts cache entries
for the removed session so a run re-created at the same path restarts
numbering at 1.

Guard every structural _SEQ_CACHE access (reserve/note/evict) with a
dedicated _SEQ_CACHE_LOCK. The eviction sweep in delete_run_journal
iterates the whole dict; the per-path locks the append path holds do NOT
serialize against it, so a concurrent append on another session inserted
a fresh key mid-iteration and raised "dictionary changed size during
iteration" (the caller swallows it, but the eviction then aborts and
leaks stale seqs, so a re-created run resumed a stale number instead of
restarting at 1). A stress test races deletes against cross-session
appends and asserts no such error.

Rollback: revert this commit; _next_seq(path) re-read behaviour returns.
No on-disk format change, so existing journals stay compatible.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

api/run_journal.py

@@ -17,6 +17,16 @@ RUN_JOURNAL_DIR_NAME = "_run_journal"
 _SAFE_ID_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
 _WRITER_LOCKS: dict[tuple[str, str, str], threading.Lock] = {}
 _WRITER_LOCKS_GUARD = threading.Lock()
+# Next-seq to assign per run-journal file path, kept in memory so repeat appends
+# to the same run do not re-parse the whole file on every call. The per-path
+# ``_lock_for(path)`` serializes same-path reserve→append so seqs stay monotonic
+# and file order matches; ``_SEQ_CACHE_LOCK`` (below) additionally guards every
+# *structural* access to the dict (reserve/note/evict) so ``delete_run_journal``
+# can iterate + drop keys while a concurrent append on ANOTHER path inserts one,
+# without a ``dictionary changed size during iteration`` crash. See
+# ``_reserve_next_seq`` and ``delete_run_journal`` (which evicts stale entries).
+_SEQ_CACHE: dict[str, int] = {}
+_SEQ_CACHE_LOCK = threading.Lock()
 _TERMINAL_SSE_EVENTS = {"done", "cancel", "apperror", "error", "stream_end"}
 _FSYNC_MODE_ENV = "HERMES_WEBUI_RUN_JOURNAL_FSYNC"
 _FSYNC_MODE_EAGER = "eager"
@@ -81,6 +91,49 @@ def _next_seq(path: Path) -> int:
     return (max(seqs) + 1) if seqs else 1
 
 
+def _reserve_next_seq(path: Path) -> int:
+    """Reserve and return the next seq for ``path``, advancing the in-memory cache.
+
+    Callers MUST hold ``_lock_for(path)``. The first append per path in this
+    process seeds the cache from ``_next_seq(path)`` (one file read); every later
+    append is a pure in-memory increment, avoiding the O(n) re-parse that
+    re-reading the whole journal on every append caused (O(n^2) over a run).
+    Because ``RunJournalWriter`` and the free ``append_run_event`` share this one
+    cache under the same per-path lock, their seqs stay monotonic and gapless
+    even when both write the same path. ``_SEQ_CACHE_LOCK`` additionally makes the
+    dict get+set atomic against a concurrent cross-path eviction.
+    """
+    key = str(path)
+    with _SEQ_CACHE_LOCK:
+        nxt = _SEQ_CACHE.get(key)
+        if nxt is not None:
+            _SEQ_CACHE[key] = nxt + 1
+            return nxt
+    # Cache miss: seed from disk WITHOUT holding the module-global lock, so a
+    # slow first-access file read for one path can't block every other path's
+    # cache ops. The caller holds the per-path lock, so only one thread per path
+    # can reach this branch — no double-seed, and no same-path writer can race
+    # the value in between.
+    seeded = _next_seq(path)
+    with _SEQ_CACHE_LOCK:
+        _SEQ_CACHE[key] = seeded + 1
+        return seeded
+
+
+def _note_assigned_seq(path: Path, seq: int) -> None:
+    """Keep the cache at least one past an explicitly-supplied ``seq``.
+
+    Callers MUST hold ``_lock_for(path)``. When an append carries a caller-chosen
+    ``seq`` rather than drawing from the cache, advance the cache so a later
+    cache-based append on the same path cannot re-issue an already-used seq.
+    """
+    key = str(path)
+    nxt = int(seq) + 1
+    with _SEQ_CACHE_LOCK:
+        if _SEQ_CACHE.get(key, 0) < nxt:
+            _SEQ_CACHE[key] = nxt
+
+
 def _terminal_state_for_event(event_name: str, payload) -> str | None:
     name = str(event_name or "")
     if name == "done" or name == "stream_end":
@@ -145,7 +198,11 @@ def append_run_event(
     if not event_name:
         raise ValueError("event_name is required")
     with _lock_for(path):
-        assigned_seq = int(seq) if seq is not None else _next_seq(path)
+        if seq is not None:
+            assigned_seq = int(seq)
+            _note_assigned_seq(path, assigned_seq)
+        else:
+            assigned_seq = _reserve_next_seq(path)
         terminal_state = _terminal_state_for_event(event_name, payload)
         event = {
             "version": 1,
@@ -183,13 +240,13 @@ class RunJournalWriter:
         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)
-        with self._lock:
-            self._next_seq = _next_seq(self._path)
 
     def append_sse_event(self, event_name: str, payload=None) -> dict:
+        # 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 = self._next_seq
-            self._next_seq += 1
+            seq = _reserve_next_seq(self._path)
         return append_run_event(
             self.session_id,
             self.run_id,
@@ -300,6 +357,16 @@ def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> b
         with _WRITER_LOCKS_GUARD:
             for key in [k for k in _WRITER_LOCKS if k[0] == dir_key]:
                 del _WRITER_LOCKS[key]
+        # Drop cached next-seq entries for the removed runs too. Every run file
+        # for this session lives directly under ``session_journal_dir``, so its
+        # cache key's parent dir matches. Without this, a run re-created at the
+        # same path would resume the stale cached seq instead of restarting at 1.
+        # Hold ``_SEQ_CACHE_LOCK`` — the SAME mutex ``_reserve_next_seq``/
+        # ``_note_assigned_seq`` take — so a concurrent append on another path
+        # cannot mutate the dict mid-iteration (``dictionary changed size``).
+        with _SEQ_CACHE_LOCK:
+            for key in [k for k in _SEQ_CACHE if str(Path(k).parent) == dir_key]:
+                del _SEQ_CACHE[key]
     return removed
 
 

tests/test_run_journal_seq_cache.py

@@ -0,0 +1,175 @@
+"""Regression tests for the in-memory run-journal seq cache.
+
+Repeat ``append_run_event`` calls used to re-read (and re-parse) the entire
+journal file on every append via ``_next_seq``, which is O(n) per append and
+O(n^2) over a run. The cache seeds once per path and then increments in memory
+while staying consistent with ``RunJournalWriter`` (both share one cache under
+the same per-path lock).
+"""
+import threading
+
+import pytest  # noqa: F401  # top-level import keeps pytest collection unambiguous
+
+from api import run_journal
+
+
+def test_append_run_event_seeds_seq_once_and_stays_gapless(tmp_path, monkeypatch):
+    calls = {"next_seq": 0, "read_jsonl": 0}
+    real_next_seq = run_journal._next_seq
+    real_read_jsonl = run_journal._read_jsonl
+
+    def counting_next_seq(path):
+        calls["next_seq"] += 1
+        return real_next_seq(path)
+
+    def counting_read_jsonl(path):
+        calls["read_jsonl"] += 1
+        return real_read_jsonl(path)
+
+    monkeypatch.setattr(run_journal, "_next_seq", counting_next_seq)
+    monkeypatch.setattr(run_journal, "_read_jsonl", counting_read_jsonl)
+
+    n = 25
+    seqs = [
+        run_journal.append_run_event(
+            "sess_cache", "run_cache", "token", {"text": str(i)}, session_dir=tmp_path
+        )["seq"]
+        for i in range(n)
+    ]
+
+    assert seqs == list(range(1, n + 1))
+    # Seeded from the file exactly once; every later append is in-memory only.
+    assert calls["next_seq"] == 1
+    assert calls["read_jsonl"] <= 1
+
+
+def test_writer_and_free_function_share_one_gapless_sequence(tmp_path):
+    writer = run_journal.RunJournalWriter("sess_shared", "run_shared", session_dir=tmp_path)
+    a = writer.append_sse_event("token", {"text": "a"})
+    b = run_journal.append_run_event(
+        "sess_shared", "run_shared", "token", {"text": "b"}, session_dir=tmp_path
+    )
+    c = writer.append_sse_event("token", {"text": "c"})
+    d = run_journal.append_run_event(
+        "sess_shared", "run_shared", "done", {"session": {}}, session_dir=tmp_path
+    )
+
+    assert [a["seq"], b["seq"], c["seq"], d["seq"]] == [1, 2, 3, 4]
+
+    journal = run_journal.read_run_events("sess_shared", "run_shared", session_dir=tmp_path)
+    file_seqs = sorted(event["seq"] for event in journal["events"])
+    assert file_seqs == [1, 2, 3, 4]
+
+
+def test_explicit_seq_keeps_cache_from_reissuing(tmp_path):
+    # A caller-supplied seq must push the cache forward so a later cache append
+    # does not collide with it.
+    run_journal.append_run_event(
+        "sess_expl", "run_expl", "token", {"text": "x"}, session_dir=tmp_path, seq=5
+    )
+    nxt = run_journal.append_run_event(
+        "sess_expl", "run_expl", "token", {"text": "y"}, session_dir=tmp_path
+    )
+    assert nxt["seq"] == 6
+
+
+def test_delete_evicts_seq_cache_so_recreated_run_restarts(tmp_path):
+    run_journal.append_run_event(
+        "sess_del", "run_del", "token", {"text": "one"}, session_dir=tmp_path
+    )
+    run_journal.append_run_event(
+        "sess_del", "run_del", "token", {"text": "two"}, session_dir=tmp_path
+    )
+
+    assert run_journal.delete_run_journal("sess_del", session_dir=tmp_path) is True
+
+    restarted = run_journal.append_run_event(
+        "sess_del", "run_del", "token", {"text": "fresh"}, session_dir=tmp_path
+    )
+    assert restarted["seq"] == 1
+
+
+def test_delete_evicts_seq_cache_concurrently_without_crash(tmp_path):
+    """delete_run_journal must evict _SEQ_CACHE under a shared lock.
+
+    The eviction iterates the whole ``_SEQ_CACHE`` to drop the deleted session's
+    keys. It ran outside any mutex the append path holds, so a concurrent append
+    on ANOTHER session — which inserts a fresh key — mutated the dict mid-iteration
+    and raised ``RuntimeError: dictionary changed size during iteration``. Both
+    paths now take ``_SEQ_CACHE_LOCK``, so the eviction and inserts serialize.
+    """
+    # Seed the cache with many keys so an eviction sweep iterates a wide dict,
+    # widening the window for a concurrent insert to collide.
+    for s in range(60):
+        run_journal.append_run_event(
+            f"sess_seed{s}", "run", "token", {"text": "x"}, session_dir=tmp_path
+        )
+
+    errors: list[BaseException] = []
+    errors_lock = threading.Lock()
+    stop = threading.Event()
+
+    def deleter():
+        i = 0
+        while not stop.is_set():
+            sid = f"sess_del{i}"
+            try:
+                run_journal.append_run_event(
+                    sid, "run", "token", {"text": "d"}, session_dir=tmp_path
+                )
+                run_journal.delete_run_journal(sid, session_dir=tmp_path)
+            except BaseException as exc:  # noqa: BLE001 - recorded for the assert
+                with errors_lock:
+                    errors.append(exc)
+            i += 1
+
+    def inserter(base):
+        i = 0
+        while not stop.is_set():
+            try:
+                # Each append to a brand-new session inserts a fresh cache key,
+                # racing the deleter's eviction comprehension.
+                run_journal.append_run_event(
+                    f"sess_ins{base}_{i}", "run", "token", {"text": "i"},
+                    session_dir=tmp_path,
+                )
+            except BaseException as exc:  # noqa: BLE001 - recorded for the assert
+                with errors_lock:
+                    errors.append(exc)
+            i += 1
+
+    workers = [threading.Thread(target=deleter)]
+    workers += [threading.Thread(target=inserter, args=(b,)) for b in range(4)]
+    for w in workers:
+        w.start()
+    # Let them contend briefly, then wind down.
+    for _ in range(200):
+        run_journal.delete_run_journal("sess_seed0", session_dir=tmp_path)
+    stop.set()
+    for w in workers:
+        w.join(timeout=10.0)
+
+    assert not any(w.is_alive() for w in workers), "worker threads did not finish"
+    assert not errors, f"eviction raced an insert: {errors[:3]}"
+
+
+def test_concurrent_appends_produce_unique_gapless_seqs(tmp_path):
+    threads = []
+    results: list[int] = []
+    results_lock = threading.Lock()
+
+    def worker(i):
+        event = run_journal.append_run_event(
+            "sess_conc", "run_conc", "token", {"text": str(i)}, session_dir=tmp_path
+        )
+        with results_lock:
+            results.append(event["seq"])
+
+    for i in range(40):
+        threads.append(threading.Thread(target=worker, args=(i,)))
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert sorted(results) == list(range(1, 41))