fix: contain exceptions in _pump_stream to prevent thread death

Wrap the readline loop in _pump_stream with two layers of exception
handling: callback exceptions (on_output_line) are caught per-line so
draining never stops, and ValueError/OSError from readline (concurrent
pipe close) cause a clean exit instead of an unhandled thread crash.

Without this, a single raising callback would kill the daemon thread
silently, causing the subprocess to block on a full pipe buffer and
either hang forever or spuriously time out.

Task: critical-02-pump-stream-exception-handling.md

Co-authored-by: Ralphify <noreply@ralphify.co>

Author: 2 people

src/ralphify/_agent.py

@@ -349,12 +349,27 @@ def _pump_stream(
 
     Runs on a background thread so stdout and stderr can be drained
     concurrently without deadlocking the child subprocess.
+
+    Exception handling:
+
+    - **Callback exceptions** are caught per-line so that a raising
+      callback never kills the drain loop.  The line is still buffered.
+    - **``ValueError`` / ``OSError``** from ``readline`` (e.g. the pipe
+      was closed concurrently) cause a clean exit so ``join()`` returns.
     """
-    for line in iter(stream.readline, ""):
-        if buffer is not None:
-            buffer.append(line)
-        if on_output_line is not None:
-            on_output_line(line.rstrip("\r\n"), stream_name)
+    try:
+        for line in iter(stream.readline, ""):
+            if buffer is not None:
+                buffer.append(line)
+            if on_output_line is not None:
+                try:
+                    on_output_line(line.rstrip("\r\n"), stream_name)
+                except Exception:
+                    # Callback is best-effort; draining must not stop.
+                    pass
+    except (ValueError, OSError):
+        # Pipe closed concurrently — exit cleanly so join() returns.
+        pass
 
 
 def _start_pump_thread(

tasks/done/critical-02-pump-stream-exception-handling.md

@@ -0,0 +1,81 @@
+# Critical 02 — `_pump_stream` exception containment
+
+**Original findings:** C2 (raising callback wedges subprocess) + M9 (`ValueError` on closed file)
+**Severity:** Critical — a single raising callback hangs or spuriously times out the agent
+**Files:** `src/ralphify/_agent.py`
+
+## Problem
+
+`_pump_stream` is the reader-thread body used by both the blocking and streaming paths to drain stdout/stderr line-by-line. Its entire body is:
+
+```python
+# src/ralphify/_agent.py:320-334
+def _pump_stream(
+    stream: IO[str],
+    buffer: list[str],
+    stream_name: OutputStream,
+    on_output_line: Callable[[str, OutputStream], None] | None,
+) -> None:
+    for line in iter(stream.readline, ""):
+        buffer.append(line)
+        if on_output_line is not None:
+            on_output_line(line.rstrip("\r\n"), stream_name)
+```
+
+Zero exception handling. Two failure modes:
+
+1. **Callback raises** (`on_output_line` is a lambda that goes through emitter → event construction → Rich print → tty write). Any raise anywhere in that chain — queue-full in a `QueueEmitter`, Rich markup bug, `UnicodeEncodeError` on a weird line, buggy user handler in the Python API — propagates out of the daemon thread. The thread dies silently (daemon thread exceptions are not surfaced), the pipe stops draining, the child fills its OS pipe buffer, and blocks.
+
+2. **`readline` raises** `ValueError("I/O operation on closed file")` if the main thread closes the pipe concurrently (GC, explicit cleanup, `with` block exit). Thread dies, log gets truncated, no warning.
+
+## Why it matters
+
+After the thread dies:
+
+- `proc.wait(timeout=timeout)` either hangs forever (when `timeout is None`) or spuriously times out (the user is told the agent timed out when it actually finished fine, but the output never reached Python).
+- The logged output is silently truncated — tail of the iteration is lost, the user has no way to know.
+- On CI where the agent produces JSON events consumed by downstream tooling, a single malformed line permanently breaks that run.
+
+Because the feature is on by default in any TTY, this is a live footgun for anyone with a Python-API integration that raises from a handler.
+
+## Fix direction
+
+Wrap the loop so (a) callback exceptions never kill draining, (b) `readline` exceptions let the thread exit cleanly:
+
+```python
+def _pump_stream(stream, buffer, stream_name, on_output_line):
+    try:
+        for line in iter(stream.readline, ""):
+            buffer.append(line)
+            if on_output_line is not None:
+                try:
+                    on_output_line(line.rstrip("\r\n"), stream_name)
+                except Exception:
+                    # Callback is best-effort; draining must not stop.
+                    # Consider sys.excepthook for visibility in dev.
+                    pass
+    except (ValueError, OSError):
+        # Pipe closed concurrently — exit cleanly so join() returns.
+        pass
+```
+
+Two design choices worth thinking through:
+
+- **Should callback exceptions be logged?** Silent `pass` hides real bugs. `sys.excepthook` on a dedicated `sys.exc_info()` call surfaces to stderr in dev but pollutes CI logs. A conservative middle ground: log to a module-level counter and emit one warning at the end of the run if any occurred.
+- **Should buffering continue after `readline` fails?** No — if the pipe is closed the fd is gone; there's nothing to read. Exit the loop.
+
+## Done when
+
+- [ ] Both try/except layers are in place.
+- [ ] New test `tests/test_agent.py::test_pump_stream_continues_when_callback_raises` — run a real subprocess, pass a callback that raises on the first line, assert that all subsequent lines are still captured in `buffer` and the subprocess exits cleanly.
+- [ ] New test `tests/test_agent.py::test_pump_stream_exits_cleanly_on_closed_stream` — start `_pump_stream` on a pipe, close the read end, assert the thread exits within a short bounded join.
+- [ ] No regression in existing `_pump_stream` tests.
+- [ ] `uv run pytest`, `uv run ruff check .`, `uv run ruff format --check .`, `uv run ty check` all pass.
+
+## Context
+
+- `_pump_stream` is called from both `_run_agent_blocking` (two threads) and `_run_agent_streaming` (one thread, stderr only). All three call sites benefit from this fix.
+- Daemon threads (`daemon=True`) do not surface exceptions; see `threading.excepthook` (Python 3.8+) if you want a module-level install instead of per-call try/except. Per-call is more localized and preferred here.
+- `buffer.append(line)` is GIL-atomic so no lock needed for the append itself.
+- Related but **out of scope** for this task: reader-thread join timeouts (that's `critical-04`) and whether the event is even emitted when peek is off (`medium-01`). This task is purely about survivability of the pump loop.
+- The fix is small (a few lines) but the test coverage is the real deliverable — these are exactly the flaky-looking "agent timed out" bugs that are impossible to diagnose later without a regression test locking in the behavior.

tests/test_agent.py

@@ -13,6 +13,7 @@
 from ralphify._agent import (
     AgentResult,
     _kill_process_group,
+    _pump_stream,
     _read_agent_stream,
     _run_agent_blocking,
     _run_agent_streaming,
@@ -936,3 +937,137 @@ def test_callback_only_does_not_buffer(self, tmp_path):
         assert result.returncode == 0
         assert result.log_file is None
         assert received == ["line1", "line2"]
+
+
+class TestPumpStreamExceptionHandling:
+    """Tests for _pump_stream resilience against callback and I/O errors."""
+
+    def test_continues_when_callback_raises(self):
+        """A callback that raises on the first line must not prevent
+        subsequent lines from being buffered."""
+        script = (
+            "import sys; "
+            "print('line1'); print('line2'); print('line3'); "
+            "sys.stdout.flush()"
+        )
+        call_count = 0
+
+        def raising_callback(line, stream):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise RuntimeError("boom")
+
+        result = _run_agent_blocking(
+            [sys.executable, "-u", "-c", script],
+            prompt="",
+            timeout=10,
+            log_path_dir=None,
+            iteration=1,
+            on_output_line=raising_callback,
+        )
+
+        assert result.returncode == 0
+        # The callback was called for all three lines, even though
+        # the first invocation raised.
+        assert call_count == 3
+
+    def test_buffers_all_lines_when_callback_raises(self, tmp_path):
+        """When logging is enabled, all lines must be captured in the log
+        even if the callback raises on every single line."""
+        script = "import sys; print('a'); print('b'); print('c'); sys.stdout.flush()"
+
+        def always_raises(line, stream):
+            raise ValueError("always fails")
+
+        result = _run_agent_blocking(
+            [sys.executable, "-u", "-c", script],
+            prompt="",
+            timeout=10,
+            log_path_dir=tmp_path,
+            iteration=1,
+            on_output_line=always_raises,
+        )
+
+        assert result.returncode == 0
+        assert result.log_file is not None
+        log_text = result.log_file.read_text()
+        assert "a" in log_text
+        assert "b" in log_text
+        assert "c" in log_text
+
+    def test_exits_cleanly_on_closed_stream(self):
+        """When the read end of a pipe is closed, the pump thread must
+        exit within a short bounded join — not hang forever."""
+        import os
+        import threading
+
+        read_fd, write_fd = os.pipe()
+        read_file = os.fdopen(read_fd, "r")
+        write_file = os.fdopen(write_fd, "w")
+
+        buffer: list[str] = []
+        thread = threading.Thread(
+            target=_pump_stream,
+            args=(read_file, buffer, "stdout", None),
+            daemon=True,
+        )
+        thread.start()
+
+        # Write a line so the thread is actively reading, then close.
+        write_file.write("hello\n")
+        write_file.flush()
+        write_file.close()
+
+        # The thread must exit promptly — EOF from the closed write end.
+        thread.join(timeout=5)
+        assert not thread.is_alive(), (
+            "_pump_stream thread did not exit after pipe closed"
+        )
+        assert buffer == ["hello\n"]
+
+        read_file.close()
+
+    def test_exits_cleanly_on_valueerror(self):
+        """A stream whose readline raises ValueError (e.g. closed file)
+        must not crash the thread — it should exit cleanly."""
+        import threading
+
+        class ClosedStream:
+            """Fake stream that raises ValueError on readline, simulating
+            a concurrent close of the underlying file descriptor."""
+
+            def readline(self):
+                raise ValueError("I/O operation on closed file")
+
+        buffer: list[str] = []
+        thread = threading.Thread(
+            target=_pump_stream,
+            args=(ClosedStream(), buffer, "stdout", None),
+            daemon=True,
+        )
+        thread.start()
+        thread.join(timeout=5)
+        assert not thread.is_alive(), (
+            "_pump_stream thread did not exit after ValueError"
+        )
+        assert buffer == []
+
+    def test_exits_cleanly_on_oserror(self):
+        """A stream whose readline raises OSError must not crash the thread."""
+        import threading
+
+        class BrokenStream:
+            def readline(self):
+                raise OSError("stream error")
+
+        buffer: list[str] = []
+        thread = threading.Thread(
+            target=_pump_stream,
+            args=(BrokenStream(), buffer, "stdout", None),
+            daemon=True,
+        )
+        thread.start()
+        thread.join(timeout=5)
+        assert not thread.is_alive(), "_pump_stream thread did not exit after OSError"
+        assert buffer == []