Merge pull request #5818 from nesquena/release/stage-5799

Release: cache run-journal next-seq per path (#5799)

CHANGELOG.md

@@ -5,6 +5,8 @@
 
 ### Fixed
 
+- **Appending run-journal events no longer gets slower as a session's journal grows.** The next sequence number was recomputed by scanning existing entries on every append (O(n²) over a session's lifetime); it's now cached per journal path (guarded by a dedicated lock, evicted on journal delete), making each append O(1). Thanks @ai-ag2026. (#5799)
+
 - **The login page's connectivity retry timer is now cleared correctly.** The retry poller is started with `setInterval` but was cleared with `clearTimeout`, so once the server came back the interval kept firing (a zombie timer) instead of stopping and reloading once. It now uses `clearInterval`. Thanks @ai-ag2026. (#5801)
 
 - **A background wakeup no longer eats the assistant reply that preceded it.** When a background-process wakeup prompt arrived after an assistant response, the response could disappear from the interactive transcript (it remained in the DB and export). Background wakeups now render as their own distinct "Background wakeup" status row aligned to the message rail — with attachment support — leaving the prior reply intact. Thanks @Isla-Liu. (#5766)

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))