Merge PR #7569 from happy5318: perf(run-journal): rebuild live snapshots without quadratic work

tests/test_7569_timed_flush_index_reset.py

@@ -0,0 +1,81 @@
+"""#7569 review fix: timed-flush must not leave the folded index stale.
+
+The ~10 Hz reasoning throttle in the stream callback clears the coalescing
+buffer before emitting:
+
+    put('reasoning', {'text': _reasoning_buffer[0]})
+    _reasoning_buffer[0] = ''
+    # ← _reasoning_buffer_index.reset() was missing here
+
+``_CompactEchoIndex`` records raw offsets *relative to the buffer's accumulated
+length*. Clearing the buffer without resetting the index leaves it describing
+text that is no longer there, so the next ``cut_to`` returns a raw offset that
+no longer corresponds to the buffer — the live-echo strip then truncates the
+buffer at a phantom position and falsely marks the interim as
+``reasoning_echo: true`` (leaks a truncated copy into both the SSE stream and
+the durable journal).
+
+This file pins the invariant at the *call site*, not just on the index class:
+revert the fix commit and the first test fails.
+"""
+
+import pathlib
+import re
+
+REPO = pathlib.Path(__file__).resolve().parent.parent
+STREAMING = (REPO / "api" / "streaming.py").read_text(encoding="utf-8")
+
+
+def _timed_flush_block() -> str:
+    """The ``if now - _reasoning_last_put[0] >= 0.1:`` throttle branch body."""
+    match = re.search(
+        r"if now - _reasoning_last_put\[0\] >= 0\.1:\n(.*?)(?=\n\s{16}(?:# Track reasoning deltas|_metering_reasoning_deltas))",
+        STREAMING,
+        re.DOTALL,
+    )
+    assert match, "timed-flush throttle branch not found in api/streaming.py"
+    return match.group(1)
+
+
+def test_timed_flush_clears_index_together_with_the_buffer():
+    """Clearing the reasoning buffer must also reset the folded index, in the
+    same throttle branch, after the clear (ordering matters: the reset cannot
+    sit in an unrelated branch or before the clear)."""
+    block = _timed_flush_block()
+
+    assert "_reasoning_buffer[0] = ''" in block, (
+        "timed-flush branch must clear the coalesced reasoning buffer"
+    )
+    assert "_reasoning_buffer_index.reset()" in block, (
+        "timed-flush branch must reset the folded index when it clears the "
+        "buffer — leaving the index stale makes later cut_to() offsets "
+        "describe text that is no longer in the buffer (#7569 re-gate)"
+    )
+    # reset must come AFTER the clear, and be at the same indent depth (8
+    # spaces inside the throttle branch), not in a sibling branch.
+    clear_pos = block.index("_reasoning_buffer[0] = ''")
+    reset_pos = block.index("_reasoning_buffer_index.reset()")
+    assert reset_pos > clear_pos, (
+        "reset must follow the buffer clear, so the next append starts from "
+        "an empty raw-length base"
+    )
+    assert re.search(r"^\s{20}_reasoning_buffer_index\.reset\(\)", block, re.M), (
+        "reset must be inside the timed-flush branch (deeper than the clear), "
+        "not in an outer/sibling scope"
+    )
+
+
+def test_all_buffer_clears_are_index_safe():
+    """Every site that zeroes ``_reasoning_buffer[0]`` must reset the folded
+    index immediately after — grep the codebase for clear sites and require
+    each to be paired with the index reset."""
+    for clear_match in re.finditer(r"_reasoning_buffer\[0\] = ''", STREAMING):
+        window = STREAMING[clear_match.start(): clear_match.start() + 400]
+        reset_present = re.search(
+            r"_reasoning_buffer_index\.reset\(\)", window
+        )
+        assert reset_present, (
+            "a buffer clear at api/streaming.py has no adjacent "
+            "_reasoning_buffer_index.reset() — stale offsets would corrupt "
+            "the next cut_to() call"
+        )

tests/test_compact_echo_index.py

@@ -0,0 +1,218 @@
+"""``_CompactEchoIndex`` must agree with the raw backward walk, cheaply.
+
+The index is the echo-match authority for both the journal rebuild
+(``api/routes.py::_run_journal_live_snapshot``) and the live stream
+(``api/streaming.py``). It replaces a per-interim raw backward walk that
+rescanned the whole transcript's whitespace — quadratic on whitespace-heavy
+text (#7569). Two contracts must hold for the swap to be safe:
+
+1. **Equivalence** — incremental appends must reach the same cut point as a
+   raw walk over the concatenated text, for every shape including chunked
+   appends, whitespace-only chunks, and exotic Unicode whitespace.
+2. **Cost** — the index must never rescan the buffer it has already folded:
+   a probe after a wide whitespace span must not walk that span.
+"""
+import random
+import re
+import string
+
+import pytest
+
+
+def _compact(value):
+    return re.sub(r'\s+', '', str(value or ''))
+
+
+def _raw_reference(value, suffix):
+    """Deliberately naive oracle: probe every cut index, fold the tail."""
+    raw = str(value or '')
+    candidate = _compact(suffix)
+    if not raw or not candidate:
+        return raw, False
+    for idx in range(len(raw) + 1):
+        if _compact(raw[idx:]) == candidate:
+            return raw[:idx].rstrip(), True
+    return raw, False
+
+
+def _cases():
+    cases = [
+        # Exact echo at the end.
+        ("bla bla bla une conclusion", "une conclusion"),
+        # Echo differing only by whitespace shape.
+        ("bla bla  une   conclusion", "une conclusion"),
+        ("bla bla\nune\tconclusion", "une conclusion"),
+        # No echo.
+        ("texte quelconque sans rapport", "une conclusion"),
+        # Partial echo must NOT match.
+        ("bla une conclu", "une conclusion"),
+        # The buffer is exactly the echo.
+        ("une conclusion", "une conclusion"),
+        # Empty / None inputs.
+        ("", "une conclusion"),
+        ("bla bla", ""),
+        ("", ""),
+        (None, "x"),
+        ("x", None),
+        # Whitespace around the cut point.
+        ("bla bla une conclusion   ", "une conclusion"),
+        ("bla bla   \n  une conclusion", "une conclusion"),
+        # Unicode, accents, emoji, exotic whitespace.
+        ("préambule ✅ conclusion émise", "conclusion émise"),
+        ("texte 🟢 Réponse", "🟢 Réponse"),
+        ("bla\u00a0bla une conclusion", "une\u00a0conclusion"),
+        ("bla\u2028une conclusion", "une conclusion"),
+        # Whitespace-only buffer.
+        ("     ", "une conclusion"),
+        # Repetitions: the leftmost cut point is contractual.
+        ("abc abc abc", "abc"),
+        ("abc abc abc", "abc abc"),
+        ("aaaa", "aa"),
+    ]
+    rnd = random.Random(20260920)
+    alphabet = string.ascii_letters + "  \n\t" + "éà✅\u00a0"
+    for _ in range(2000):
+        base = "".join(rnd.choice(alphabet) for _ in range(rnd.randint(0, 120)))
+        if rnd.random() < 0.5 and len(base) > 4:
+            k = rnd.randint(1, max(1, len(base) // 2))
+            suffix = base[-k:]
+        else:
+            suffix = "".join(rnd.choice(alphabet) for _ in range(rnd.randint(0, 30)))
+        cases.append((base, suffix))
+    return cases
+
+
+def test_index_matches_raw_reference_oracle():
+    """Incremental index == naive raw probe, for one-shot and chunked appends."""
+    import api.streaming as streaming
+
+    divergences = []
+    for value, suffix in _cases():
+        expected = _raw_reference(value, suffix)
+        index = streaming._CompactEchoIndex()
+        index.append(value)
+        actual = index.cut_to(suffix)
+        if actual is None:
+            got = (str(value or ''), False)
+        else:
+            got = (str(value or '')[:actual].rstrip(), True)
+        if got != expected:
+            divergences.append((value, suffix, expected, got))
+
+    assert not divergences[:5], (
+        f"{len(divergences)} divergence(s) from the reference implementation; "
+        f"first: {divergences[:1]}"
+    )
+
+
+def test_chunked_appends_reach_the_same_cut_point():
+    """Appending in pieces must equal appending the whole string."""
+    import api.streaming as streaming
+
+    rnd = random.Random(99)
+    for _ in range(500):
+        parts = [
+            "".join(
+                rnd.choice(string.ascii_letters + "  \n\té")
+                for _ in range(rnd.randint(1, 20))
+            )
+            for _ in range(rnd.randint(1, 8))
+        ]
+        value = "".join(parts)
+        if rnd.random() < 0.6 and len(value) > 4:
+            suffix = value[-rnd.randint(1, len(value) // 2):]
+        else:
+            suffix = "".join(rnd.choice("abc \t") for _ in range(rnd.randint(0, 12)))
+
+        chunked = streaming._CompactEchoIndex()
+        for part in parts:
+            chunked.append(part)
+        whole = streaming._CompactEchoIndex()
+        whole.append(value)
+
+        assert chunked.cut_to(suffix) == whole.cut_to(suffix), (
+            f"chunked append diverged for value={value!r} suffix={suffix!r}"
+        )
+
+
+def test_matches_tail_and_cut_to_agree():
+    """The boolean probe and the cut probe must never disagree."""
+    import api.streaming as streaming
+
+    for value, suffix in _cases():
+        index = streaming._CompactEchoIndex()
+        index.append(value)
+        assert index.matches_tail(suffix) == (index.cut_to(suffix) is not None), (
+            f"matches_tail/cut_to disagree for value={value!r} suffix={suffix!r}"
+        )
+
+
+def test_reset_clears_state_after_truncation():
+    """After reset + re-append the index must describe only the new text."""
+    import api.streaming as streaming
+
+    index = streaming._CompactEchoIndex()
+    index.append("old reasoning tail with a wide span " + " " * 5000)
+    index.reset()
+    index.append("fresh start only")
+
+    assert index.compact_length == len("freshstartonly")
+    assert index.cut_to("start only") == len("fresh ")
+    assert index.cut_to("old reasoning") is None
+
+
+def test_index_probe_does_not_rescan_a_wide_whitespace_span(monkeypatch):
+    """The probe must not walk raw characters it already folded.
+
+    Pins the cost property the raw backward walk violated: after a 1 MB
+    whitespace span is appended, a probe must cost O(candidate), not
+    O(span). Counting ``str.isspace``-adjacent raw access is impractical, so
+    this counts how many characters ``append`` has to fold: exactly the
+    appended length, once — and the probe itself must not fold anything.
+    """
+    import api.streaming as streaming
+
+    fold_calls = {'n': 0}
+    original = streaming._compact_for_echo_compare
+
+    def counting(value):
+        fold_calls['n'] += 1
+        return original(value)
+
+    monkeypatch.setattr(streaming, '_compact_for_echo_compare', counting)
+
+    index = streaming._CompactEchoIndex()
+    index.append("alpha" + " " * 100_000)
+    fold_calls['n'] = 0
+
+    # 20 probes of a short candidate: the raw walk would fold/walk the span
+    # each time; the index folds only the candidate.
+    for i in range(20):
+        assert index.matches_tail(f"progress {i}") is False
+
+    assert fold_calls['n'] <= 20, (
+        f"echo probe folded {fold_calls['n']} times — it is rescanning the "
+        "raw transcript instead of using the indexed tail"
+    )
+
+
+@pytest.mark.timeout(60)
+def test_index_stays_cheap_on_a_wide_span_with_many_probes():
+    """Wide whitespace span + many probes must not cost seconds of CPU."""
+    import time
+
+    import api.streaming as streaming
+
+    index = streaming._CompactEchoIndex()
+    index.append("alpha" + " " * 1_000_000)
+    probes = [f"progress {i}" for i in range(200)]
+
+    start = time.perf_counter()
+    for probe in probes:
+        index.matches_tail(probe)
+        index.cut_to(probe)
+    elapsed = time.perf_counter() - start
+
+    # The raw backward walk needs ~0.05s per probe here (2M+ character walks
+    # total). A generous budget still fails it by an order of magnitude.
+    assert elapsed < 0.5, f"200 probes on a 1MB span took {elapsed:.3f}s"

tests/test_compact_echo_suffix_linear_scan.py

@@ -7,11 +7,16 @@ with ``window x suffix`` instead of with the data actually inspected. On a
 6000-character conclusion block that is seconds of pure CPU, held under the
 GIL, which stalls every other stream in the process.
 
+The current implementation walks ``value`` and ``suffix`` backwards from the
+end (skipping whitespace in ``value``) and has no fixed search window, so a
+compact-equivalent echo is found regardless of how much interior whitespace
+stretches its raw span.
+
 These tests pin both halves of the contract:
 
 * the fast path must return byte-identical results to a straightforward
   reference implementation, including the leftmost-cut tie-break; and
-* it must not re-fold the search window once per candidate cut index.
+* it must not fold the whole buffer once per candidate cut index.
 """
 import random
 import re
@@ -20,11 +25,12 @@ import string
 import pytest
 
 
-def _reference_strip_compact_echo_suffix(value, suffix, *, search_window: int = 4096):
-    """Deliberately naive oracle: probe every cut index, fold the whole tail.
+def _reference_strip_compact_echo_suffix(value, suffix):
+    """Deliberately naive oracle: probe every cut index, fold the tail.
 
-    This mirrors the original behaviour exactly and exists only so the
-    optimised implementation can be proven equivalent to it.
+    This mirrors the pre-window semantics (the original probing loop) with no
+    search window, and exists only so the optimised backward walk can be
+    proven equivalent to it.
     """
     def compact(v):
         return re.sub(r'\s+', '', str(v or ''))
@@ -33,11 +39,9 @@ def _reference_strip_compact_echo_suffix(value, suffix, *, search_window: int =
     candidate = compact(suffix)
     if not raw or not candidate:
         return raw, False
-    tail = raw[-max(len(str(suffix or '')) * 3, search_window):]
-    offset = len(raw) - len(tail)
-    for idx in range(len(tail) + 1):
-        if compact(tail[idx:]) == candidate:
-            return raw[: offset + idx].rstrip(), True
+    for idx in range(len(raw) + 1):
+        if compact(raw[idx:]) == candidate:
+            return raw[:idx].rstrip(), True
     return raw, False
 
 
@@ -74,8 +78,8 @@ def _cases():
         ("aaaa", "aa"),
         # Long buffer with the echo at the very end.
         ("remplissage " * 500 + "la vraie conclusion", "la vraie conclusion"),
-        # Echo whose whitespace-padded span is wider than the search window:
-        # the window only exposes ``conclusion``, so nothing may be stripped.
+        # Echo whose whitespace-padded span is wider than any fixed fold
+        # window: the backward walk must still strip it.
         ("bla une" + " " * 5000 + "conclusion", "une conclusion"),
         # Exotic whitespace that ``\s`` and ``str.isspace`` must treat alike.
         ("bla\u00a0bla une conclusion", "une\u00a0conclusion"),
@@ -111,45 +115,29 @@ def test_strip_compact_echo_suffix_matches_reference_oracle():
     )
 
 
-def test_strip_compact_echo_suffix_honours_a_custom_search_window():
-    """The bounded window must keep its meaning on the fast path."""
-    import api.streaming as streaming
-
-    buffer_text = "z" * 400 + " la conclusion"
-    for window in (16, 64, 512, 4096):
-        assert streaming._strip_compact_echo_suffix(
-            buffer_text, "la conclusion", search_window=window
-        ) == _reference_strip_compact_echo_suffix(
-            buffer_text, "la conclusion", search_window=window
-        ), f"window={window}"
-
-
-def test_strip_compact_echo_suffix_rejects_an_echo_wider_than_the_window():
-    """An echo that only fits inside a wider window must not be found.
+def test_strip_compact_echo_suffix_finds_an_echo_wider_than_any_fixed_window():
+    """An echo stretched past any fold window must still be stripped.
 
     The buffer ends with the echo, but interior whitespace stretches its raw
-    span past the default window, so the folded window only exposes the last
-    word. Widening the window to cover the whole span makes the same echo
-    visible again, which proves the non-match is caused by the window and
-    not by the text.
+    span far beyond a fixed fold window (the retired windowed implementation
+    exposed only the last word and left the buffer untouched). The backward
+    walk has no window, so the cut point is found and the echo is removed.
     """
     import api.streaming as streaming
 
     buffer_text = "bla une" + " " * 5000 + "conclusion"
     suffix = "une conclusion"
 
-    for impl in (streaming._strip_compact_echo_suffix, _reference_strip_compact_echo_suffix):
-        assert impl(buffer_text, suffix) == (buffer_text, False), impl.__name__
-        assert impl(buffer_text, suffix, search_window=len(buffer_text)) == ("bla", True), impl.__name__
+    assert streaming._strip_compact_echo_suffix(buffer_text, suffix) == ("bla", True)
+    assert _reference_strip_compact_echo_suffix(buffer_text, suffix) == ("bla", True)
 
 
-def test_strip_compact_echo_suffix_does_not_refold_the_window_per_cut(monkeypatch):
+def test_strip_compact_echo_suffix_does_not_refold_per_cut(monkeypatch):
     """One pass, not one fold per candidate cut index.
 
-    The original implementation called the whitespace-folding helper once per
-    possible cut position — roughly 4097 times for a default window — and each
-    of those calls re-scanned the remaining tail. Counting the calls pins the
-    algorithmic property directly, without depending on wall-clock timing.
+    Counting the folding-helper calls pins the algorithmic property directly,
+    without depending on wall-clock timing: the backward walk folds the
+    suffix once and then compares characters.
     """
     import api.streaming as streaming
 
@@ -168,7 +156,7 @@ def test_strip_compact_echo_suffix_does_not_refold_the_window_per_cut(monkeypatc
 
     assert calls['n'] <= 8, (
         f"whitespace folding ran {calls['n']} times for a single call; the scan "
-        "is re-folding the window once per candidate cut index"
+        "is re-folding once per candidate cut index"
     )
 
 
@@ -193,4 +181,4 @@ def test_strip_compact_echo_suffix_stays_cheap_on_a_long_conclusion():
     # The original implementation needs ~7s here. A very loose budget keeps the
     # test meaningful on a loaded CI box while still failing the quadratic scan
     # by three orders of magnitude.
-    assert elapsed < 0.5, f"single call took {elapsed:.3f}s"
+    assert elapsed < 0.5, f"single call took {elapsed:.3f}s"

tests/test_live_snapshot_echo_dedup.py

@@ -0,0 +1,104 @@
+"""Journal-reconstruction regression: a whitespace-stretched reasoning echo
+must be de-duplicated exactly once.
+
+``_run_journal_live_snapshot`` strips reasoning echoes of interim assistant
+text so the restored transcript does not show the same content in both the
+thinking block and the assistant output (run-state consistency invariant:
+replay must not duplicate interim assistant text). The previous bounded echo
+probe could miss a compact-equivalent suffix whose raw span exceeded its fold
+window, leaving the duplicate in the snapshot; the backward walk has no
+window.
+
+Both directions are pinned: a stretched echo is stripped (appears once), and
+genuine interim text that is NOT a reasoning echo is left alone.
+"""
+
+import json
+import re
+
+import api.routes as routes
+from api import run_journal as RJ
+
+
+def _compact(value):
+    return re.sub(r"\s+", "", str(value or ""))
+
+
+def _write_journal(root, session_id, run_id, events):
+    session_root = root / RJ.RUN_JOURNAL_DIR_NAME / session_id
+    session_root.mkdir(parents=True, exist_ok=True)
+    path = session_root / f"{run_id}.jsonl"
+    rows = []
+    for seq, (event, payload) in enumerate(events, start=1):
+        rows.append(
+            json.dumps(
+                {
+                    "version": 1,
+                    "event_id": f"{run_id}:{seq}",
+                    "seq": seq,
+                    "run_id": run_id,
+                    "session_id": session_id,
+                    "event": event,
+                    "type": event,
+                    "created_at": 1000 + seq,
+                    "payload": payload,
+                }
+            )
+        )
+    path.write_text("\n".join(rows) + "\n", encoding="utf-8")
+    return path
+
+
+def test_whitespace_stretched_reasoning_echo_appears_once(tmp_path, monkeypatch):
+    gap = " " * 5000  # raw echo span far beyond any fixed fold window
+    _write_journal(
+        tmp_path,
+        "session_1",
+        "run_1",
+        [
+            ("reasoning", {"text": "prefix " + "alpha" + gap + "beta"}),
+            ("interim_assistant", {"text": "alpha beta"}),
+        ],
+    )
+    monkeypatch.setattr(RJ, "_default_session_dir", lambda: tmp_path)
+
+    snapshot = routes._run_journal_live_snapshot("run_1")
+
+    message = snapshot["messages"][-1]
+    reasoning_text = message.get("reasoning") or ""
+    content_text = message.get("content") or ""
+    occurrences = _compact(reasoning_text).count("alphabeta") + _compact(
+        content_text
+    ).count("alphabeta")
+    assert occurrences == 1, (
+        f"the interim echo must appear once after reconstruction; "
+        f"reasoning={reasoning_text[-60:]!r} content={content_text[-60:]!r}"
+    )
+    # The echo span is stripped from the reasoning copy; the prefix stays.
+    assert "alphabeta" not in _compact(reasoning_text)
+    assert "alpha beta" in content_text
+
+
+def test_genuine_interim_text_is_not_treated_as_echo(tmp_path, monkeypatch):
+    gap = " " * 5000
+    _write_journal(
+        tmp_path,
+        "session_1",
+        "run_1",
+        [
+            ("reasoning", {"text": "prefix " + "alpha" + gap + "beta"}),
+            ("interim_assistant", {"text": "gamma delta"}),
+        ],
+    )
+    monkeypatch.setattr(RJ, "_default_session_dir", lambda: tmp_path)
+
+    snapshot = routes._run_journal_live_snapshot("run_1")
+
+    message = snapshot["messages"][-1]
+    reasoning_text = message.get("reasoning") or ""
+    content_text = message.get("content") or ""
+    # The genuine interim answer is preserved in full...
+    assert "gamma delta" in content_text
+    # ...and the reasoning transcript is not suppressed by the matcher.
+    assert _compact(reasoning_text).count("alphabeta") == 1
+    assert "prefix" in reasoning_text

tests/test_live_snapshot_echo_scan_scaling.py

@@ -0,0 +1,127 @@
+"""Scaling regression for the journal-rebuild echo path (#7569).
+
+A fixed-size fixture cannot catch a per-interim quadratic rescan: the shape
+that exposes it is a whitespace span that grows in proportion to the interim
+event count, so a quadratic implementation's cost grows with the square of
+the row count while a linear one stays flat per row.
+
+The retired implementations both rescan the raw reasoning transcript per
+interim event (the folded-tail probe, then the window-free backward walk):
+with a 1 MB whitespace span and 20 interim events that is 20M raw-character
+walks. This test pins the scaling property directly, without depending on
+wall-clock noise, by counting the raw characters the matcher touches.
+"""
+import json
+import time
+
+import pytest
+
+
+def _write_journal(root, session_id, run_id, events):
+    from api import run_journal as RJ
+
+    session_root = root / RJ.RUN_JOURNAL_DIR_NAME / session_id
+    session_root.mkdir(parents=True, exist_ok=True)
+    path = session_root / f"{run_id}.jsonl"
+    rows = []
+    for seq, (event, payload) in enumerate(events, start=1):
+        rows.append(
+            json.dumps(
+                {
+                    "version": 1,
+                    "event_id": f"{run_id}:{seq}",
+                    "seq": seq,
+                    "run_id": run_id,
+                    "session_id": session_id,
+                    "event": event,
+                    "type": event,
+                    "created_at": 1000 + seq,
+                    "payload": payload,
+                }
+            )
+        )
+    path.write_text("\n".join(rows) + "\n", encoding="utf-8")
+    return path
+
+
+def _whitespace_heavy_journal(interim_count: int, span: int):
+    """Reasoning ends in a wide whitespace span; every interim probes it.
+
+    ``interim_count`` interim events each run the echo matcher against the
+    accumulated transcript. Doubling both the span and the interim count
+    quadruples the raw characters a per-interim raw walk would touch, but
+    only doubles them for an incremental index.
+    """
+    events = [("reasoning", {"text": "alpha" + " " * span})]
+    events += [
+        ("interim_assistant", {"text": f"progress {i}", "reasoning_echo": False})
+        for i in range(interim_count)
+    ]
+    events.append(("token", {"text": "final answer"}))
+    return events
+
+
+def _rebuild(root, run_id):
+    import api.routes as routes
+    from api import run_journal as RJ
+
+    original = RJ._default_session_dir
+    RJ._default_session_dir = lambda: root
+    try:
+        return routes._run_journal_live_snapshot(run_id)
+    finally:
+        RJ._default_session_dir = original
+
+
+@pytest.mark.parametrize("scale", [1, 2, 4])
+def test_rebuild_echo_scan_is_not_per_interim_quadratic(tmp_path, scale):
+    """Doubling span AND interim count must stay far under quadratic cost.
+
+    The budget is a per-row constant: a quadratic implementation spends
+    ``span * interim_count`` raw-character walks, a linear one spends
+    ``span`` (folded once) plus ``O(interim_count * echo_len)``. The
+    assertion below fails the quadratic shape by roughly an order of
+    magnitude while leaving the linear one with generous headroom.
+    """
+    interim_count = 20 * scale
+    span = 100_000 * scale
+    _write_journal(
+        tmp_path, "sess", f"run_{scale}", _whitespace_heavy_journal(interim_count, span)
+    )
+
+    start = time.perf_counter()
+    snapshot = _rebuild(tmp_path, f"run_{scale}")
+    elapsed = time.perf_counter() - start
+
+    # Quadratic reference: 20*1 interims x 100k span took ~0.05s on the
+    # window-free walk, and ~1s at 1MB/20. Linear cost at scale=4 is a
+    # handful of single-span folds (400k chars) plus 80 short probes.
+    assert elapsed < 0.5, (
+        f"rebuild of a {span // 1000}KB whitespace span with {interim_count} "
+        f"interim events took {elapsed:.3f}s — the echo matcher is rescanning "
+        "the raw transcript per interim event (quadratic)"
+    )
+    # The snapshot must still project the final answer (behaviour preserved).
+    messages = (snapshot or {}).get("messages") or []
+    assert any("final answer" in (m.get("content") or "") for m in messages)
+
+
+def test_rebuild_scaling_is_subquadratic(tmp_path):
+    """Ratio check: 4x the rows must cost far less than 16x the time."""
+    times = {}
+    for scale in (1, 2, 4):
+        interim_count = 20 * scale
+        span = 100_000 * scale
+        _write_journal(
+            tmp_path, "sess", f"run_{scale}", _whitespace_heavy_journal(interim_count, span)
+        )
+        start = time.perf_counter()
+        _rebuild(tmp_path, f"run_{scale}")
+        times[scale] = time.perf_counter() - start
+
+    ratio = times[4] / times[1]
+    assert ratio < 8.0, (
+        f"4x the interim events and span cost {ratio:.1f}x the time "
+        f"({times}); a linear implementation stays near 4x, a quadratic one "
+        "reaches 16x"
+    )

tests/test_live_snapshot_single_parse.py

@@ -0,0 +1,114 @@
+"""Regression: live-snapshot rebuild locates the journal with a single parse.
+
+``_run_journal_live_snapshot`` used ``find_run_summary`` (full parse) and
+then ``read_run_events`` (second full parse) - two passes over the whole
+journal per rebuild (~0.6s+ of a 4.4s rebuild on a 22.7k-row live run).
+``find_run_file`` now locates the file without parsing and the durable
+summary is derived from the single events parse; the historical
+``find_run_summary`` lookup remains as the fallback seam when no journal
+file exists on disk.
+"""
+
+import json
+
+import api.routes as routes
+from api import run_journal as RJ
+
+
+def _write_journal(root, session_id, run_id, count=5):
+    session_root = root / RJ.RUN_JOURNAL_DIR_NAME / session_id
+    session_root.mkdir(parents=True, exist_ok=True)
+    path = session_root / f"{run_id}.jsonl"
+    rows = []
+    for seq in range(1, count + 1):
+        rows.append(
+            json.dumps(
+                {
+                    "version": 1,
+                    "event_id": f"{run_id}:{seq}",
+                    "seq": seq,
+                    "run_id": run_id,
+                    "session_id": session_id,
+                    "event": "token",
+                    "type": "token",
+                    "created_at": 1000 + seq,
+                    "payload": {"text": "x"},
+                }
+            )
+        )
+    path.write_text("\n".join(rows) + "\n", encoding="utf-8")
+    return path
+
+
+def test_find_run_file_locates_without_parsing(tmp_path):
+    path = _write_journal(tmp_path, "session_1", "run_1")
+    located = RJ.find_run_file("run_1", session_dir=tmp_path)
+    assert located is not None
+    session_id, located_path = located
+    assert session_id == "session_1"
+    assert located_path == path
+    assert RJ.find_run_file("missing_run", session_dir=tmp_path) is None
+    assert RJ.find_run_file("../evil", session_dir=tmp_path) is None
+
+
+def _forbid_summary_parse(*_a, **_k):
+    raise AssertionError(
+        "find_run_summary must not be parsed when the journal file exists on "
+        "disk - the single-parse locate path regressed"
+    )
+
+
+def test_live_snapshot_rebuilds_from_single_parse(tmp_path, monkeypatch):
+    _write_journal(tmp_path, "session_1", "run_1", count=3)
+    monkeypatch.setattr(RJ, "_default_session_dir", lambda: tmp_path)
+    # The double-parse fix: when a journal file exists, the rebuild must NOT
+    # route through find_run_summary (that would parse the file a second time).
+    monkeypatch.setattr(routes, "find_run_summary", _forbid_summary_parse)
+
+    snapshot = routes._run_journal_live_snapshot("run_1")
+
+    assert snapshot is not None
+    assert snapshot["session_id"] == "session_1"
+    assert snapshot["stream_id"] == "run_1"
+    assert snapshot["last_seq"] == 3
+    assert snapshot["event_count"] == 3
+    assert snapshot["last_assistant_text"] == "xxx"
+
+
+def test_live_snapshot_summary_fallback_seam(tmp_path, monkeypatch):
+    """No journal file on disk -> the historical find_run_summary seam is used."""
+    monkeypatch.setattr(RJ, "_default_session_dir", lambda: tmp_path)
+    monkeypatch.setattr(
+        routes,
+        "find_run_summary",
+        lambda stream_id: {
+            "session_id": "session_1",
+            "run_id": stream_id,
+            "last_seq": 1,
+            "last_event_id": f"{stream_id}:1",
+        },
+    )
+    monkeypatch.setattr(
+        routes,
+        "read_run_events",
+        lambda sid, rid: {
+            "events": [
+                {
+                    "version": 1,
+                    "event_id": f"{rid}:1",
+                    "seq": 1,
+                    "run_id": rid,
+                    "session_id": sid,
+                    "event": "token",
+                    "type": "token",
+                    "created_at": 1000,
+                    "payload": {"text": "hello"},
+                }
+            ],
+        },
+    )
+
+    snapshot = routes._run_journal_live_snapshot("run_1")
+
+    assert snapshot is not None
+    assert snapshot["last_assistant_text"] == "hello"