From b163ae284982e32daa3fa699565cb6539a3397b0 Mon Sep 17 00:00:00 2001 From: Jorge Polanco <55784702+Jorge-Polanco-Roque@users.noreply.github.com> Date: Sat, 29 Aug 2026 21:06:29 -0600 Subject: [PATCH 1/4] fix(sandbox): don't corrupt a UTF-8 char split across PTY windows collect_pty_output decoded each collection window with errors="replace". Because PTY output is drained in repeated windows over one persistent deque, a multi-byte character whose bytes land either side of a window boundary was decoded as two partial sequences and both halves became U+FFFD, silently corrupting the text. Hold back a trailing incomplete UTF-8 sequence and push it to the front of the shared deque so the next window completes it, unless the process is done (no further windows). Adds a regression test. Fixes #4744 --- src/agents/sandbox/session/pty_output.py | 40 ++++++++++++++++++++++++ tests/sandbox/test_pty_output.py | 39 +++++++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/src/agents/sandbox/session/pty_output.py b/src/agents/sandbox/session/pty_output.py index 25cbe774e7..91766b2dfb 100644 --- a/src/agents/sandbox/session/pty_output.py +++ b/src/agents/sandbox/session/pty_output.py @@ -8,6 +8,32 @@ from .pty_types import truncate_text_by_tokens +def _incomplete_utf8_tail_len(buf: bytes | bytearray) -> int: + """Return the number of trailing bytes that begin an incomplete UTF-8 sequence. + + Returns 0 when ``buf`` ends on a character boundary or with bytes that cannot + start a longer sequence, so decoding the whole buffer is safe. + """ + for i in range(1, min(4, len(buf)) + 1): + byte = buf[-i] + if byte & 0xC0 == 0x80: + # Continuation byte; keep scanning back for its lead byte. + continue + if byte & 0x80 == 0x00: + seq_len = 1 + elif byte & 0xE0 == 0xC0: + seq_len = 2 + elif byte & 0xF0 == 0xE0: + seq_len = 3 + elif byte & 0xF8 == 0xF0: + seq_len = 4 + else: + # Invalid lead byte; nothing worth holding back. + return 0 + return i if i < seq_len else 0 + return 0 + + async def collect_pty_output( *, output_chunks: deque[bytes], @@ -20,6 +46,7 @@ async def collect_pty_output( """Collect and truncate PTY output until the deadline or provider completion.""" deadline = time.monotonic() + (yield_time_ms / 1000) output = bytearray() + final = False while True: async with output_lock: @@ -33,6 +60,7 @@ async def collect_pty_output( async with output_lock: while output_chunks: output.extend(output_chunks.popleft()) + final = True break remaining_s = deadline - time.monotonic() @@ -45,6 +73,18 @@ async def collect_pty_output( break output_notify.clear() + # A multi-byte character may be split across two collection windows. Unless + # the process is done (no further windows will arrive), hold back a trailing + # incomplete UTF-8 sequence and push it to the front of the shared deque so + # the next window can complete it, instead of both halves decoding to U+FFFD. + if not final: + tail_len = _incomplete_utf8_tail_len(output) + if tail_len: + tail = bytes(output[-tail_len:]) + del output[-tail_len:] + async with output_lock: + output_chunks.appendleft(tail) + text = output.decode("utf-8", errors="replace") truncated, original_token_count = truncate_text_by_tokens(text, max_output_tokens) return truncated.encode("utf-8", errors="replace"), original_token_count diff --git a/tests/sandbox/test_pty_output.py b/tests/sandbox/test_pty_output.py index f15bcf85b4..7a16869fa4 100644 --- a/tests/sandbox/test_pty_output.py +++ b/tests/sandbox/test_pty_output.py @@ -57,3 +57,42 @@ def mark_done() -> bool: assert output == b"before done after done" assert original_token_count is None + + +@pytest.mark.asyncio +async def test_collect_pty_output_preserves_char_split_across_windows() -> None: + # A multi-byte character split across two collection windows over the same + # persistent deque must not be corrupted into U+FFFD. Regression for #4744. + text = "héllo wörld" + raw = text.encode("utf-8") + split = raw.index(b"\xb6") # inside the "ö": its trailing continuation byte + + output_chunks: deque[bytes] = deque([raw[:split]]) + output_lock = asyncio.Lock() + output_notify = asyncio.Event() + + # First, non-final window: returns on the deadline and must hold back the + # incomplete tail for the next window instead of decoding it to U+FFFD. + first, _ = await collect_pty_output( + output_chunks=output_chunks, + output_lock=output_lock, + output_notify=output_notify, + is_done=lambda: False, + yield_time_ms=10, + max_output_tokens=None, + ) + + # Second, final window: the rest of the character arrives. + output_chunks.append(raw[split:]) + second, _ = await collect_pty_output( + output_chunks=output_chunks, + output_lock=output_lock, + output_notify=output_notify, + is_done=lambda: True, + yield_time_ms=10, + max_output_tokens=None, + ) + + combined = (first + second).decode("utf-8") + assert combined == text + assert "�" not in combined From 4c1dd6ae118a08cfecb14c3e0ba9351e85a354b2 Mon Sep 17 00:00:00 2001 From: Jorge Polanco <55784702+Jorge-Polanco-Roque@users.noreply.github.com> Date: Sat, 29 Aug 2026 22:13:01 -0600 Subject: [PATCH 2/4] pty: use an incremental UTF-8 decoder; flush tail on completion Addresses review feedback: - Invalid lead bytes (0xC0/0xC1, 0xF5-0xF7) were treated as incomplete and withheld; decode with codecs.getincrementaldecoder so they become U+FFFD immediately and only genuinely incomplete sequences are buffered. - Re-check is_done() at the finalization boundary so a process that finishes after the deadline flushes the held tail instead of requeueing it into an entry that is about to be discarded. Adds an invalid-byte regression test. --- src/agents/sandbox/session/pty_output.py | 52 ++++++++---------------- tests/sandbox/test_pty_output.py | 19 +++++++++ 2 files changed, 35 insertions(+), 36 deletions(-) diff --git a/src/agents/sandbox/session/pty_output.py b/src/agents/sandbox/session/pty_output.py index 91766b2dfb..486debc550 100644 --- a/src/agents/sandbox/session/pty_output.py +++ b/src/agents/sandbox/session/pty_output.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import codecs import time from collections import deque from collections.abc import Callable @@ -8,32 +9,6 @@ from .pty_types import truncate_text_by_tokens -def _incomplete_utf8_tail_len(buf: bytes | bytearray) -> int: - """Return the number of trailing bytes that begin an incomplete UTF-8 sequence. - - Returns 0 when ``buf`` ends on a character boundary or with bytes that cannot - start a longer sequence, so decoding the whole buffer is safe. - """ - for i in range(1, min(4, len(buf)) + 1): - byte = buf[-i] - if byte & 0xC0 == 0x80: - # Continuation byte; keep scanning back for its lead byte. - continue - if byte & 0x80 == 0x00: - seq_len = 1 - elif byte & 0xE0 == 0xC0: - seq_len = 2 - elif byte & 0xF0 == 0xE0: - seq_len = 3 - elif byte & 0xF8 == 0xF0: - seq_len = 4 - else: - # Invalid lead byte; nothing worth holding back. - return 0 - return i if i < seq_len else 0 - return 0 - - async def collect_pty_output( *, output_chunks: deque[bytes], @@ -73,18 +48,23 @@ async def collect_pty_output( break output_notify.clear() - # A multi-byte character may be split across two collection windows. Unless - # the process is done (no further windows will arrive), hold back a trailing - # incomplete UTF-8 sequence and push it to the front of the shared deque so - # the next window can complete it, instead of both halves decoding to U+FFFD. + # Re-check completion at the finalization boundary: if the process finished + # after the loop broke on its deadline, there is no further window, so we + # must flush rather than requeue (the requeued entry would be discarded). + final = final or is_done() + + # A multi-byte character can be split across two collection windows. Decode + # incrementally so a genuinely incomplete trailing sequence is buffered and + # pushed back for the next window to complete, while invalid bytes still + # decode to U+FFFD immediately (matching errors="replace"). When the process + # is done the decoder is finalized, flushing any pending bytes. + decoder = codecs.getincrementaldecoder("utf-8")("replace") + text = decoder.decode(bytes(output), final=final) if not final: - tail_len = _incomplete_utf8_tail_len(output) - if tail_len: - tail = bytes(output[-tail_len:]) - del output[-tail_len:] + buffered = decoder.getstate()[0] + if buffered: async with output_lock: - output_chunks.appendleft(tail) + output_chunks.appendleft(bytes(buffered)) - text = output.decode("utf-8", errors="replace") truncated, original_token_count = truncate_text_by_tokens(text, max_output_tokens) return truncated.encode("utf-8", errors="replace"), original_token_count diff --git a/tests/sandbox/test_pty_output.py b/tests/sandbox/test_pty_output.py index 7a16869fa4..877a01f66a 100644 --- a/tests/sandbox/test_pty_output.py +++ b/tests/sandbox/test_pty_output.py @@ -96,3 +96,22 @@ async def test_collect_pty_output_preserves_char_split_across_windows() -> None: combined = (first + second).decode("utf-8") assert combined == text assert "�" not in combined + + +@pytest.mark.asyncio +async def test_collect_pty_output_replaces_invalid_byte_without_withholding() -> None: + # An invalid UTF-8 lead byte (0xC0) must decode to U+FFFD immediately in a + # non-final window, not be buffered as if it were an incomplete sequence. + output_chunks: deque[bytes] = deque([b"ok\xc0"]) + + out, _ = await collect_pty_output( + output_chunks=output_chunks, + output_lock=asyncio.Lock(), + output_notify=asyncio.Event(), + is_done=lambda: False, + yield_time_ms=10, + max_output_tokens=None, + ) + + assert out.decode("utf-8") == "ok�" + assert not output_chunks # nothing was withheld From ca82462ba15397468a98a94493dd9dad99db8120 Mon Sep 17 00:00:00 2001 From: Jorge Polanco <55784702+Jorge-Polanco-Roque@users.noreply.github.com> Date: Sun, 30 Aug 2026 09:31:19 -0600 Subject: [PATCH 3/4] pty: route the Modal collector through the shared decoder The Modal backend has its own _collect_pty_output that decoded each window independently, so a multi-byte character split across windows was still corrupted there. Extract the boundary-safe decode into a shared decode_pty_window() helper, carry the incomplete tail on the Modal entry across windows, and re-check completion at the finalization boundary. Adds a unit test for decode_pty_window. --- .../extensions/sandbox/modal/sandbox.py | 16 ++++++++-- src/agents/sandbox/session/pty_output.py | 32 ++++++++++++------- tests/sandbox/test_pty_output.py | 23 ++++++++++++- 3 files changed, 56 insertions(+), 15 deletions(-) diff --git a/src/agents/extensions/sandbox/modal/sandbox.py b/src/agents/extensions/sandbox/modal/sandbox.py index 848b8e12a3..b58cacb4e3 100644 --- a/src/agents/extensions/sandbox/modal/sandbox.py +++ b/src/agents/extensions/sandbox/modal/sandbox.py @@ -64,6 +64,7 @@ _settle_mount_transition, _terminate_ambiguous_mount_session, ) +from ....sandbox.session.pty_output import decode_pty_window from ....sandbox.session.pty_types import ( PTY_PROCESSES_MAX, PTY_PROCESSES_WARNING, @@ -489,6 +490,8 @@ class _ModalPtyProcessEntry: stderr_iter: AsyncIterator[object] | None = None stdout_read_task: asyncio.Task[object] | None = None stderr_read_task: asyncio.Task[object] | None = None + # Trailing incomplete UTF-8 bytes carried over to the next collection window. + pty_output_tail: bytes = b"" class ModalSandboxSession(BaseSandboxSession): @@ -983,7 +986,9 @@ async def _collect_pty_output( max_output_tokens: int | None, ) -> tuple[bytes, int | None]: deadline = time.monotonic() + (yield_time_ms / 1000) - chunks = bytearray() + chunks = bytearray(entry.pty_output_tail) + entry.pty_output_tail = b"" + final = False while True: stdout_chunk = await self._read_modal_stream(entry=entry, stream_name="stdout") @@ -1002,6 +1007,7 @@ async def _collect_pty_output( stderr_chunks = await self._drain_modal_stream(entry=entry, stream_name="stderr") chunks.extend(stdout_chunks) chunks.extend(stderr_chunks) + final = True break if not stdout_chunk and not stderr_chunk: @@ -1010,7 +1016,13 @@ async def _collect_pty_output( break await asyncio.sleep(min(_PTY_POLL_INTERVAL_S, remaining_s)) - text = chunks.decode("utf-8", errors="replace") + # Re-check completion so a process that finished after the deadline + # flushes the held tail instead of carrying it on a discarded entry. + if not final and await self._peek_exit_code(entry.process) is not None: + final = True + + text, leftover = decode_pty_window(bytes(chunks), final=final) + entry.pty_output_tail = leftover truncated_text, original_token_count = truncate_text_by_tokens(text, max_output_tokens) return truncated_text.encode("utf-8", errors="replace"), original_token_count diff --git a/src/agents/sandbox/session/pty_output.py b/src/agents/sandbox/session/pty_output.py index 486debc550..6775dcb1a3 100644 --- a/src/agents/sandbox/session/pty_output.py +++ b/src/agents/sandbox/session/pty_output.py @@ -9,6 +9,22 @@ from .pty_types import truncate_text_by_tokens +def decode_pty_window(data: bytes, *, final: bool) -> tuple[str, bytes]: + """Decode one window of PTY output. + + PTY output is collected in repeated windows over one persistent stream, so a + multi-byte character can be split across a window boundary. Decode + incrementally: invalid bytes still become U+FFFD immediately (matching + ``errors="replace"``), while a genuinely incomplete trailing sequence is + returned as ``leftover`` for the caller to prepend to the next window. When + ``final`` is set there is no next window, so nothing is held back. + """ + decoder = codecs.getincrementaldecoder("utf-8")("replace") + text = decoder.decode(data, final=final) + leftover = b"" if final else bytes(decoder.getstate()[0]) + return text, leftover + + async def collect_pty_output( *, output_chunks: deque[bytes], @@ -53,18 +69,10 @@ async def collect_pty_output( # must flush rather than requeue (the requeued entry would be discarded). final = final or is_done() - # A multi-byte character can be split across two collection windows. Decode - # incrementally so a genuinely incomplete trailing sequence is buffered and - # pushed back for the next window to complete, while invalid bytes still - # decode to U+FFFD immediately (matching errors="replace"). When the process - # is done the decoder is finalized, flushing any pending bytes. - decoder = codecs.getincrementaldecoder("utf-8")("replace") - text = decoder.decode(bytes(output), final=final) - if not final: - buffered = decoder.getstate()[0] - if buffered: - async with output_lock: - output_chunks.appendleft(bytes(buffered)) + text, leftover = decode_pty_window(bytes(output), final=final) + if leftover: + async with output_lock: + output_chunks.appendleft(leftover) truncated, original_token_count = truncate_text_by_tokens(text, max_output_tokens) return truncated.encode("utf-8", errors="replace"), original_token_count diff --git a/tests/sandbox/test_pty_output.py b/tests/sandbox/test_pty_output.py index 877a01f66a..bb7d942d15 100644 --- a/tests/sandbox/test_pty_output.py +++ b/tests/sandbox/test_pty_output.py @@ -5,7 +5,7 @@ import pytest -from agents.sandbox.session.pty_output import collect_pty_output +from agents.sandbox.session.pty_output import collect_pty_output, decode_pty_window @pytest.mark.asyncio @@ -115,3 +115,24 @@ async def test_collect_pty_output_replaces_invalid_byte_without_withholding() -> assert out.decode("utf-8") == "ok�" assert not output_chunks # nothing was withheld + + +def test_decode_pty_window_holds_incomplete_and_replaces_invalid() -> None: + raw = "wörld".encode() # b"w\xc3\xb6rld" + + # A genuinely incomplete trailing sequence is returned as leftover. + text, leftover = decode_pty_window(raw[:2], final=False) # b"w\xc3" + assert leftover == b"\xc3" + text2, leftover2 = decode_pty_window(leftover + raw[2:], final=False) + assert text + text2 == "wörld" + assert leftover2 == b"" + + # An invalid byte becomes U+FFFD immediately, nothing held back. + text3, leftover3 = decode_pty_window(b"ok\xc0", final=False) + assert text3 == "ok�" + assert leftover3 == b"" + + # final=True flushes any pending bytes as U+FFFD. + text4, leftover4 = decode_pty_window(b"w\xc3", final=True) + assert text4 == "w�" + assert leftover4 == b"" From 1954f38e6fe406f88e1767029b5f33b2fc38b994 Mon Sep 17 00:00:00 2001 From: Jorge Polanco <55784702+Jorge-Polanco-Roque@users.noreply.github.com> Date: Sun, 30 Aug 2026 09:54:11 -0600 Subject: [PATCH 4/4] pty(modal): keep the carried-over tail durable across cancellation Do not clear entry.pty_output_tail before the awaited reads; leave it on the entry until a decode commits, so a cancelled collection retains the deferred bytes instead of losing them. --- src/agents/extensions/sandbox/modal/sandbox.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/agents/extensions/sandbox/modal/sandbox.py b/src/agents/extensions/sandbox/modal/sandbox.py index b58cacb4e3..48065db78b 100644 --- a/src/agents/extensions/sandbox/modal/sandbox.py +++ b/src/agents/extensions/sandbox/modal/sandbox.py @@ -986,8 +986,10 @@ async def _collect_pty_output( max_output_tokens: int | None, ) -> tuple[bytes, int | None]: deadline = time.monotonic() + (yield_time_ms / 1000) + # Read the carried-over tail but leave it on the entry until we commit a + # decode below, so a cancelled collection keeps it durable rather than + # losing those bytes. chunks = bytearray(entry.pty_output_tail) - entry.pty_output_tail = b"" final = False while True: