mirror of
https://github.com/usestrix/strix.git
synced 2026-08-21 02:45:31 +02:00
fix(context): skip partial leading char on mid-character read offset
read_tool_output accepts arbitrary byte offsets; one landing inside a multi-byte character previously decoded to a replacement character. Trim the orphaned leading continuation bytes (their lead byte is on an earlier page) so every accepted offset returns valid UTF-8, without affecting forward paging offsets.
This commit is contained in:
@@ -177,6 +177,20 @@ def bound_and_store(text: str, *, max_lines: int, max_bytes: int) -> str:
|
||||
return _join(head, tail, notice)
|
||||
|
||||
|
||||
def _trim_incomplete_utf8_head(chunk: bytes) -> bytes:
|
||||
"""Drop leading continuation bytes so a chunk starting mid-character decodes cleanly.
|
||||
|
||||
A caller-supplied ``offset`` can land inside a multi-byte character; its
|
||||
orphaned continuation bytes (0b10xxxxxx) belong to a character whose lead
|
||||
byte is before ``offset``, so we skip forward to the next character start
|
||||
instead of emitting replacement characters.
|
||||
"""
|
||||
index = 0
|
||||
while index < len(chunk) and chunk[index] & 0xC0 == 0x80:
|
||||
index += 1
|
||||
return chunk[index:]
|
||||
|
||||
|
||||
def _trim_incomplete_utf8_tail(chunk: bytes) -> bytes:
|
||||
"""Drop a trailing partial UTF-8 sequence so ``chunk`` decodes cleanly.
|
||||
|
||||
@@ -239,9 +253,15 @@ def read_stored_output(output_id: str, *, offset: int = 0, limit: int = _PAGE_MA
|
||||
has_more = start + len(chunk) < size
|
||||
if has_more:
|
||||
chunk = _trim_incomplete_utf8_tail(chunk)
|
||||
next_offset = start + len(chunk)
|
||||
# Drop a partial leading character when an arbitrary offset lands mid-char.
|
||||
# This never removes content: the previous page's tail-trim guarantees a
|
||||
# forward-paged offset starts on a boundary, so this only affects an
|
||||
# explicit caller-chosen offset (whose partial char began on an earlier page).
|
||||
if start > 0:
|
||||
chunk = _trim_incomplete_utf8_head(chunk)
|
||||
shown = chunk.decode("utf-8", errors="replace")
|
||||
if has_more:
|
||||
next_offset = start + len(chunk)
|
||||
shown += (
|
||||
"\n\n[... more; call read_tool_output("
|
||||
f'output_id="{output_id}", offset={next_offset}) to continue ...]'
|
||||
|
||||
@@ -185,6 +185,27 @@ def test_read_stored_output_pages_multibyte_without_corruption(tmp_path: Path) -
|
||||
assert collected == text
|
||||
|
||||
|
||||
def test_read_stored_output_offset_inside_char_returns_valid_utf8(tmp_path: Path) -> None:
|
||||
# An arbitrary caller-chosen offset that lands inside a 4-byte char must skip
|
||||
# the partial leading char rather than emit a replacement character.
|
||||
configure_output_store(tmp_path)
|
||||
text = "😀" * 100
|
||||
output_id = re.search(
|
||||
r'output_id="([0-9a-f]{32})"',
|
||||
bound_and_store(text, max_lines=4, max_bytes=200),
|
||||
)
|
||||
assert output_id is not None
|
||||
oid = output_id.group(1)
|
||||
|
||||
# Byte 1 is inside the first emoji (each 😀 is 4 bytes).
|
||||
page = read_stored_output(oid, offset=1, limit=1_000)
|
||||
body = page.partition("\n\n[... more;")[0]
|
||||
assert "\ufffd" not in body
|
||||
assert body == body.encode("utf-8").decode("utf-8")
|
||||
# The partial leading char is skipped; content resumes at the next boundary.
|
||||
assert body.startswith("😀")
|
||||
|
||||
|
||||
def test_read_stored_output_rejects_traversal(tmp_path: Path) -> None:
|
||||
configure_output_store(tmp_path)
|
||||
assert "Invalid output_id" in read_stored_output("../../etc/passwd")
|
||||
|
||||
Reference in New Issue
Block a user