Add redirect_stdout_to_stderr() utility for native libraries

llm/utils.py

@@ -1,9 +1,11 @@
+import contextlib
 import hashlib
 import itertools
 import json
 import os
 import pathlib
 import re
+import sys
 import textwrap
 import threading
 import time
@@ -20,6 +22,29 @@
 }
 
 
+@contextlib.contextmanager
+def redirect_stdout_to_stderr():
+    """Temporarily redirect OS-level stdout (file descriptor 1) to stderr.
+
+    Some native libraries write log messages straight to file descriptor 1,
+    bypassing both ``sys.stdout`` and ``contextlib.redirect_stdout``. For
+    example, llama.cpp (used by the ``llm-llama-cpp`` plugin) and gpt4all
+    emit diagnostics this way. Wrapping those native calls in this context
+    manager routes that noise to stderr, leaving stdout free for model output.
+    """
+    stdout = sys.__stdout__ or sys.stdout
+    stderr = sys.__stderr__ or sys.stderr
+    stdout_fd = stdout.fileno()
+    stderr_fd = stderr.fileno()
+    saved_stdout_fd = os.dup(stdout_fd)
+    try:
+        os.dup2(stderr_fd, stdout_fd)
+        yield
+    finally:
+        os.dup2(saved_stdout_fd, stdout_fd)
+        os.close(saved_stdout_fd)
+
+
 class Fragment(str):
     def __new__(cls, content, *args, **kwargs):
         # For immutable classes like str, __new__ creates the string object

tests/test_utils.py

@@ -1,4 +1,5 @@
 import json
+import os
 
 import pytest
 
@@ -8,6 +9,7 @@
     instantiate_from_spec,
     maybe_fenced_code,
     monotonic_ulid,
+    redirect_stdout_to_stderr,
     schema_dsl,
     simplify_usage_dict,
     truncate_string,
@@ -558,3 +560,34 @@ class Tool6(Toolbox):
         pass
 
     assert Tool6()._config == {}
+
+
+def test_redirect_stdout_to_stderr():
+    # Redirect the real stdout and stderr file descriptors to pipes so we can
+    # observe OS-level writes that bypass Python's sys.stdout/sys.stderr.
+    stdout_read, stdout_write = os.pipe()
+    stderr_read, stderr_write = os.pipe()
+    original_stdout = os.dup(1)
+    original_stderr = os.dup(2)
+    os.dup2(stdout_write, 1)
+    os.dup2(stderr_write, 2)
+    try:
+        with redirect_stdout_to_stderr():
+            os.write(1, b"native-noise")
+        os.write(1, b"clean-output")
+    finally:
+        os.dup2(original_stdout, 1)
+        os.dup2(original_stderr, 2)
+        os.close(original_stdout)
+        os.close(original_stderr)
+        os.close(stdout_write)
+        os.close(stderr_write)
+
+    stdout_data = os.read(stdout_read, 1024)
+    stderr_data = os.read(stderr_read, 1024)
+    os.close(stdout_read)
+    os.close(stderr_read)
+
+    assert b"native-noise" not in stdout_data
+    assert b"clean-output" in stdout_data
+    assert b"native-noise" in stderr_data