From 0ac2b3a3fab1e3a6f7e3174ef2f981496eb42910 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:39:16 +0200 Subject: [PATCH 1/2] Cache next-seq per run journal path to avoid O(n^2) appends MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- api/run_journal.py | 77 +++++++++++- tests/test_run_journal_seq_cache.py | 175 ++++++++++++++++++++++++++++ 2 files changed, 247 insertions(+), 5 deletions(-) create mode 100644 tests/test_run_journal_seq_cache.py diff --git a/api/run_journal.py b/api/run_journal.py index 5098a133..81e92e75 100644 --- a/api/run_journal.py +++ b/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 diff --git a/tests/test_run_journal_seq_cache.py b/tests/test_run_journal_seq_cache.py new file mode 100644 index 00000000..23a18fb3 --- /dev/null +++ b/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)) From cbf949f6a8c4b11dfe4be64721ee4cd3a77534a9 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Thu, 9 Jul 2026 03:49:40 +0000 Subject: [PATCH 2/2] docs(changelog): #5799 run-journal next-seq cache --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d2ea7512..8e3595b5 100644 --- a/CHANGELOG.md +++ b/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)