fix: guard callbacks in _read_agent_stream against exceptions

_pump_stream (blocking path) already wraps on_output_line in try/except
so a raising callback cannot kill the drain loop. _read_agent_stream
(streaming path) did not — a transient error in the console emitter or
any other callback would crash the stream reader and kill the entire run.

Guard both on_output_line and on_activity with per-call try/except,
matching _pump_stream's resilience contract.

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

Author: 2 people

src/ralphify/_agent.py

@@ -237,7 +237,11 @@ def _read_agent_stream(
     for line in stdout:
         stdout_lines.append(line)
         if on_output_line is not None:
-            on_output_line(line.rstrip("\r\n"), _STDOUT)
+            try:
+                on_output_line(line.rstrip("\r\n"), _STDOUT)
+            except Exception:
+                # Callback is best-effort; draining must not stop.
+                pass
 
         stripped = line.strip()
         if stripped:
@@ -251,7 +255,11 @@ def _read_agent_stream(
                 ):
                     result_text = parsed[_RESULT_FIELD]
                 if on_activity is not None:
-                    on_activity(parsed)
+                    try:
+                        on_activity(parsed)
+                    except Exception:
+                        # Callback is best-effort; draining must not stop.
+                        pass
 
         if deadline is not None and time.monotonic() > deadline:
             return _StreamResult(

tests/test_agent.py

@@ -204,6 +204,52 @@ def test_last_result_wins(self):
 
         assert result.result_text == "second"
 
+    def test_continues_when_on_output_line_raises(self):
+        """A raising on_output_line callback must not crash the stream reader.
+
+        This mirrors _pump_stream's behavior where callback exceptions are
+        caught per-line so that draining continues.  Without this guard, a
+        transient rendering error in the console emitter would kill the
+        entire streaming run."""
+        call_count = 0
+
+        def raising_callback(line, stream):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise RuntimeError("boom")
+
+        stream = io.StringIO("line1\nline2\nline3\n")
+        result = _read_agent_stream(
+            stream, deadline=None, on_activity=None, on_output_line=raising_callback
+        )
+
+        assert len(result.stdout_lines) == 3
+        assert call_count == 3
+        assert result.timed_out is False
+
+    def test_continues_when_on_activity_raises(self):
+        """A raising on_activity callback must not crash the stream reader."""
+        call_count = 0
+
+        def raising_activity(data):
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise RuntimeError("boom")
+
+        stream = io.StringIO(
+            '{"type": "status", "msg": "a"}\n'
+            '{"type": "status", "msg": "b"}\n'
+        )
+        result = _read_agent_stream(
+            stream, deadline=None, on_activity=raising_activity
+        )
+
+        assert len(result.stdout_lines) == 2
+        assert call_count == 2
+        assert result.timed_out is False
+
 
 class TestExecuteAgentBlocking:
     @patch(MOCK_SUBPROCESS, side_effect=ok_proc)