fix(context): reserve notice budget so bounded output honors max_bytes

The head+tail slices could each take half of max_bytes, then the
truncation notice and its separators were appended on top, so the value
persisted to history could exceed the configured maximum. Reserve an
upper bound for the notice (and separators) out of the byte budget before
slicing so the whole joined result stays within max_bytes.
This commit is contained in:
Ahmed Allam
2026-07-25 23:06:40 +00:00
parent ce358aa879
commit aac59de1e5
2 changed files with 13 additions and 2 deletions
+10 -1
View File
@@ -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:
+3 -1
View File
@@ -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: