fix(tools): Forward build subprocess stdout as UTF-8 bytes on Windows

tools/idf_py_actions/tools.py

@@ -349,6 +349,21 @@ def fit_text_in_terminal(out: str) -> str:
     return out
 
 
+# ANSI SGR / CSI sequences and Ninja-style progression lines are 7-bit; keep
+# readline forwarding on bytes so CMake UTF-8 pipe output is not re-encoded
+# through a legacy TextIOWrapper (cp1252) layer on Windows.
+_ANSI_ESCAPE_BYTES = re.compile(rb'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
+_PROGRESSION_BYTES = re.compile(rb'^\[\d+/\d+\]|.*\(\d+ \%\)$')
+
+
+def delete_ansi_escape_bytes(data: bytes) -> bytes:
+    return _ANSI_ESCAPE_BYTES.sub(b'', data)
+
+
+def is_progression_bytes(line: bytes) -> bool:
+    return _PROGRESSION_BYTES.match(line) is not None
+
+
 class RunTool:
     def __init__(
         self,
@@ -480,10 +495,23 @@ class RunTool:
             ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
             return ansi_escape.sub('', text)
 
+        def write_text_maybe_illencoded(stream: TextIO, text: str) -> None:
+            """Write ``text`` to ``stream``; on narrow Windows stdio, fall back so the build does not abort."""
+            try:
+                stream.write(text)
+                stream.flush()
+            except UnicodeEncodeError:
+                enc = getattr(stream, 'encoding', None) or 'ascii'
+                safe = text.encode(enc, errors='backslashreplace').decode(enc)
+                stream.write(safe)
+                stream.flush()
+
         def print_progression(output: str) -> None:
             # Print a new line on top of the previous line
-            print('\r' + fit_text_in_terminal(output.strip('\n\r')) + '\x1b[K', end='', file=output_stream)
+            write_text_maybe_illencoded(
-            output_stream.flush()
+                output_stream,
+                '\r' + fit_text_in_terminal(output.strip('\n\r')) + '\x1b[K',
+            )
 
         def is_progression(output: str) -> bool:
             # try to find possible progression by a pattern match
@@ -491,10 +519,12 @@ class RunTool:
                 return True
             return False
 
-        async def read_stream() -> str | None:
+        async def read_stream_bytes() -> bytes | None:
             try:
                 output_b = await input_stream.readline()
-                return output_b.decode(errors='ignore')
+                if not output_b:
+                    return None
+                return output_b
             except (asyncio.LimitOverrunError, asyncio.IncompleteReadError) as e:
                 print(e, file=sys.stderr)
                 return None
@@ -516,6 +546,23 @@ class RunTool:
                         # and still can not decode it we can just ignore some bytes
                         return buffer.decode(errors='replace')
 
+        def write_stdout_bytes(data: bytes) -> None:
+            """Write raw subprocess bytes to the underlying binary stream (UTF-8 from CMake, no TextIO re-encode)."""
+            binary_buf = getattr(output_stream, 'buffer', None)
+            if binary_buf is not None:
+                binary_buf.write(data)
+                binary_buf.flush()
+            else:
+                write_text_maybe_illencoded(output_stream, data.decode('utf-8', errors='replace'))
+
+        def write_stdout_after_progression_line() -> None:
+            """Emit a newline after a progression ``\\r`` line (binary when possible)."""
+            if not self.convert_output and getattr(output_stream, 'buffer', None) is not None:
+                output_stream.buffer.write(os.linesep.encode('ascii'))
+                output_stream.buffer.flush()
+            else:
+                write_text_maybe_illencoded(output_converter, os.linesep)
+
         # use ANSI color converter for Monitor on Windows
         output_converter = get_ansi_converter(output_stream) if self.convert_output else output_stream
 
@@ -526,47 +573,60 @@ class RunTool:
         is_progression_processing_enabled = self.force_progression and output_stream.isatty() and '-v' not in self.args
 
         try:
-            # The command output from asyncio stream already contains OS specific line ending,
+            # Binary log: subprocess lines are written as received (ANSI stripped for the file).
-            # because it's read in as bytes and decoded to string. On Windows "output" already
+            with open(output_filename, 'wb') as output_file:
-            # contains CRLF. Use "newline=''" to prevent python to convert CRLF into CRCRLF.
+                output_file.write('Command: {}\n'.format(' '.join(self.args)).encode('utf-8'))
-            # Please see "newline" description at https://docs.python.org/3/library/functions.html#open
-            with open(output_filename, 'w', encoding='utf8', newline='') as output_file:
-                # Log the command arguments.
-                output_file.write('Command: {}\n'.format(' '.join(self.args)))
                 while True:
                     if self.interactive:
                         output = await read_interactive_stream()
-                    else:
+                        if not output:
-                        output = await read_stream()
+                            break
-                    if not output:
-                        break
 
-                    output_noescape = delete_ansi_escape(output)
+                        output_noescape = delete_ansi_escape(output)
-                    # Always remove escape sequences when writing the build log.
+                        output_file.write(output_noescape.encode('utf-8'))
-                    output_file.write(output_noescape)
+                        if not output_stream.isatty():
-                    # If idf.py output is redirected and the output stream is not a TTY,
+                            output = output_noescape
-                    # strip the escape sequences as well.
-                    # (There shouldn't be any, but just in case.)
-                    if not output_stream.isatty():
-                        output = output_noescape
 
-                    if is_progression_processing_enabled and is_progression(output):
+                        if is_progression_processing_enabled and is_progression(output):
-                        print_progression(output)
+                            print_progression(output)
-                        is_progression_last_line = True
+                            is_progression_last_line = True
-                    else:
+                        else:
-                        if is_progression_last_line:
+                            if is_progression_last_line:
-                            output_converter.write(os.linesep)
+                                write_stdout_after_progression_line()
-                            is_progression_last_line = False
+                                is_progression_last_line = False
-                        output_converter.write(output)
+                            write_text_maybe_illencoded(output_converter, output)
-                        output_converter.flush()
 
-                        # process hints for last line and print them right away
-                        if self.interactive:
                             last_line += output
                             if last_line[-1] == '\n':
                                 for hint in generate_hints_buffer(last_line, hints):
                                     yellow_print(hint)
                                 last_line = ''
+                    else:
+                        output_b = await read_stream_bytes()
+                        if not output_b:
+                            break
+
+                        output_noescape_b = delete_ansi_escape_bytes(output_b)
+                        output_file.write(output_noescape_b)
+                        if not output_stream.isatty():
+                            forward_b = output_noescape_b
+                        else:
+                            forward_b = output_b
+
+                        if is_progression_processing_enabled and is_progression_bytes(output_b):
+                            print_progression(output_b.decode('utf-8', errors='replace'))
+                            is_progression_last_line = True
+                        else:
+                            if is_progression_last_line:
+                                write_stdout_after_progression_line()
+                                is_progression_last_line = False
+                            if self.convert_output:
+                                write_text_maybe_illencoded(
+                                    output_converter,
+                                    forward_b.decode('utf-8', errors='replace'),
+                                )
+                            else:
+                                write_stdout_bytes(forward_b)
         except (OSError, RuntimeError) as e:
             yellow_print(
                 "WARNING: The exception {} was raised and we can't capture all your {} and "

tools/test_idf_py/test_read_and_write_stream_unicode.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python
+#
+# SPDX-FileCopyrightText: 2026 Espressif Systems (Shanghai) CO LTD
+#
+# SPDX-License-Identifier: Apache-2.0
+"""Regression for UTF-8 build lines on narrow Windows stdio (pipe + cp1252)."""
+
+from __future__ import annotations
+
+import asyncio
+import io
+import os
+import sys
+import tempfile
+import unittest
+
+try:
+    os.environ['IDF_PATH']  # noqa: B018
+except KeyError:
+    print('This test needs IDF_PATH set (export.bat / export.sh).', file=sys.stderr)
+    sys.exit(1)
+
+CWD = os.path.dirname(__file__)
+try:
+    from idf_py_actions.tools import RunTool
+except ImportError:
+    sys.path.append(os.path.join(CWD, '..'))
+    from idf_py_actions.tools import RunTool
+
+
+class TestReadAndWriteStreamUnicode(unittest.TestCase):
+    def test_utf8_line_survives_cp1252_text_wrapper(self) -> None:
+        """Subprocess UTF-8 must reach the binary layer without TextIO cp1252 re-encode (esp-hosted-mcu#176)."""
+        # U+1F600 is not representable in cp1252; \u2014 (em dash) is 0x97 in cp1252 and would not fail strict encode.
+        exotic = '\U0001f600'
+        line_bytes = f'ninja: unicode {exotic} done\n'.encode()
+
+        narrow = io.TextIOWrapper(io.BytesIO(), encoding='cp1252', errors='strict')
+        with self.assertRaises(UnicodeEncodeError):
+            narrow.write(exotic)
+
+        raw_out = io.BytesIO()
+        text_out = io.TextIOWrapper(
+            raw_out,
+            encoding='cp1252',
+            errors='strict',
+            newline='',
+            line_buffering=True,
+        )
+        self.assertFalse(text_out.isatty())
+
+        captured: dict[str, bytes] = {}
+
+        async def run() -> None:
+            reader = asyncio.StreamReader()
+            reader.feed_data(line_bytes)
+            reader.feed_eof()
+            with tempfile.TemporaryDirectory() as tmp:
+                log_path = os.path.join(tmp, 'cap.log')
+                tool = RunTool(
+                    'unittest_build',
+                    ['ninja', 'all'],
+                    tmp,
+                    build_dir=tmp,
+                    hints=False,
+                    interactive=False,
+                    convert_output=False,
+                )
+                await tool.read_and_write_stream(reader, log_path, text_out)
+                with open(log_path, 'rb') as f:
+                    captured['log'] = f.read()
+
+        asyncio.run(run())
+
+        text_out.flush()
+        emitted = raw_out.getvalue()
+        self.assertIn(exotic.encode('utf-8'), emitted)
+        self.assertIn(exotic.encode('utf-8'), captured['log'])
+        self.assertTrue(captured['log'].startswith(b'Command:'))