hermes-webui/tests/test_run_journal_seq_cache.py

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