diff --git a/api/routes.py b/api/routes.py index de75c1fd..db182efb 100644 --- a/api/routes.py +++ b/api/routes.py @@ -3513,6 +3513,10 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | # most once per interim segment. reasoning_parts: list[str] = [] reasoning_dirty = False + # Incremental folded index over the reasoning transcript: the per-interim + # echo probe consults this instead of re-walking the raw text (see + # ``_CompactEchoIndex``). + reasoning_index = _CompactEchoIndex() messages: list[dict] = [] tool_calls: list[dict] = [] activity_burst_anchors: list[dict] = [] @@ -3592,30 +3596,32 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | tool_calls.append(call) def reasoning_echo_tail_matches(text: str) -> bool: - full = _materialize_reasoning_text() - if not full: - return False - # 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 + # Indexed tail match (no raw-text walk): the incremental index folds + # each reasoning chunk once as it is appended, so this probe costs + # O(len(text)) regardless of how much interior whitespace stretches + # the raw span. The previous raw backward walk re-walked that span + # once per interim event, which turned the replay quadratic on a + # whitespace-heavy transcript (#7569 review, ~55x slower than master + # on a production-shaped journal). + return reasoning_index.matches_tail(text) def strip_reasoning_echo_tail(text: str) -> bool: nonlocal reasoning_text, reasoning_dirty, reasoning_first_tool_count - next_reasoning, did_remove = _strip_compact_echo_suffix( - _materialize_reasoning_text(), text - ) - if did_remove: - reasoning_text = next_reasoning - reasoning_parts.clear() - reasoning_parts.append(next_reasoning) - reasoning_dirty = False - if not _compact_for_echo_compare(reasoning_text): - reasoning_first_tool_count = None - return did_remove + cut = reasoning_index.cut_to(text) + if cut is None: + return False + next_reasoning = _materialize_reasoning_text()[:cut].rstrip() + reasoning_text = next_reasoning + reasoning_parts.clear() + reasoning_parts.append(next_reasoning) + reasoning_dirty = False + # Re-index the truncated transcript so later probes match against it + # instead of the pre-strip tail (the index is the match authority). + reasoning_index.reset() + reasoning_index.append(next_reasoning) + if not _compact_for_echo_compare(reasoning_text): + reasoning_first_tool_count = None + return True for event in events: event_name = str(event.get("event") or event.get("type") or "") @@ -3640,6 +3646,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict | if reasoning_first_tool_count is None: reasoning_first_tool_count = len(tool_calls) reasoning_parts.append(text) + reasoning_index.append(text) reasoning_dirty = True continue if event_name == "interim_assistant": @@ -10755,6 +10762,7 @@ from api.streaming import ( _materialize_pending_user_turn_before_error, generate_session_title_for_session, _compact_for_echo_compare, + _CompactEchoIndex, _find_compact_echo_suffix_start, _strip_compact_echo_suffix, ) diff --git a/api/streaming.py b/api/streaming.py index 4c86104e..b55cfb9d 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -197,6 +197,104 @@ def _compact_for_echo_compare(value: str) -> str: return re.sub(r'\s+', '', str(value or '')) +class _CompactEchoIndex: + """Incremental whitespace-folded index over a growing text buffer. + + The journal-rebuild echo check runs once per interim event against the + whole accumulated reasoning transcript. Rescanning the raw buffer each + time walks every whitespace character between the tail and the first + non-whitespace character — on a whitespace-heavy transcript that span is + megabytes, and the per-interim walk turns the replay quadratic (the + ``_find_compact_echo_suffix_start`` shape measured ~55x slower than + master on a production-shaped journal in the #7569 review). + + This class keeps the folded view and the raw cut offsets *incrementally*: + + * ``append`` folds each new chunk once and records, per folded character, + its raw index — so the cost of a chunk is proportional to the chunk, + never to the buffer. + * ``matches_tail`` compares only the candidate against the end of the + folded view: O(len(candidate)), with no raw-text walk at all. + * ``cut_to`` converts a folded length back to a raw index by bisecting + the recorded offsets, so one call returns the same cut point the raw + backward walk would have produced. + + ``str.isspace`` is used for folding instead of the ``\\s`` pattern used + by :func:`_compact_for_echo_compare`. The two agree on every Unicode code + point, so the indexed view and the regex-folded view stay consistent. + """ + + __slots__ = ('_compact', '_offsets', '_raw_len') + + def __init__(self) -> None: + self._compact: list[str] = [] + self._offsets: list[int] = [] + self._raw_len = 0 + + def append(self, text: str) -> None: + """Fold ``text`` onto the end of the index (one pass, no rescan).""" + raw = str(text or '') + if not raw: + return + compact = self._compact + offsets = self._offsets + base = self._raw_len + for i, ch in enumerate(raw): + if not ch.isspace(): + compact.append(ch) + offsets.append(base + i) + self._raw_len = base + len(raw) + + def reset(self) -> None: + """Drop all indexed state (used after the raw text is truncated).""" + self._compact.clear() + self._offsets.clear() + self._raw_len = 0 + + @property + def compact_length(self) -> int: + return len(self._compact) + + def matches_tail(self, suffix: str) -> bool: + """True when the folded ``suffix`` equals the folded tail.""" + candidate = _compact_for_echo_compare(suffix) + if not candidate: + return False + n = len(candidate) + if n > len(self._compact): + return False + start = len(self._compact) - n + compact = self._compact + for i in range(n): + if compact[start + i] != candidate[i]: + return False + return True + + def cut_to(self, suffix: str) -> int | None: + """Raw index at which the echo of ``suffix`` starts, else ``None``. + + The returned index is the first raw character of the echo, matching + the leftmost-cut semantics of the raw backward walk. Trailing + whitespace before the echo is left to the caller's ``rstrip``. + """ + candidate = _compact_for_echo_compare(suffix) + if not candidate: + return None + n = len(candidate) + if n > len(self._compact): + return None + start = len(self._compact) - n + compact = self._compact + for i in range(n): + if compact[start + i] != candidate[i]: + return None + return self._offsets[start] + + def compact_view(self) -> str: + """The folded text (materialized on demand — never kept as a string).""" + return ''.join(self._compact) + + 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. @@ -10023,6 +10121,14 @@ def _run_agent_streaming( # (#3587) replaces the flat _reasoning_text string so each intermediate # assistant turn (before tool calls) keeps its own reasoning segment. _reasoning_segments: dict = {} + # Incremental folded indexes mirroring each reasoning buffer. The + # live echo strip consults them instead of re-walking the raw + # buffers: an unbounded raw scan of the growing transcript is + # quadratic on whitespace-heavy output (#7569 review). Each index + # is fed on every append and re-indexed after a strip. + _reasoning_segment_indexes: dict = {} + _stream_reasoning_index = _CompactEchoIndex() + _reasoning_buffer_index = _CompactEchoIndex() _current_reasoning_idx = 0 _tool_boundary_advanced = False _live_tool_calls = [] # tool progress fallback when final messages omit tool IDs @@ -10045,6 +10151,10 @@ def _run_agent_streaming( if _reasoning_buffer[0]: put('reasoning', {'text': _reasoning_buffer[0]}) _reasoning_buffer[0] = '' + # The folded index mirrors this buffer 1:1 — dropping the + # text without dropping the index would leave it describing + # text that is no longer there. + _reasoning_buffer_index.reset() def _emit_metering(): @@ -10084,30 +10194,37 @@ def _run_agent_streaming( nonlocal _reasoning_segments removed = False if stream_id in STREAM_REASONING_TEXT: - next_text, did_remove = _strip_compact_echo_suffix( - STREAM_REASONING_TEXT.get(stream_id, ''), - text, - ) - if did_remove: + cut = _stream_reasoning_index.cut_to(text) + if cut is not None: + next_text = STREAM_REASONING_TEXT.get(stream_id, '')[:cut].rstrip() STREAM_REASONING_TEXT[stream_id] = next_text + _stream_reasoning_index.reset() + _stream_reasoning_index.append(next_text) removed = True - next_buffer, did_remove_buffer = _strip_compact_echo_suffix(_reasoning_buffer[0], text) - if did_remove_buffer: + cut = _reasoning_buffer_index.cut_to(text) + if cut is not None: + next_buffer = _reasoning_buffer[0][:cut].rstrip() _reasoning_buffer[0] = next_buffer + _reasoning_buffer_index.reset() + _reasoning_buffer_index.append(next_buffer) removed = True for idx in (_current_reasoning_idx, _current_reasoning_idx - 1): if idx not in _reasoning_segments: continue - next_segment, did_remove_segment = _strip_compact_echo_suffix( - _reasoning_segments.get(idx, ''), - text, - ) - if not did_remove_segment: + segment_index = _reasoning_segment_indexes.get(idx) + if segment_index is None: continue + cut = segment_index.cut_to(text) + if cut is None: + continue + next_segment = _reasoning_segments.get(idx, '')[:cut].rstrip() if next_segment: _reasoning_segments[idx] = next_segment + segment_index.reset() + segment_index.append(next_segment) else: _reasoning_segments.pop(idx, None) + _reasoning_segment_indexes.pop(idx, None) removed = True break return removed @@ -10171,15 +10288,21 @@ def _run_agent_streaming( _reasoning_segments[_current_reasoning_idx] = ( _reasoning_segments.get(_current_reasoning_idx, '') + reasoning_delta ) + # Keep the folded index in step with the segment text. + _reasoning_segment_indexes.setdefault( + _current_reasoning_idx, _CompactEchoIndex() + ).append(reasoning_delta) # Mirror full concatenation to shared dict so cancel_stream() can persist # it (#1361 §A). Cancel only creates one partial message, so the flat # concatenation is correct there. # Lock-free GIL-atomic mirror — see the STREAMS_LOCK contract in on_token. if stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reasoning_delta + _stream_reasoning_index.append(reasoning_delta) # Accumulate into a coalescing buffer so every delta reaches the # browser — reasoning deltas are incremental, not idempotent. _reasoning_buffer[0] += reasoning_delta + _reasoning_buffer_index.append(reasoning_delta) # Throttle reasoning SSE events to ~10 Hz to avoid overwhelming the # frontend renderer. Each event triggers _parseStreamState() which # scans the full accumulated text — 10k+ reasoning tokens/second @@ -10298,10 +10421,14 @@ def _run_agent_streaming( _reasoning_segments[_current_reasoning_idx] = ( _reasoning_segments.get(_current_reasoning_idx, '') + reason_delta ) + _reasoning_segment_indexes.setdefault( + _current_reasoning_idx, _CompactEchoIndex() + ).append(reason_delta) # Mirror full concatenation to shared dict (#1361 §A) # Lock-free GIL-atomic mirror — see STREAMS_LOCK contract in on_token. if stream_id in STREAM_REASONING_TEXT: STREAM_REASONING_TEXT[stream_id] += reason_delta + _stream_reasoning_index.append(reason_delta) put('reasoning', {'text': reason_delta}) _metering_reasoning_deltas[0] += 1 meter().record_reasoning(stream_id, _metering_reasoning_deltas[0]) diff --git a/docs/rfcs/webui-run-state-consistency-contract.md b/docs/rfcs/webui-run-state-consistency-contract.md index 1954abca..d1362548 100644 --- a/docs/rfcs/webui-run-state-consistency-contract.md +++ b/docs/rfcs/webui-run-state-consistency-contract.md @@ -135,14 +135,19 @@ and 5; it does not mark every run-state boundary implemented. transcript is stripped from the reasoning copy so the restored snapshot shows the content once. That echo match is whitespace-insensitive and carries no fixed search window: a compact-equivalent suffix is recognized - however much interior whitespace stretches its raw span - (`_find_compact_echo_suffix_start` / `_strip_compact_echo_suffix` in - `api/streaming.py`, shared with the live-stream echo path; the live journal - replay in `api/routes.py` uses the same matcher). The retired windowed - variants could drop the strip when the span exceeded the window, duplicating - the interim text; the journal-reconstruction regression - (`tests/test_live_snapshot_echo_dedup.py`) pins the single-occurrence - result. + however much interior whitespace stretches its raw span. Both consumers + (the live-stream echo path and the journal replay in `api/routes.py`) + match through an incremental folded index (`_CompactEchoIndex` in + `api/streaming.py`): the folded view and its raw cut offsets are built as + each chunk is appended, so a probe costs O(len(candidate)) and never + rescans the transcript's whitespace. The retired windowed variants could + drop the strip when the span exceeded the window, duplicating the interim + text; the retired raw backward walk was correct but re-walked the span per + interim event, quadratic on whitespace-heavy transcripts. Regressions: + `tests/test_live_snapshot_echo_dedup.py` pins the single-occurrence + result, `tests/test_live_snapshot_echo_scan_scaling.py` pins the scaling + property (a fixed-size fixture cannot catch a per-interim rescan), and + `tests/test_compact_echo_index.py` pins index/oracle equivalence. 6. **Compression is not current intent.** Automatic compression summaries and reference cards are recovery/handoff material. They must not be treated as a new user request, active-turn content, or the default visible explanation for diff --git a/tests/test_compact_echo_index.py b/tests/test_compact_echo_index.py new file mode 100644 index 00000000..bdfa431d --- /dev/null +++ b/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" diff --git a/tests/test_live_snapshot_echo_scan_scaling.py b/tests/test_live_snapshot_echo_scan_scaling.py new file mode 100644 index 00000000..6f10c4a3 --- /dev/null +++ b/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" + )