Merge remote-tracking branch 'origin/master' into paperclip/pr-6222

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)

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
 
 

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)