From be29c3f7c0d34dee007b9b1cf8aaaa80c8d71ec4 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Mon, 20 Jul 2026 02:12:32 -0700 Subject: [PATCH] Release: run-journal summary cache (#6291, @sjungwon03) (#6355) * perf(run-journal): cache unchanged run summaries * fix(run-journal): reject cache after missing-file race * run-journal cache: add st_ctime_ns to signature (close same-size mtime-preserving rewrite window) + regression test [maintainer fix on @sjungwon03 #6291] * docs(changelog): stamp #6291 run-journal summary cache --------- Co-authored-by: sjungwon03 Co-authored-by: nesquena-hermes --- CHANGELOG.md | 2 + api/run_journal.py | 98 ++++++++++++++++++++++++++++++-- tests/test_run_journal.py | 115 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 209 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 15a83735..fd7bd432 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,8 @@ ### Fixed +- **Content-search and sidebar polling are faster: run-journal summaries are now cached.** Repeated reads of an unchanged run journal reuse a cached summary instead of re-parsing the whole file, keyed on a complete filesystem signature (device, inode, size, mtime, and ctime) so the cache is always dropped the instant the journal is appended to, replaced, truncated, or recreated — including a same-size rewrite that preserves mtime. Bounded LRU; a first-write race can never pin a stale empty result. Thanks @sjungwon03. (#6291) + - **Live Stream: a settled Worklog now shows the side effects a turn performed (memory/skill saves).** The `state_saved` SSE event (memory + skill saves) was classified as an Assistant-Turn-Anchor side effect but the renderer-neutral scene projection discarded it, so the browser only flashed a toast and a reloaded Worklog lost that record. Anchor `artifacts`/`side_effects` are now projected into the settled `activity_scene_v1`. Purely additive — turns with no side effects render byte-identical (side-effect-only turns still don't force a Worklog), and the change is compatible with the stable-run-identity plumbing. Thanks @franksong2702. (#6204, #3400) - **Kanban board columns are now scrollable on mobile.** On a phone, a touch-drag inside a Kanban column was trapped in the column's nested scroller, so you couldn't scroll the board. The mobile layout now lets the column overscroll chain to the page/section, restoring natural scrolling; desktop and the pull-to-refresh behavior are unchanged. Thanks @jpalazz2. (#6306) diff --git a/api/run_journal.py b/api/run_journal.py index 44f6c34d..81abf2fc 100644 --- a/api/run_journal.py +++ b/api/run_journal.py @@ -10,6 +10,7 @@ import os import re import threading import time +from collections import OrderedDict from pathlib import Path from typing import Any, Iterable @@ -27,6 +28,14 @@ _WRITER_LOCKS_GUARD = threading.Lock() # ``_reserve_next_seq`` and ``delete_run_journal`` (which evicts stale entries). _SEQ_CACHE: dict[str, int] = {} _SEQ_CACHE_LOCK = threading.Lock() +# Summary callers only need terminal state and the latest cursor. Re-parsing a +# completed journal's full payload (which can include multi-megabyte tool or +# session results) on every status/reconnect probe is needless. This process +# cache is keyed by a complete stat identity, so it is never used after an +# atomic replacement, append, truncate, or same-path file recreation. +_SUMMARY_CACHE_MAX_ENTRIES = 128 +_SUMMARY_CACHE: OrderedDict[str, tuple[tuple[int, int, int, int, int], dict]] = OrderedDict() +_SUMMARY_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" @@ -71,6 +80,68 @@ def _lock_for(path: Path) -> threading.Lock: return lock +def _summary_cache_signature(path: Path) -> tuple[int, int, int, int, int] | None: + """Return the complete filesystem identity used for summary-cache validity. + + Includes ``st_ctime_ns`` so a same-inode, same-size rewrite that restores the + original ``mtime_ns`` (e.g. an atomic replace) still invalidates the cache — + ctime advances on any metadata/content change and cannot be forged back. + """ + try: + stat = path.stat() + except OSError: + return None + return ( + int(stat.st_dev), + int(stat.st_ino), + int(stat.st_size), + int(stat.st_mtime_ns), + int(stat.st_ctime_ns), + ) + + +def _get_cached_summary(path: Path) -> dict | None: + signature = _summary_cache_signature(path) + if signature is None: + return None + key = str(path) + with _SUMMARY_CACHE_LOCK: + cached = _SUMMARY_CACHE.get(key) + if cached is None: + return None + cached_signature, summary = cached + if cached_signature != signature: + _SUMMARY_CACHE.pop(key, None) + return None + _SUMMARY_CACHE.move_to_end(key) + return dict(summary) + + +def _cache_summary( + path: Path, + summary: dict, + *, + expected_signature: tuple[int, int, int, int, int] | None = None, +) -> None: + signature = _summary_cache_signature(path) + # The pre-read signature is an enforced TOCTOU precondition. In particular, + # a journal created after a missing-file read has ``None -> signature`` and + # must not cache the empty/unknown result under the new file's identity. + if signature is None or signature != expected_signature: + return + key = str(path) + with _SUMMARY_CACHE_LOCK: + _SUMMARY_CACHE[key] = (signature, dict(summary)) + _SUMMARY_CACHE.move_to_end(key) + while len(_SUMMARY_CACHE) > _SUMMARY_CACHE_MAX_ENTRIES: + _SUMMARY_CACHE.popitem(last=False) + + +def _discard_cached_summary(path: Path) -> None: + with _SUMMARY_CACHE_LOCK: + _SUMMARY_CACHE.pop(str(path), None) + + def _read_jsonl(path: Path) -> tuple[list[dict], list[dict]]: events: list[dict] = [] malformed: list[dict] = [] @@ -351,6 +422,7 @@ def append_run_event( fh.flush() if _should_fsync_event(terminal_state): os.fsync(fh.fileno()) + _discard_cached_summary(path) if created_file: _fsync_parent_dir(path) return event @@ -427,8 +499,15 @@ def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) - def latest_run_summary(session_id: str, run_id: str, *, session_dir: Path | None = None) -> dict: - journal = read_run_events(session_id, run_id, session_dir=session_dir) - return _summary_from_events(session_id, run_id, journal.get("events") or []) + path = _run_path(session_id, run_id, session_dir=session_dir) + cached = _get_cached_summary(path) + if cached is not None: + return cached + pre_read_signature = _summary_cache_signature(path) + events, _malformed = _read_jsonl(path) + summary = _summary_from_events(session_id, run_id, events) + _cache_summary(path, summary, expected_signature=pre_read_signature) + return summary def session_journal_fingerprint(session_id: str, *, session_dir: Path | None = None) -> tuple[int, float, int]: @@ -471,8 +550,12 @@ def find_run_summary(run_id: str, *, session_dir: Path | None = None) -> dict | journal_root = root / RUN_JOURNAL_DIR_NAME for path in journal_root.glob(f"*/{rid}.jsonl"): session_id = path.parent.name - events, _malformed = _read_jsonl(path) - summary = _summary_from_events(session_id, rid, events) + summary = _get_cached_summary(path) + if summary is None: + pre_read_signature = _summary_cache_signature(path) + events, _malformed = _read_jsonl(path) + summary = _summary_from_events(session_id, rid, events) + _cache_summary(path, summary, expected_signature=pre_read_signature) summary["path"] = str(path) return summary return None @@ -640,8 +723,11 @@ def delete_run_journal(session_id: str, *, session_dir: Path | None = None) -> b # ``_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] + for cache_key in [entry for entry in _SEQ_CACHE if str(Path(entry).parent) == dir_key]: + del _SEQ_CACHE[cache_key] + with _SUMMARY_CACHE_LOCK: + for cache_key in [entry for entry in _SUMMARY_CACHE if str(Path(entry).parent) == dir_key]: + del _SUMMARY_CACHE[cache_key] return removed diff --git a/tests/test_run_journal.py b/tests/test_run_journal.py index 6681e89f..cce7186f 100644 --- a/tests/test_run_journal.py +++ b/tests/test_run_journal.py @@ -1,4 +1,5 @@ import json +from pathlib import Path from api.run_journal import ( RunJournalWriter, @@ -104,6 +105,120 @@ def test_latest_summary_and_find_run_summary_classify_terminal_state(tmp_path): assert found["terminal_state"] == "interrupted-by-user" +def test_latest_summary_reuses_unchanged_journal_summary_without_reparsing(tmp_path, monkeypatch): + append_run_event("session_1", "run_1", "token", {"text": "ok"}, session_dir=tmp_path) + append_run_event("session_1", "run_1", "done", {"session": {}}, session_dir=tmp_path) + + first = latest_run_summary("session_1", "run_1", session_dir=tmp_path) + + monkeypatch.setattr( + "api.run_journal._read_jsonl", + lambda _path: (_ for _ in ()).throw(AssertionError("unchanged journal was reparsed")), + ) + repeated = latest_run_summary("session_1", "run_1", session_dir=tmp_path) + + assert repeated == first + + +def test_summary_cache_invalidates_on_same_size_rewrite_with_restored_mtime(tmp_path, monkeypatch): + # A same-inode, same-size rewrite that restores the original mtime_ns (e.g. an + # atomic replace, or a tool that preserves mtime) must still invalidate the + # cached summary. The signature includes st_ctime_ns — which advances on any + # content/metadata change and cannot be forged back — so device/inode/size/ + # mtime collisions alone can never serve a stale summary. Proven at the + # signature level (the enforced TOCTOU precondition for the cache) with a + # deterministic stat where ONLY ctime differs. + import api.run_journal as run_journal + + append_run_event("session_1", "run_1", "token", {"text": "ok"}, session_dir=tmp_path) + path = run_journal._run_path("session_1", "run_1", session_dir=tmp_path) + real = path.stat() + + class _Stat: + st_dev = real.st_dev + st_ino = real.st_ino + st_size = real.st_size + st_mtime_ns = real.st_mtime_ns + st_ctime_ns = real.st_ctime_ns # overwritten per-call below + + seq = {"ctime": real.st_ctime_ns} + + def fake_stat(self, *a, **k): + s = _Stat() + s.st_ctime_ns = seq["ctime"] + return s + + monkeypatch.setattr(Path, "stat", fake_stat) + sig_before = run_journal._summary_cache_signature(path) + # Same dev/inode/size/mtime, but a same-size in-place rewrite advanced ctime. + seq["ctime"] = real.st_ctime_ns + 1 + sig_after = run_journal._summary_cache_signature(path) + + assert sig_after is not None and sig_before is not None + assert sig_after != sig_before, "signature must change when only ctime advances" + assert sig_before[:4] == sig_after[:4], "dev/inode/size/mtime_ns unexpectedly changed" + + +def test_summary_cache_does_not_store_result_when_journal_changes_during_read(tmp_path, monkeypatch): + append_run_event("session_1", "run_1", "token", {"text": "ok"}, session_dir=tmp_path) + append_run_event("session_1", "run_1", "done", {"session": {}}, session_dir=tmp_path) + + import api.run_journal as run_journal + + original_read = run_journal._read_jsonl + + def append_after_read(path): + events, malformed = original_read(path) + append_run_event( + "session_1", + "run_1", + "cancel", + {"message": "Cancelled by user"}, + session_dir=tmp_path, + ) + return events, malformed + + monkeypatch.setattr(run_journal, "_read_jsonl", append_after_read) + + first = latest_run_summary("session_1", "run_1", session_dir=tmp_path) + second = latest_run_summary("session_1", "run_1", session_dir=tmp_path) + + assert first["terminal_state"] == "completed" + assert second["terminal_state"] == "interrupted-by-user" + + + +def test_summary_cache_rejects_first_append_that_races_missing_journal_read(tmp_path, monkeypatch): + import api.run_journal as run_journal + + original_read = run_journal._read_jsonl + appended = False + + def append_after_missing_read(path): + nonlocal appended + events, malformed = original_read(path) + if not appended: + appended = True + append_run_event( + "session_1", + "run_first_append", + "done", + {"session": {}}, + session_dir=tmp_path, + ) + return events, malformed + + monkeypatch.setattr(run_journal, "_read_jsonl", append_after_missing_read) + + raced = latest_run_summary("session_1", "run_first_append", session_dir=tmp_path) + refreshed = latest_run_summary("session_1", "run_first_append", session_dir=tmp_path) + + assert raced["terminal_state"] == "unknown" + assert refreshed["terminal_state"] == "completed" + assert refreshed["last_seq"] == 1 + assert refreshed["last_event_id"] == "run_first_append:1" + + def test_terminal_state_classification_distinguishes_crash_from_user_cancel(tmp_path): append_run_event("session_1", "run_cancelled", "cancel", {"message": "Cancelled by user"}, session_dir=tmp_path) append_run_event("session_1", "run_crashed", "apperror", {"type": "interrupted"}, session_dir=tmp_path)