serve: console output goes through a bounded queue, so a blocked console cannot hold a request (#4479, #2481)

print() in the server blocks when the console stops taking output (a Windows console in selection mode, a paused
terminal, a launcher that stopped reading the pipe), in whatever thread called it - including those that own the
engine's turn, a cancel or an admission. While serve main() runs, stdout and stderr (when they are the process's own)
are replaced by serve/nonblocking_out.NonBlockingOut: write() queues and returns, one daemon thread writes to the real
stream, the oldest text is dropped (and counted) when the queue is over 1 MiB / 4000 chunks. Restored and drained at the
end of main(). Protocol output (engine pipe, HTTP) is not touched. STRATA_BLOCKING_CONSOLE=1 turns it off.
No visible text is added; a drop is counted, not announced.

Tests: a plain print to an unread pipe blocks (control); the wrapper never blocks, keeps the newest lines in order, and
drains. Approach credit: the proposal in #4479 (one process-owned output consumer).

serve/nonblocking_out.py

@@ -0,0 +1,123 @@
+"""#4479 / #2481: the server's console output must not hold a request.
+
+A console that stops taking output (a Windows console in QuickEdit/selection mode, a terminal that is paused, a launcher
+that stopped reading the server's pipe) blocks `print()` in whatever thread called it - and some of those threads own the
+engine's turn, a cancel, or an admission.  `NonBlockingOut` stands in for sys.stdout / sys.stderr while the server runs:
+write() puts the text on a bounded queue and returns; one daemon thread writes it to the real stream, and may block there
+as long as it likes.  When the queue is full the OLDEST text is dropped (the newest lines matter most) and counted in
+`dropped_chunks` / `dropped_bytes`; nothing about a request waits for the console.
+
+Only diagnostics go through here.  Protocol output (the engine's pipe, the HTTP replies) does not.
+"""
+from __future__ import annotations
+
+import collections
+import contextlib
+import sys
+import threading
+
+
+class NonBlockingOut:
+    def __init__(self, real, max_bytes: int = 1 << 20, max_chunks: int = 4000):
+        self._real = real
+        self.max_bytes, self.max_chunks = max_bytes, max_chunks
+        self._q: collections.deque[str] = collections.deque()
+        self._bytes = 0
+        self._cv = threading.Condition()
+        self._closing = False
+        self._idle = True
+        self.dropped_chunks = 0
+        self.dropped_bytes = 0
+        self._thread = threading.Thread(target=self._run, name="strata-console", daemon=True)
+        self._thread.start()
+
+    # ---- what the program calls: never blocks on the console
+    def write(self, text) -> int:
+        if not isinstance(text, str):
+            text = str(text)
+        if not text:
+            return 0
+        with self._cv:
+            if self._closing:
+                return len(text)
+            self._q.append(text)
+            self._bytes += len(text)
+            while self._q and (self._bytes > self.max_bytes or len(self._q) > self.max_chunks):
+                old = self._q.popleft()
+                self._bytes -= len(old)
+                self.dropped_chunks += 1
+                self.dropped_bytes += len(old)
+            self._cv.notify()
+        return len(text)
+
+    def flush(self) -> None:
+        return None                                    # the writer thread flushes the real stream
+
+    def writable(self) -> bool:
+        return True
+
+    def isatty(self) -> bool:
+        try:
+            return bool(self._real.isatty())
+        except (OSError, ValueError, AttributeError):
+            return False
+
+    def __getattr__(self, name):                        # encoding, errors, fileno, buffer ...: the real stream's
+        return getattr(self._real, name)
+
+    # ---- the one thread that may block
+    def _run(self) -> None:
+        while True:
+            with self._cv:
+                while not self._q and not self._closing:
+                    self._idle = True
+                    self._cv.notify_all()
+                    self._cv.wait()
+                if not self._q:
+                    self._idle = True
+                    self._cv.notify_all()
+                    return
+                self._idle = False
+                text = self._q.popleft()
+                self._bytes -= len(text)
+            try:
+                self._real.write(text)
+                self._real.flush()
+            except (OSError, ValueError, UnicodeError):
+                pass                                    # a closed or broken console: the text is lost, the server is not held
+
+    def drain(self, timeout: float = 2.0) -> bool:
+        """Wait (bounded) until everything queued has been handed to the real stream."""
+        import time
+        end = time.monotonic() + timeout
+        with self._cv:
+            while (self._q or not self._idle) and time.monotonic() < end:
+                self._cv.wait(timeout=max(0.01, end - time.monotonic()))
+            return not self._q and self._idle
+
+    def close(self, timeout: float = 2.0) -> None:
+        self.drain(timeout)
+        with self._cv:
+            self._closing = True
+            self._cv.notify_all()
+
+
+@contextlib.contextmanager
+def decoupled_console():
+    """While the server runs: stdout and stderr go through NonBlockingOut, when they are the process's own streams (a
+    test runner's capture, or a stream somebody else replaced, is left alone).  Restored, and drained for a moment, at
+    the end."""
+    installed = []
+    for name in ("stdout", "stderr"):
+        cur = getattr(sys, name, None)
+        if cur is not None and cur is getattr(sys, "__" + name + "__", None):
+            wrapper = NonBlockingOut(cur)
+            setattr(sys, name, wrapper)
+            installed.append((name, cur, wrapper))
+    try:
+        yield [w for _, _, w in installed]
+    finally:
+        for name, orig, wrapper in installed:
+            wrapper.close()
+            if getattr(sys, name, None) is wrapper:
+                setattr(sys, name, orig)

serve/server.py

@@ -9680,6 +9680,17 @@ def tool_call_recovery_setting(cfg: dict) -> bool:
 
 
 def main(argv=None, *, lifecycle_event=None, interrupt_exit_code: int = 0) -> int:
+    """The server.  Its console output goes through a bounded queue (serve/nonblocking_out.py, #4479 / #2481): a console
+    that stops taking output must not hold a request's turn, a cancel or an admission.  STRATA_BLOCKING_CONSOLE=1 turns
+    it off (testers)."""
+    if os.environ.get("STRATA_BLOCKING_CONSOLE") == "1":
+        return _main(argv, lifecycle_event=lifecycle_event, interrupt_exit_code=interrupt_exit_code)
+    from serve.nonblocking_out import decoupled_console
+    with decoupled_console():
+        return _main(argv, lifecycle_event=lifecycle_event, interrupt_exit_code=interrupt_exit_code)
+
+
+def _main(argv=None, *, lifecycle_event=None, interrupt_exit_code: int = 0) -> int:
     try:                                               # files made by a run as administrator still belong to the user
         from strata_control import private_files as _pf
         _pf.adopt_user_defaults()

serve/test_serve_robustness.py

@@ -206,6 +206,67 @@ class StalledClient(unittest.TestCase):
             httpd.server_close()
 
 
+class BlockedConsole(unittest.TestCase):
+    """#4479 / #2481: a console that stops taking output must not hold the thread that prints."""
+
+    def pipe(self):
+        import os
+        r, w = os.pipe()
+        self.addCleanup(lambda: [os.close(x) for x in (r,) if True])
+        return r, os.fdopen(w, "w", encoding="utf-8", buffering=1)
+
+    def test_a_print_to_a_full_pipe_blocks_without_the_wrapper(self):
+        """The control: this is the dependency the request lifecycle had on the console."""
+        r, w = self.pipe()
+        done = threading.Event()
+
+        def spam():
+            for i in range(20000):
+                w.write(f"[strata] progress line {i} " + "x" * 40 + "\n")
+                w.flush()
+            done.set()
+        threading.Thread(target=spam, daemon=True).start()
+        self.assertFalse(done.wait(1.0), "a pipe nobody reads must block a plain print")
+
+    def test_the_wrapper_never_blocks_and_keeps_the_newest_lines(self):
+        import os
+        from serve.nonblocking_out import NonBlockingOut
+        r, w = self.pipe()
+        out = NonBlockingOut(w, max_bytes=64 * 1024, max_chunks=500)
+        done = threading.Event()
+
+        def spam():
+            for i in range(20000):
+                print(f"[strata] progress line {i} " + "x" * 40, file=out, flush=True)
+            done.set()
+        threading.Thread(target=spam, daemon=True).start()
+        self.assertTrue(done.wait(5.0), "printing into a console nobody reads must not block")
+        self.assertGreater(out.dropped_chunks, 0)
+        self.assertLessEqual(len(out._q), 500)
+        # the console wakes up: everything that was kept arrives, in order, ending with the newest line
+        data = b""
+        end = __import__("time").monotonic() + 10
+        while __import__("time").monotonic() < end and b"progress line 19999" not in data:
+            data += os.read(r, 65536)
+        text = data.decode()
+        self.assertIn("progress line 19999 ", text)
+        seen = [int(l.split()[3]) for l in text.splitlines() if l.startswith("[strata] progress line")]
+        self.assertEqual(seen, sorted(seen), "order is kept")
+        out.close(1.0)
+
+    def test_the_console_is_restored_and_drained(self):
+        import io
+        from serve.nonblocking_out import NonBlockingOut
+        real = io.StringIO()
+        out = NonBlockingOut(real)
+        out.write("hello\n")
+        out.write("world\n")
+        self.assertTrue(out.drain(2.0))
+        self.assertEqual(real.getvalue(), "hello\nworld\n")
+        out.close()
+        self.assertEqual(out.write("late\n"), 5)                     # a late write after close is not an error
+
+
 class EngineOutputDecoding(FakeEngines, unittest.TestCase):
     """#5044: a byte that is not UTF-8 on the engine's stdout (Windows can print one) must not end the reader."""