fix: implement non-blocking pipe reads on Windows to prevent agent freeze

Author: mpfaffenberger

code_puppy/tools/command_runner.py

@@ -1,3 +1,4 @@
+import ctypes
 import os
 import select
 import signal
@@ -38,6 +39,60 @@ def _truncate_line(line: str) -> str:
     return line
 
 
+# Windows-specific: Check if pipe has data available without blocking
+# This is needed because select() doesn't work on pipes on Windows
+if sys.platform.startswith("win"):
+    import msvcrt
+
+    # Load kernel32 for PeekNamedPipe
+    _kernel32 = ctypes.windll.kernel32
+
+    def _win32_pipe_has_data(pipe) -> bool:
+        """Check if a Windows pipe has data available without blocking.
+
+        Uses PeekNamedPipe from kernel32.dll to check if there's data
+        in the pipe buffer without actually reading it.
+
+        Args:
+            pipe: A file object with a fileno() method (e.g., process.stdout)
+
+        Returns:
+            True if data is available, False otherwise (including on error)
+        """
+        try:
+            # Get the Windows handle from the file descriptor
+            handle = msvcrt.get_osfhandle(pipe.fileno())
+
+            # PeekNamedPipe parameters:
+            # - hNamedPipe: handle to the pipe
+            # - lpBuffer: buffer to receive data (NULL = don't read)
+            # - nBufferSize: size of buffer (0 = don't read)
+            # - lpBytesRead: receives bytes read (NULL)
+            # - lpTotalBytesAvail: receives total bytes available
+            # - lpBytesLeftThisMessage: receives bytes left (NULL)
+            bytes_available = ctypes.c_ulong(0)
+
+            result = _kernel32.PeekNamedPipe(
+                handle,
+                None,  # Don't read data
+                0,  # Buffer size 0
+                None,  # Don't care about bytes read
+                ctypes.byref(bytes_available),  # Get bytes available
+                None,  # Don't care about bytes left in message
+            )
+
+            if result:
+                return bytes_available.value > 0
+            return False
+        except (ValueError, OSError, ctypes.ArgumentError):
+            # Handle closed, invalid, or other errors
+            return False
+else:
+    # POSIX stub - not used, but keeps the code clean
+    def _win32_pipe_has_data(pipe) -> bool:
+        return False
+
+
 _AWAITING_USER_INPUT = False
 
 _CONFIRMATION_LOCK = threading.Lock()
@@ -468,17 +523,35 @@ def read_stdout():
 
                 # Use select to check if data is available (with timeout)
                 if sys.platform.startswith("win"):
-                    # Windows doesn't support select on pipes, use a different approach
-                    # Just try to read with a check on the stop event
+                    # Windows doesn't support select on pipes
+                    # Use PeekNamedPipe via _win32_pipe_has_data() to check
+                    # if data is available without blocking
                     try:
-                        line = process.stdout.readline()
-                        if not line:  # EOF
-                            break
-                        line = line.rstrip("\n\r")
-                        line = _truncate_line(line)
-                        stdout_lines.append(line)
-                        emit_shell_line(line, stream="stdout")
-                        last_output_time[0] = time.time()
+                        if _win32_pipe_has_data(process.stdout):
+                            line = process.stdout.readline()
+                            if not line:  # EOF
+                                break
+                            line = line.rstrip("\n\r")
+                            line = _truncate_line(line)
+                            stdout_lines.append(line)
+                            emit_shell_line(line, stream="stdout")
+                            last_output_time[0] = time.time()
+                        else:
+                            # No data available, check if process has exited
+                            if process.poll() is not None:
+                                # Process exited, do one final drain
+                                try:
+                                    remaining = process.stdout.read()
+                                    if remaining:
+                                        for line in remaining.splitlines():
+                                            line = _truncate_line(line)
+                                            stdout_lines.append(line)
+                                            emit_shell_line(line, stream="stdout")
+                                except (ValueError, OSError):
+                                    pass
+                                break
+                            # Sleep briefly to avoid busy-waiting (100ms like POSIX)
+                            time.sleep(0.1)
                     except (ValueError, OSError):
                         break
                 else:
@@ -516,15 +589,35 @@ def read_stderr():
                     break
 
                 if sys.platform.startswith("win"):
+                    # Windows doesn't support select on pipes
+                    # Use PeekNamedPipe via _win32_pipe_has_data() to check
+                    # if data is available without blocking
                     try:
-                        line = process.stderr.readline()
-                        if not line:  # EOF
-                            break
-                        line = line.rstrip("\n\r")
-                        line = _truncate_line(line)
-                        stderr_lines.append(line)
-                        emit_shell_line(line, stream="stderr")
-                        last_output_time[0] = time.time()
+                        if _win32_pipe_has_data(process.stderr):
+                            line = process.stderr.readline()
+                            if not line:  # EOF
+                                break
+                            line = line.rstrip("\n\r")
+                            line = _truncate_line(line)
+                            stderr_lines.append(line)
+                            emit_shell_line(line, stream="stderr")
+                            last_output_time[0] = time.time()
+                        else:
+                            # No data available, check if process has exited
+                            if process.poll() is not None:
+                                # Process exited, do one final drain
+                                try:
+                                    remaining = process.stderr.read()
+                                    if remaining:
+                                        for line in remaining.splitlines():
+                                            line = _truncate_line(line)
+                                            stderr_lines.append(line)
+                                            emit_shell_line(line, stream="stderr")
+                                except (ValueError, OSError):
+                                    pass
+                                break
+                            # Sleep briefly to avoid busy-waiting (100ms like POSIX)
+                            time.sleep(0.1)
                     except (ValueError, OSError):
                         break
                 else:

tests/test_windows_pipe.py

@@ -0,0 +1,107 @@
+"""Test Windows pipe non-blocking read functionality."""
+import subprocess
+import sys
+import threading
+import time
+
+import pytest
+
+from code_puppy.tools.command_runner import _win32_pipe_has_data
+
+
+@pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows only")
+class TestWin32PipeHasData:
+    """Tests for the _win32_pipe_has_data function."""
+
+    def test_returns_true_when_data_available(self):
+        """Test that function returns True when pipe has data."""
+        proc = subprocess.Popen(
+            "echo hello",
+            shell=True,
+            stdout=subprocess.PIPE,
+            text=True,
+        )
+        proc.wait()
+        time.sleep(0.1)  # Give buffer time to fill
+        
+        assert _win32_pipe_has_data(proc.stdout) is True
+        proc.stdout.close()
+
+    def test_returns_false_when_no_data(self):
+        """Test that function returns False when pipe has no data yet."""
+        # Use a command that takes time to produce output
+        proc = subprocess.Popen(
+            "ping -n 3 127.0.0.1 >nul && echo done",
+            shell=True,
+            stdout=subprocess.PIPE,
+            text=True,
+        )
+        
+        # Check immediately - should have no data
+        result = _win32_pipe_has_data(proc.stdout)
+        proc.kill()
+        proc.wait()
+        
+        # Should be False (no output yet)
+        assert result is False
+
+    def test_reader_loop_can_be_stopped(self):
+        """Test that a reader loop using _win32_pipe_has_data can be stopped via event."""
+        # This is the key test - simulates the frozen agent scenario
+        proc = subprocess.Popen(
+            "ping -n 1000 127.0.0.1 >nul",  # Long-running, no stdout output
+            shell=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+        )
+        
+        stop_event = threading.Event()
+        loop_exited = threading.Event()
+        
+        def reader_loop():
+            iterations = 0
+            while not stop_event.is_set():
+                iterations += 1
+                if _win32_pipe_has_data(proc.stdout):
+                    proc.stdout.readline()
+                else:
+                    time.sleep(0.05)  # Brief sleep when no data
+                
+                if iterations > 100:  # Safety limit
+                    break
+            loop_exited.set()
+        
+        # Start reader thread
+        reader_thread = threading.Thread(target=reader_loop, daemon=True)
+        reader_thread.start()
+        
+        # Wait briefly, then signal stop
+        time.sleep(0.2)
+        stop_event.set()
+        
+        # Thread should exit quickly
+        reader_thread.join(timeout=1.0)
+        
+        # Cleanup
+        proc.kill()
+        proc.wait()
+        
+        # The key assertion: thread should have exited
+        assert loop_exited.is_set(), "Reader loop should have exited when stop event was set"
+        assert not reader_thread.is_alive(), "Reader thread should not be alive"
+
+    def test_handles_closed_pipe(self):
+        """Test that function handles closed pipes gracefully."""
+        proc = subprocess.Popen(
+            "echo test",
+            shell=True,
+            stdout=subprocess.PIPE,
+            text=True,
+        )
+        proc.wait()
+        proc.stdout.close()
+        
+        # Should return False, not raise exception
+        result = _win32_pipe_has_data(proc.stdout)
+        assert result is False