Spool native warnings without pipes during EP compilation

src/winml/modelkit/utils/native_stderr.py

@@ -10,7 +10,7 @@
 
 * ``suppress_native_stderr``  - discard to devnull  (startup noise)
 * ``capture_native_stderr``   - capture via pipe and re-log  (compilation output)
-* ``suppress_native_warnings`` - hide warning lines, replay everything else
+* ``suppress_native_warnings`` - spool to disk, hide warnings, replay other lines
 
 The first two are no-ops on non-Windows. ``suppress_native_warnings`` works via
 fd 2 on all platforms and also keeps the Win32 stderr handle in sync on Windows.
@@ -22,8 +22,9 @@
 import os
 import re
 import sys
+import tempfile
 import threading
-from contextlib import contextmanager
+from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
@@ -198,19 +199,25 @@ def suppress_native_warnings(
     ``[E:...]``. This process-wide fd redirect is opt-in; CLI entry points pass
     ``enabled=True`` only around native-heavy work. ``-v`` / ``-vv`` or
     ``WINMLCLI_SHOW_ALL_WARNINGS=1`` leaves stderr untouched.
+
+    Output is spooled to a temporary file and replayed only after fd 2 is
+    restored. A pipe is deliberately not used here: some native execution-
+    provider compilers change behavior or hang when stderr is a pipe, even when
+    that pipe is drained concurrently. A file also bounds Python memory usage
+    without imposing a finite producer buffer.
     """
     if not enabled or _show_native_warnings_requested():
         yield
         return
 
-    read_fd: int | None = None
-    write_fd: int | None = None
     old_fd: int | None = None
+    capture_stack = ExitStack()
     try:
-        read_fd, write_fd = os.pipe()
+        capture = capture_stack.enter_context(
+            tempfile.TemporaryFile(mode="w+b")  # noqa: SIM115 - owned by ExitStack
+        )
     except OSError:
-        _close_fd(read_fd)
-        _close_fd(write_fd)
+        capture_stack.close()
         logger.debug(
             "Native warning suppression setup failed; leaving stderr unchanged",
             exc_info=True,
@@ -223,49 +230,45 @@ def suppress_native_warnings(
         try:
             old_fd = os.dup(2)
         except OSError:
-            _close_fd(read_fd)
-            _close_fd(write_fd)
+            capture_stack.close()
             logger.debug(
                 "Native warning suppression setup failed; leaving stderr unchanged",
                 exc_info=True,
             )
             redirect_failed = True
         if not redirect_failed:
             assert old_fd is not None
-            reader = threading.Thread(
-                target=_drain_filtered_native_stderr,
-                args=(read_fd, old_fd, preserve_unclassified),
-                name="suppress-native-warnings",
-                daemon=True,
-            )
             try:
-                os.dup2(write_fd, 2)
+                os.dup2(capture.fileno(), 2)
             except OSError:
-                _close_fd(read_fd)
-                _close_fd(write_fd)
+                capture_stack.close()
                 _close_fd(old_fd)
                 logger.debug(
                     "Native warning suppression redirect failed; leaving stderr unchanged",
                     exc_info=True,
                 )
                 redirect_failed = True
             if not redirect_failed:
-                _close_fd(write_fd)
                 _set_win32_std_handle_to_current_fd(2)
-                reader.start()
                 try:
                     yield
                 finally:
                     _restore_redirected_fd(2, old_fd)
                     _set_win32_std_handle_to_current_fd(2)
                     _refresh_click_windows_console_stream(2)
-                    reader.join(timeout=_NATIVE_READER_JOIN_TIMEOUT_SECONDS)
-                    if reader.is_alive():
+                    try:
+                        capture.flush()
+                        capture.seek(0)
+                        for line in capture:
+                            if _should_preserve_native_line(line, preserve_unclassified):
+                                _write_all(old_fd, line)
+                    except OSError:
                         logger.debug(
-                            "Native warning suppression reader did not finish after stderr restore"
+                            "Could not replay filtered native stderr",
+                            exc_info=True,
                         )
-                    else:
-                        _close_fd(old_fd)
+                    capture_stack.close()
+                    _close_fd(old_fd)
     if redirect_failed:
         yield
 
@@ -307,32 +310,6 @@ def _restore_redirected_fd(fd: int, old_fd: int | None) -> None:
         )
 
 
-def _drain_filtered_native_stderr(
-    read_fd: int,
-    target_fd: int,
-    preserve_unclassified: bool,
-) -> None:
-    pending = b""
-    try:
-        while chunk := os.read(read_fd, 4096):
-            pending += chunk
-            while b"\n" in pending:
-                line, pending = pending.split(b"\n", 1)
-                _write_non_warning_line(target_fd, line + b"\n", preserve_unclassified)
-        if pending:
-            _write_non_warning_line(target_fd, pending, preserve_unclassified)
-    except OSError:
-        # fd redirection is best-effort cleanup; restore path handles usability.
-        pass
-    finally:
-        os.close(read_fd)
-
-
-def _write_non_warning_line(fd: int, line: bytes, preserve_unclassified: bool) -> None:
-    if _should_preserve_native_line(line, preserve_unclassified):
-        _write_all(fd, line)
-
-
 def _should_preserve_native_line(line: bytes, preserve_unclassified: bool) -> bool:
     match = _NATIVE_SEVERITY_TOKEN_RE.search(line)
     if match is not None:

tests/unit/utils/test_native_stderr.py

@@ -243,6 +243,23 @@ def test_filters_native_warning_lines_and_preserves_errors(self, monkeypatch, ca
         assert "useful error" in stderr
         assert "plain diagnostic" in stderr
 
+    def test_uses_file_backed_capture_instead_of_pipe(self, monkeypatch):
+        monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False)
+        logging.getLogger().setLevel(logging.WARNING)
+        pipe_called = False
+
+        def fail_pipe() -> tuple[int, int]:
+            nonlocal pipe_called
+            pipe_called = True
+            raise AssertionError("warning suppression must not create a pipe")
+
+        monkeypatch.setattr(native_stderr_module.os, "pipe", fail_pipe)
+
+        with native_stderr_module.suppress_native_warnings(enabled=True):
+            os.write(2, b"2026 [W:custom-native:, file.cc:1 WarningFunc] hidden\n")
+
+        assert pipe_called is False
+
     def test_filters_native_prefix_info_without_dropping_python_stderr(self, monkeypatch, capfd):
         monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False)
         logging.getLogger().setLevel(logging.WARNING)
@@ -299,52 +316,36 @@ def test_show_all_warnings_env_leaves_native_warnings_visible(self, monkeypatch,
 
         assert "env warning" in capfd.readouterr().err
 
-    def test_pipe_setup_failure_does_not_abort_wrapped_code(self, monkeypatch):
+    def test_capture_setup_failure_does_not_abort_wrapped_code(self, monkeypatch):
         monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False)
         logging.getLogger().setLevel(logging.WARNING)
 
-        def fail_pipe() -> tuple[int, int]:
+        def fail_capture(*args: object, **kwargs: object):
             raise OSError(1, "Incorrect function")
 
-        monkeypatch.setattr(native_stderr_module.os, "pipe", fail_pipe)
+        monkeypatch.setattr(native_stderr_module.tempfile, "TemporaryFile", fail_capture)
 
         ran = False
         with native_stderr_module.suppress_native_warnings(enabled=True):
             ran = True
 
         assert ran
 
-    def test_restore_failure_closes_redirected_fd_and_bounds_reader_join(self, monkeypatch):
+    def test_restore_failure_closes_redirected_fd(self, monkeypatch):
         monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False)
         logging.getLogger().setLevel(logging.WARNING)
         closed: list[int] = []
-        join_timeouts: list[float | None] = []
         dup2_calls = 0
 
-        class FakeThread:
-            def __init__(self, *args: object, **kwargs: object) -> None:
-                pass
-
-            def start(self) -> None:
-                pass
-
-            def join(self, timeout: float | None = None) -> None:
-                join_timeouts.append(timeout)
-
-            def is_alive(self) -> bool:
-                return False
-
         def fake_dup2(src: int, dst: int) -> None:
             nonlocal dup2_calls
             dup2_calls += 1
             if dup2_calls == 2:
                 raise OSError(1, "restore failed")
 
-        monkeypatch.setattr(native_stderr_module.os, "pipe", lambda: (10, 11))
         monkeypatch.setattr(native_stderr_module.os, "dup", lambda fd: 12)
         monkeypatch.setattr(native_stderr_module.os, "dup2", fake_dup2)
         monkeypatch.setattr(native_stderr_module.os, "close", lambda fd: closed.append(fd))
-        monkeypatch.setattr(native_stderr_module.threading, "Thread", FakeThread)
         monkeypatch.setattr(
             native_stderr_module,
             "_set_win32_std_handle_to_current_fd",
@@ -360,48 +361,7 @@ def fake_dup2(src: int, dst: int) -> None:
             pass
 
         assert 2 in closed
-        assert join_timeouts == [native_stderr_module._NATIVE_READER_JOIN_TIMEOUT_SECONDS]
-
-    def test_reader_owned_old_fd_stays_open_when_reader_outlives_join(self, monkeypatch):
-        monkeypatch.delenv("WINMLCLI_SHOW_ALL_WARNINGS", raising=False)
-        logging.getLogger().setLevel(logging.WARNING)
-        closed: list[int] = []
-        join_timeouts: list[float | None] = []
-
-        class FakeThread:
-            def __init__(self, *args: object, **kwargs: object) -> None:
-                pass
-
-            def start(self) -> None:
-                pass
-
-            def join(self, timeout: float | None = None) -> None:
-                join_timeouts.append(timeout)
-
-            def is_alive(self) -> bool:
-                return True
-
-        monkeypatch.setattr(native_stderr_module.os, "pipe", lambda: (10, 11))
-        monkeypatch.setattr(native_stderr_module.os, "dup", lambda fd: 12)
-        monkeypatch.setattr(native_stderr_module.os, "dup2", lambda src, dst: None)
-        monkeypatch.setattr(native_stderr_module.os, "close", lambda fd: closed.append(fd))
-        monkeypatch.setattr(native_stderr_module.threading, "Thread", FakeThread)
-        monkeypatch.setattr(
-            native_stderr_module,
-            "_set_win32_std_handle_to_current_fd",
-            lambda fd: None,
-        )
-        monkeypatch.setattr(
-            native_stderr_module,
-            "_refresh_click_windows_console_stream",
-            lambda fd, handle=None: None,
-        )
-
-        with native_stderr_module.suppress_native_warnings(enabled=True):
-            pass
-
-        assert join_timeouts == [native_stderr_module._NATIVE_READER_JOIN_TIMEOUT_SECONDS]
-        assert 12 not in closed
+        assert 12 in closed
 
     @pytest.mark.skipif(sys.platform != "win32", reason="Win32 only")
     def test_win32_std_handle_sync_failure_does_not_abort_wrapped_code(self, monkeypatch):