diff --git a/strix/tools/output_store.py b/strix/tools/output_store.py index 92390ce9..3e622c08 100644 --- a/strix/tools/output_store.py +++ b/strix/tools/output_store.py @@ -48,19 +48,28 @@ def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: Truncation happens on whichever limit is hit first (line count or UTF-8 byte size). The removed middle is replaced with a notice recording how many lines and bytes were dropped so the agent knows output was elided. + ``max_bytes`` bounds the *entire* joined result, notice and separators + included. """ lines = text.split("\n") total_bytes = _byte_len(text) if len(lines) <= max_lines and total_bytes <= max_bytes: return text + # Reserve room for the notice and its two blank-line separators so the + # head+tail slices can't consume the whole budget and push the persisted + # value over max_bytes. Upper-bound the notice with the largest possible + # counts; the real notice is never longer. ``+ 4`` covers the separators. + notice_overhead = _byte_len(_TRUNCATION_NOTICE.format(lines=len(lines), bytes=total_bytes)) + 4 + byte_budget = max(2, max_bytes - notice_overhead) + head_lines = max(1, max_lines // 2) tail_lines = max_lines - head_lines head = "\n".join(lines[:head_lines]) tail = "\n".join(lines[len(lines) - tail_lines :]) if tail_lines > 0 else "" # Enforce the byte budget even when the line count alone was fine. - half_bytes = max(1, max_bytes // 2) + half_bytes = max(1, byte_budget // 2) if _byte_len(head) > half_bytes: head = _take_prefix(head, half_bytes) if tail and _byte_len(tail) > half_bytes: diff --git a/tests/test_output_store.py b/tests/test_output_store.py index 2dc1906f..14bff19c 100644 --- a/tests/test_output_store.py +++ b/tests/test_output_store.py @@ -28,7 +28,9 @@ def test_byte_limit_enforced_on_single_long_line() -> None: bounded = bound_text(text, max_lines=2_000, max_bytes=1_000) assert "truncated" in bounded - assert len(bounded.encode("utf-8")) < 3_000 + # The whole joined result (head + tail + notice + separators) honours the + # configured maximum, not just the head/tail slices. + assert len(bounded.encode("utf-8")) <= 1_000 def test_multibyte_characters_not_split() -> None: