From 3369bc6a6d9260300fec3c99195830afc4131c22 Mon Sep 17 00:00:00 2001 From: happy5318 Date: Tue, 15 Sep 2026 08:00:51 +0800 Subject: [PATCH] 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 | 19 ++-- api/streaming.py | 73 +++++++----- tests/test_compact_echo_suffix_linear_scan.py | 70 +++++------- tests/test_live_snapshot_echo_dedup.py | 104 ++++++++++++++++++ 4 files changed, 186 insertions(+), 80 deletions(-) create mode 100644 tests/test_live_snapshot_echo_dedup.py diff --git a/api/routes.py b/api/routes.py index cf1c86d7..de75c1fd 100644 --- a/api/routes.py +++ b/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 diff --git a/api/streaming.py b/api/streaming.py index 29cda8be..4c86104e 100644 --- a/api/streaming.py +++ b/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: diff --git a/tests/test_compact_echo_suffix_linear_scan.py b/tests/test_compact_echo_suffix_linear_scan.py index 21c66aec..1a0b0c00 100644 --- a/tests/test_compact_echo_suffix_linear_scan.py +++ b/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" \ No newline at end of file diff --git a/tests/test_live_snapshot_echo_dedup.py b/tests/test_live_snapshot_echo_dedup.py new file mode 100644 index 00000000..6bd55cb3 --- /dev/null +++ b/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