fix(run-journal): match echo suffixes with a window-free backward walk

Review follow-up: the bounded echo probe folded at most len(text)*3 raw characters, so a compact-equivalent suffix whose raw span was stretched by interior whitespace was missed - the echo strip was dropped and the interim text was duplicated in the restored transcript. The same retired window existed in the reverse helper (_strip_compact_echo_suffix), pinned by its own test.

Both now share _find_compact_echo_suffix_start: a backward walk over value/suffix that skips whitespace in value, inspects only the echo span, and allocates nothing. It restores the previous output semantics (origin/master folded the whole reasoning text) and is cheaper than the windowed version (no per-call tail fold).

Tests: new journal-reconstruction regression (a 5000-space-stretched echo must appear exactly once; genuine interim text must not be treated as an echo), the suffix-helper oracle updated to window-free semantics, and the window-contract test now pins that a stretched echo IS stripped.

api/routes.py

@@ -3592,20 +3592,16 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict |
         tool_calls.append(call)
 
     def reasoning_echo_tail_matches(text: str) -> bool:
-        candidate = _compact_for_echo_compare(text)
-        if not candidate:
-            return False
         full = _materialize_reasoning_text()
         if not full:
             return False
-        # Suffix test — folding the WHOLE (growing) reasoning transcript per
-        # interim segment ran re.sub over hundreds of KB 100+ times per
-        # rebuild (~1.5s). Fold only a bounded tail: compact() only drops
-        # whitespace, so a window of len(text)*3 characters covers the echo's
-        # non-whitespace characters — the same window contract
-        # _strip_compact_echo_suffix already uses for the reverse operation.
-        tail = full[-max(len(str(text or "")) * 3, 4096):]
-        return _compact_for_echo_compare(tail).endswith(candidate)
+        # Backward compact-suffix walk (no fixed window): linear in the echo
+        # length, allocation-free, and it cannot miss a compact-equivalent
+        # suffix whose raw span is stretched by interior whitespace. The
+        # previous bounded-tail probe folded up to len(text)*3 raw characters
+        # per interim segment - and when its window fell short it dropped the
+        # echo strip, duplicating the interim text in the restored transcript.
+        return _find_compact_echo_suffix_start(full, text) is not None
 
     def strip_reasoning_echo_tail(text: str) -> bool:
         nonlocal reasoning_text, reasoning_dirty, reasoning_first_tool_count
@@ -10759,6 +10755,7 @@ from api.streaming import (
     _materialize_pending_user_turn_before_error,
     generate_session_title_for_session,
     _compact_for_echo_compare,
+    _find_compact_echo_suffix_start,
     _strip_compact_echo_suffix,
 )
 from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_enabled

api/streaming.py

@@ -197,41 +197,58 @@ def _compact_for_echo_compare(value: str) -> str:
     return re.sub(r'\s+', '', str(value or ''))
 
 
-def _strip_compact_echo_suffix(value: str, suffix: str, *, search_window: int = 4096) -> tuple[str, bool]:
+def _find_compact_echo_suffix_start(value: str, suffix: str) -> int | None:
+    """Return the index where a whitespace-folded ``suffix`` starts at the
+    end of ``value``, or ``None`` when the tail does not echo it.
+
+    The match walks ``value`` and ``suffix`` from the end, skipping
+    whitespace in ``value``; only the echo span itself is inspected, so the
+    cost is linear in the echo length and allocation-free. Unlike a fixed
+    fold window, the walk cannot miss a compact-equivalent suffix whose raw
+    span is stretched by interior whitespace.
+
+    ``str.isspace`` is used for the walk instead of the ``\\s`` pattern used
+    by :func:`_compact_for_echo_compare`. The two agree on every Unicode code
+    point, so the folded view and the walk stay consistent.
+    """
+    candidate = _compact_for_echo_compare(suffix)
+    if not candidate:
+        return None
+    i = len(value) - 1
+    j = len(candidate) - 1
+    while j >= 0:
+        while i >= 0 and value[i].isspace():
+            i -= 1
+        if i < 0 or value[i] != candidate[j]:
+            return None
+        i -= 1
+        j -= 1
+    return i + 1
+
+
+def _strip_compact_echo_suffix(value: str, suffix: str) -> tuple[str, bool]:
     """Remove ``suffix`` from ``value`` when they match after whitespace folding.
 
-    The search window is folded once and the cut point is then located by
-    walking backwards across the echo itself. The previous implementation
-    probed every candidate cut index and re-folded the whole remaining tail for
-    each probe, which is quadratic in the window size: a 6000-character final
-    message cost seconds of CPU, held under the GIL, stalling every other
-    stream in the process.
+    The cut point is located by the same backward walk as
+    :func:`_find_compact_echo_suffix_start`: no fixed search window is
+    involved, so a compact-equivalent suffix is removed no matter how much
+    interior whitespace stretches its raw span. The previous windowed
+    implementation folded a bounded tail on every call; its retired probing
+    variant re-folded the remaining tail per candidate cut index, which is
+    quadratic: a 6000-character final message cost seconds of CPU, held under
+    the GIL, stalling every other stream in the process.
 
-    ``str.isspace`` is used for the backwards walk instead of the ``\\s``
-    pattern used by :func:`_compact_for_echo_compare`. The two agree on every
-    Unicode code point, so the folded view and the walk stay consistent.
+    Whitespace sitting between the kept text and the echo is removed by
+    ``rstrip``, which lands on the same result as the leftmost cut index the
+    probing loop used to return.
     """
     raw = str(value or '')
-    candidate = _compact_for_echo_compare(suffix)
-    if not raw or not candidate:
+    if not raw:
         return raw, False
-    tail = raw[-max(len(str(suffix or '')) * 3, search_window):]
-    offset = len(raw) - len(tail)
-    compact_tail = _compact_for_echo_compare(tail)
-    if len(candidate) > len(compact_tail) or not compact_tail.endswith(candidate):
+    start = _find_compact_echo_suffix_start(raw, suffix)
+    if start is None:
         return raw, False
-    # Consume exactly as many non-whitespace characters as the folded suffix
-    # holds; ``idx`` then sits on the first character of the echo. Whitespace
-    # sitting between the kept text and the echo is removed by ``rstrip``,
-    # which is why this lands on the same result as the leftmost cut index the
-    # probing loop used to return.
-    remaining = len(candidate)
-    idx = len(tail)
-    while remaining and idx:
-        idx -= 1
-        if not tail[idx].isspace():
-            remaining -= 1
-    return raw[: offset + idx].rstrip(), True
+    return raw[:start].rstrip(), True
 
 
 def _redacted_session_payload_with_full_messages(session, *, tool_calls=None) -> dict | None:

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