test(cli): cover the retry bound and cancellation, refresh a stale comment

The bound and the cancellation behaviour were verified by hand but not pinned
by the suite, so both could regress silently.

- A reader whose readline() raises without consuming anything must stop after
  MAX_CONSECUTIVE_READ_ERRORS rather than spinning. This is the case the bound
  exists for, and it hangs the suite if the bound is removed.
- Cancelling the streaming task must raise CancelledError out of it, since the
  auto-reload path cancels these tasks and swallowing it would hang restarts.

Also corrects the outer handler's comment, which said make_logger pins loggers
to INFO. That was true when it was written and is no longer, since LOG_LEVEL is
now honored. It now records what the handler is for: the escalation path for the
bounded re-raise, and why cancellation passes straight through it.

Author: chakrris

src/agentex/lib/cli/handlers/run_handlers.py

@@ -297,9 +297,12 @@ async def stream_process_output(process: asyncio.subprocess.Process, prefix: str
             if decoded_line:  # Only print non-empty lines
                 console.print(f"[dim]{prefix}:[/dim] {decoded_line}")
     except Exception as e:
-        # Anything reaching here ends the loop, so the child is now at risk of
-        # blocking on a full pipe. make_logger pins loggers to INFO, so the
-        # previous debug() here could never be emitted.
+        # The escalation path, including for the re-raise above. Anything reaching
+        # here ends the loop, so the child is now at risk of blocking on a full pipe.
+        # Warning rather than debug: this used to be a debug() that make_logger could
+        # never emit, which is why three freezes produced no clue.
+        # CancelledError derives from BaseException, so the auto-reload path that
+        # cancels these tasks passes straight through and is unaffected.
         logger.warning(
             f"Output streaming for {prefix} stopped on {e!r}. "
             f"Nothing is draining its stdout now, so {prefix} will hang once the pipe fills."

tests/lib/cli/test_run_handlers_streaming.py

@@ -93,6 +93,57 @@ async def test_large_line_within_the_limit_is_streamed_in_full(
     assert out.count(MARKER) == oversized, "the large line was dropped rather than streamed"
 
 
+class _AlwaysFailingReader:
+    """A reader whose readline() raises without consuming anything.
+
+    The dangerous shape: skipping it makes no progress, so an unbounded retry
+    would spin at 100% CPU while still not draining the pipe.
+    """
+
+    def __init__(self) -> None:
+        self.attempts = 0
+
+    async def readline(self) -> bytes:
+        self.attempts += 1
+        raise ValueError("unreadable, and nothing was consumed")
+
+
+class _FakeProcess:
+    def __init__(self, stdout: Any) -> None:
+        self.stdout = stdout
+
+
+async def test_repeated_unreadable_lines_give_up_instead_of_spinning() -> None:
+    """A ValueError that consumes nothing must not loop forever."""
+    reader = _AlwaysFailingReader()
+
+    await asyncio.wait_for(
+        stream_process_output(_FakeProcess(reader), "TEST"), timeout=30
+    )
+
+    assert reader.attempts == run_handlers.MAX_CONSECUTIVE_READ_ERRORS + 1
+
+
+async def test_cancellation_is_not_swallowed() -> None:
+    """The auto-reload path cancels these tasks, so cancel must propagate.
+
+    CancelledError derives from BaseException, so the outer `except Exception`
+    does not catch it. This pins that, since swallowing it would hang restarts.
+    """
+
+    class _NeverReturns:
+        async def readline(self) -> bytes:
+            await asyncio.sleep(3600)
+            return b""
+
+    task = asyncio.create_task(stream_process_output(_FakeProcess(_NeverReturns()), "TEST"))
+    await asyncio.sleep(0)
+    task.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+
 async def test_agent_subprocesses_are_spawned_with_the_larger_limit(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Any
 ) -> None: