diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py index a5fa3fed7d..fe04da6bee 100644 --- a/tests/tools/test_process_registry.py +++ b/tests/tools/test_process_registry.py @@ -170,6 +170,107 @@ def test_reader_loop_streams_incremental_chunks_from_read1(registry, monkeypatch assert moved == ["proc_reader_live"] +# ========================================================================= +# Incremental UTF-8 decoding across chunk boundaries +# (ported from openclaw/openclaw#112325) +# ========================================================================= + + +class _FakeChunkBuffer: + def __init__(self, chunks): + self._chunks = list(chunks) + + def read1(self, _n): + if self._chunks: + return self._chunks.pop(0) + return b"" + + +class _FakeChunkStdout: + def __init__(self, chunks): + self.buffer = _FakeChunkBuffer(chunks) + + +class _FakeChunkProcess: + def __init__(self, chunks): + self.stdout = _FakeChunkStdout(chunks) + self.returncode = 0 + + def wait(self, timeout=None): + return 0 + + +def _run_reader(registry, monkeypatch, chunks, sid="proc_utf8"): + session = _make_session(sid=sid) + session.process = _FakeChunkProcess(chunks) + monkeypatch.setattr(registry, "_check_watch_patterns", lambda _s, _c: None) + monkeypatch.setattr(registry, "_emit_output", lambda _s, _c: None) + monkeypatch.setattr(registry, "_move_to_finished", lambda _s: None) + registry._reader_loop(session) + return session + + +def test_reader_loop_reassembles_multibyte_char_split_across_chunks(registry, monkeypatch): + """A UTF-8 char split across two read1() chunks must not become U+FFFD. + + Before the incremental decoder, each chunk was decoded statelessly with + ``errors="replace"``, so ``é`` (0xC3 0xA9) straddling a 4096-byte read + boundary decoded as two replacement characters. + """ + session = _run_reader(registry, monkeypatch, [b"caf\xc3", b"\xa9 ok\n"]) + assert session.output_buffer == "café ok\n" + assert "\ufffd" not in session.output_buffer + + +def test_reader_loop_reassembles_four_byte_char_split_three_ways(registry, monkeypatch): + """A 4-byte emoji fragmented across three reads reassembles cleanly.""" + session = _run_reader(registry, monkeypatch, [b"\xf0", b"\x9f\x92", b"\xa9\n"]) + assert session.output_buffer == "\U0001f4a9\n" + + +def test_reader_loop_flushes_truncated_multibyte_tail_at_eof(registry, monkeypatch): + """A sequence truncated by process exit flushes as a single U+FFFD.""" + session = _run_reader(registry, monkeypatch, [b"ok \xe2\x82"]) + assert session.output_buffer == "ok \ufffd" + + +def test_reader_loop_still_replaces_genuinely_invalid_bytes(registry, monkeypatch): + """Truly invalid bytes keep the errors="replace" behavior.""" + session = _run_reader(registry, monkeypatch, [b"ok\xffdone\n"]) + assert session.output_buffer == "ok\ufffddone\n" + + +def test_pty_reader_loop_reassembles_multibyte_char_split_across_chunks(registry, monkeypatch): + """The PTY reader gets the same incremental-decode treatment.""" + + class _FakePty: + def __init__(self, chunks): + self._chunks = list(chunks) + self.exitstatus = 0 + + def isalive(self): + return bool(self._chunks) + + def read(self, _n): + if self._chunks: + return self._chunks.pop(0) + raise EOFError + + def wait(self): + return 0 + + session = _make_session(sid="proc_pty_utf8") + session._pty = _FakePty([b"caf\xc3", b"\xa9\n"]) + monkeypatch.setattr(registry, "_check_watch_patterns", lambda _s, _c: None) + monkeypatch.setattr(registry, "_emit_output", lambda _s, _c: None) + monkeypatch.setattr(registry, "_move_to_finished", lambda _s: None) + + registry._pty_reader_loop(session) + + assert session.output_buffer == "café\n" + assert "\ufffd" not in session.output_buffer + + # ========================================================================= # Orphaned-pipe reconciliation (issue #17327) # ========================================================================= diff --git a/tools/process_registry.py b/tools/process_registry.py index 7daf74b2e3..1cfc7ee7d0 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -29,6 +29,7 @@ Usage: process_registry.kill(session.id) """ +import codecs import json import logging import os @@ -960,6 +961,14 @@ class ProcessRegistry: lazy reconcile in poll()/wait() remains the safety net. """ first_chunk = True + # Incremental decoder: raw pipe reads can split a multibyte UTF-8 + # character across two read1() chunks. A stateless per-chunk + # ``bytes.decode(errors="replace")`` turns both halves into U+FFFD + # mojibake. The incremental decoder holds the partial sequence until + # the continuation bytes arrive — same treatment the foreground path + # already has in ``tools/environments/base.py::_wait_for_process``. + # (Ported from openclaw/openclaw#112325.) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") def _append_chunk(chunk: str): nonlocal first_chunk @@ -1007,7 +1016,9 @@ class ProcessRegistry: raw = raw_read(4096) if not raw: break # true EOF — all writers closed - _append_chunk(raw.decode("utf-8", errors="replace")) + chunk = decoder.decode(raw) + if chunk: + _append_chunk(chunk) idle_after_exit = 0 elif proc.poll() is not None: # Direct child is gone and the pipe was idle for @@ -1024,7 +1035,9 @@ class ProcessRegistry: raw = raw_read(4096) if not raw: break - chunk = raw.decode("utf-8", errors="replace") + chunk = decoder.decode(raw) + if not chunk: + continue # partial multibyte sequence — wait for more bytes else: # Fallback for mocked/alternate streams without a buffered raw # interface. This may be less "live", but keeps compatibility. @@ -1036,6 +1049,15 @@ class ProcessRegistry: except Exception as e: logger.debug("Process stdout reader ended: %s", e) finally: + # Flush any bytes still pending in the incremental decoder (a + # truncated multibyte sequence at EOF becomes one U+FFFD instead + # of being dropped silently). + try: + tail = decoder.decode(b"", final=True) + if tail: + _append_chunk(tail) + except Exception: + pass # Always reap the child to prevent zombie processes. try: session.process.wait(timeout=5) @@ -1108,19 +1130,28 @@ class ProcessRegistry: def _pty_reader_loop(self, session: ProcessSession): """Background thread: read output from a PTY process.""" pty = session._pty + # PTY reads can split a multibyte UTF-8 character across chunks just + # like pipe reads — hold partial sequences until the rest arrives. + # (Ported from openclaw/openclaw#112325.) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + + def _append_text(text: str): + with session._lock: + session.output_buffer += text + if len(session.output_buffer) > session.max_output_chars: + session.output_buffer = session.output_buffer[-session.max_output_chars:] + self._check_watch_patterns(session, text) + self._emit_output(session, text) + try: while pty.isalive(): try: chunk = pty.read(4096) if chunk: - # ptyprocess returns bytes - text = chunk if isinstance(chunk, str) else chunk.decode("utf-8", errors="replace") - with session._lock: - session.output_buffer += text - if len(session.output_buffer) > session.max_output_chars: - session.output_buffer = session.output_buffer[-session.max_output_chars:] - self._check_watch_patterns(session, text) - self._emit_output(session, text) + # ptyprocess returns bytes; pywinpty returns str + text = chunk if isinstance(chunk, str) else decoder.decode(chunk) + if text: + _append_text(text) except EOFError: break except Exception: @@ -1128,6 +1159,14 @@ class ProcessRegistry: except Exception as e: logger.debug("PTY stdout reader ended: %s", e) + # Flush any partial multibyte sequence held by the decoder. + try: + tail = decoder.decode(b"", final=True) + if tail: + _append_text(tail) + except Exception: + pass + # Process exited try: pty.wait()