fix(process): decode background process output with incremental UTF-8 decoders

Port from openclaw/openclaw#112325: multibyte UTF-8 characters split across a 4096-byte pipe or PTY read boundary were decoded statelessly per chunk with errors='replace', corrupting both halves into U+FFFD mojibake in background process output (poll/log/wait/completion notifications). The foreground path already used an incremental decoder (tools/environments/base.py::_wait_for_process); this applies the same treatment to the background reader loops: - _reader_loop (select and blocking paths): one codecs.getincrementaldecoder('utf-8') per reader holds partial sequences across chunks; the finally block flushes a truncated tail as a single U+FFFD instead of dropping it. - _pty_reader_loop: same treatment for ptyprocess byte chunks (pywinpty str chunks pass through unchanged). Genuinely invalid bytes keep errors='replace' behavior.
2026-07-26 17:10:55 -07:00
parent 89f920901b
commit 7fb5d2bc39
2 changed files with 150 additions and 10 deletions
@@ -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)
 # =========================================================================
@@ -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()