fix: keep agent output streaming alive on an unreadable line, and honor LOG_LEVEL

src/agentex/lib/cli/debug/debug_handlers.py

@@ -16,6 +16,7 @@
     pass
 
 from agentex.lib.utils.logging import make_logger
+from agentex.lib.cli.utils.cli_utils import SUBPROCESS_STREAM_LIMIT
 
 from .debug_config import DebugConfig, resolve_debug_port
 
@@ -66,6 +67,7 @@ async def start_temporal_worker_debug(
         env=debug_env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 
@@ -119,6 +121,7 @@ async def start_acp_server_debug(
         env=debug_env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 

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

@@ -12,6 +12,7 @@
 from agentex.lib.cli.debug import DebugConfig, start_acp_server_debug, start_temporal_worker_debug
 from agentex.lib.utils.logging import make_logger
 from agentex.config.agent_manifest import AgentManifest
+from agentex.lib.cli.utils.cli_utils import SUBPROCESS_STREAM_LIMIT
 from agentex.lib.cli.utils.path_utils import (
     get_file_paths,
     calculate_uvicorn_target_for_local,
@@ -23,6 +24,11 @@
 logger = make_logger(__name__)
 console = Console()
 
+# How many consecutive unreadable lines to skip before giving up on the stream.
+# Skipping is only known-safe for the limit-overrun case; this bounds the damage
+# if some other error repeats without consuming anything.
+MAX_CONSECUTIVE_READ_ERRORS = 100
+
 
 class RunError(Exception):
     """An error occurred during agent run"""
@@ -215,6 +221,7 @@ async def start_acp_server(
         env=env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 
@@ -234,23 +241,68 @@ async def start_temporal_worker(
         env=env,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.STDOUT,
+        limit=SUBPROCESS_STREAM_LIMIT,
     )
 
 
 async def stream_process_output(process: asyncio.subprocess.Process, prefix: str):
-    """Stream process output with prefix"""
+    """Stream process output with prefix.
+
+    This loop is the only reader of the child's stdout pipe. If it ever stops
+    reading, the pipe fills and the child blocks forever inside ``write()``,
+    which presents as a silent freeze: 0% CPU, no further logs, no traceback.
+    So a single unreadable line must never end the loop.
+    """
     try:
         if process.stdout is None:
             return
+        consecutive_read_errors = 0
         while True:
-            line = await process.stdout.readline()
+            try:
+                line = await process.stdout.readline()
+            except ValueError as e:
+                # readline() raises ValueError when a line exceeds the stream limit.
+                # In *that* case it has already discarded the line and resumed the
+                # transport, so skipping it makes guaranteed progress. Any other
+                # ValueError carries no such guarantee, and retrying it forever would
+                # spin without draining. We cannot tell the two apart (readline
+                # flattens LimitOverrunError into a bare ValueError), so bound the
+                # retries and let the outer handler report the hang risk.
+                consecutive_read_errors += 1
+                if consecutive_read_errors > MAX_CONSECUTIVE_READ_ERRORS:
+                    raise
+                logger.warning(
+                    f"Skipping an unreadable line from {prefix}: {e!r} "
+                    f"(consecutive failure {consecutive_read_errors}/{MAX_CONSECUTIVE_READ_ERRORS}). "
+                    f"If this says the chunk exceeded the limit, raise limit= on this "
+                    f"process's create_subprocess_exec."
+                )
+                continue
+
+            consecutive_read_errors = 0
+
             if not line:
                 break
-            decoded_line = line.decode("utf-8").rstrip()
+
+            try:
+                decoded_line = line.decode("utf-8").rstrip()
+            except UnicodeDecodeError as e:
+                logger.warning(f"Dropped an undecodable log line from {prefix} ({e}).")
+                continue
+
             if decoded_line:  # Only print non-empty lines
                 console.print(f"[dim]{prefix}:[/dim] {decoded_line}")
     except Exception as e:
-        logger.debug(f"Output streaming ended for {prefix}: {e}")
+        # 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."
+        )
 
 
 async def run_agent(manifest_path: str, debug_config: "DebugConfig | None" = None):

src/agentex/lib/cli/utils/cli_utils.py

@@ -5,6 +5,18 @@
 
 console = Console()
 
+# asyncio's StreamReader defaults to 64 KiB, and a single log line above that makes
+# readline() raise. Agents legitimately emit large lines (serialized charts, payloads
+# echoed back by validation errors), so give the reader room before it has to drop one.
+#
+# Lives here rather than beside its users so that both the normal spawns in
+# cli/handlers/run_handlers.py and the debug spawns in cli/debug/debug_handlers.py can
+# import it: run_handlers imports cli.debug, so the constant cannot live in either one.
+# Keep the two in step. A subprocess left on the asyncio default overruns far more
+# easily, and enough consecutive overruns exhaust the reader's retry bound and stop it
+# draining, which is the deadlock the bound is there to avoid.
+SUBPROCESS_STREAM_LIMIT = 8 * 1024 * 1024
+
 
 def handle_questionary_cancellation(
     result: str | None, operation: str = "operation"

src/agentex/lib/utils/logging.py

@@ -11,6 +11,25 @@
 
 ctx_var_request_id = contextvars.ContextVar[str]("request_id")
 
+DEFAULT_LOG_LEVEL = logging.INFO
+
+
+def resolve_log_level() -> int:
+    """Read the log level from ``LOG_LEVEL``, falling back to INFO.
+
+    Read straight from the environment rather than through ``EnvVarKeys``, since
+    ``environment_variables`` imports this module and the reverse would be a cycle.
+
+    ``getLevelName`` returns the string ``"Level FOO"`` for anything it does not
+    recognise, so the isinstance check is what stops a typo in ``LOG_LEVEL`` from
+    silently turning logging off.
+    """
+    configured = os.getenv("LOG_LEVEL")
+    if not configured:
+        return DEFAULT_LOG_LEVEL
+    level = logging.getLevelName(configured.strip().upper())
+    return level if isinstance(level, int) else DEFAULT_LOG_LEVEL
+
 
 class CustomJSONFormatter(json_log_formatter.JSONFormatter):
     def json_record(self, message: str, extra: dict, record: logging.LogRecord) -> dict:  # type: ignore[override]
@@ -51,7 +70,7 @@ def make_logger(name: str) -> logging.Logger:
     """
     # Create a console object to print colored text
     logger = logging.getLogger(name)
-    logger.setLevel(logging.INFO)
+    logger.setLevel(resolve_log_level())
 
     environment = os.getenv("ENVIRONMENT")
     if environment == "local":

tests/lib/cli/test_run_handlers_streaming.py

@@ -0,0 +1,180 @@
+"""Tests for run_handlers output streaming.
+
+stream_process_output is the only reader of a child's stdout pipe. If it stops
+reading, the pipe fills and the child blocks forever inside write(), which
+presents as a silent freeze with no traceback. These tests pin the behaviour
+that prevents that: a line the reader cannot handle is skipped, not fatal.
+"""
+
+from __future__ import annotations
+
+import sys
+import asyncio
+from typing import Any
+
+import pytest
+
+from agentex.lib.cli.debug import DebugMode, DebugConfig
+from agentex.lib.cli.handlers import run_handlers
+from agentex.lib.cli.debug.debug_handlers import (
+    start_acp_server_debug,
+    start_temporal_worker_debug,
+)
+from agentex.lib.cli.handlers.run_handlers import (
+    SUBPROCESS_STREAM_LIMIT,
+    start_acp_server,
+    start_temporal_worker,
+    stream_process_output,
+)
+
+# Emits a line of MARKER over the reader's limit, then enough further output to
+# more than fill a 64 KiB pipe. If the reader stops draining, the child cannot
+# finish its writes and never exits.
+MARKER = "X"
+
+CHILD_SCRIPT = """
+print("before")
+print("{marker}" * {oversized})
+for i in range(2000):
+    print("after", i, "y" * 60)
+print("done")
+"""
+
+
+async def _drain(limit: int, oversized: int) -> int | None:
+    """Run the child under stream_process_output. None means it never exited."""
+    process = await asyncio.create_subprocess_exec(
+        sys.executable,
+        "-c",
+        CHILD_SCRIPT.format(marker=MARKER, oversized=oversized),
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.STDOUT,
+        limit=limit,
+    )
+    streamer = asyncio.create_task(stream_process_output(process, "TEST"))
+    try:
+        await asyncio.wait_for(asyncio.gather(streamer, process.wait()), timeout=60)
+    except TimeoutError:
+        process.kill()
+        await process.wait()
+        return None
+    return process.returncode
+
+
+async def test_oversized_line_is_skipped_without_stalling_the_child(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A line past the reader's limit is dropped, and streaming continues.
+
+    Before this was handled per line, readline() raised, the loop exited, and the
+    child deadlocked on a full pipe. The child reaching exit is the assertion.
+    """
+    limit = 64 * 1024
+    oversized = limit + 16_000
+
+    returncode = await _drain(limit=limit, oversized=oversized)
+    out = capsys.readouterr().out
+
+    assert returncode == 0, "child did not exit: the reader stopped draining its pipe"
+    # The offending line is gone, but everything after it still streamed.
+    assert out.count(MARKER) == 0
+    assert "done" in out
+
+
+async def test_large_line_within_the_limit_is_streamed_in_full(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """A line over asyncio's 64 KiB default still reaches the console under our limit.
+
+    Counts marker characters rather than matching the line, because rich wraps
+    long output across terminal-width lines.
+    """
+    oversized = 82_000
+
+    returncode = await _drain(limit=SUBPROCESS_STREAM_LIMIT, oversized=oversized)
+    out = capsys.readouterr().out
+
+    assert returncode == 0
+    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_every_spawn_uses_the_larger_limit(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Any
+) -> None:
+    """Every spawn must pass limit=, including the debug ones.
+
+    A subprocess left on asyncio's default overruns far more easily, and enough
+    consecutive overruns exhaust MAX_CONSECUTIVE_READ_ERRORS and stop the reader
+    draining, which is the deadlock the bound exists to avoid.
+    """
+    seen: list[int | None] = []
+
+    async def fake_exec(*_args: Any, **kwargs: Any) -> None:
+        seen.append(kwargs.get("limit"))
+
+    monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec)
+    monkeypatch.setattr(run_handlers, "calculate_uvicorn_target_for_local", lambda *_: "project.acp")
+
+    await start_acp_server(tmp_path / "acp.py", 8000, {}, tmp_path)
+    await start_temporal_worker(tmp_path / "run_worker.py", {}, tmp_path)
+
+    # BOTH, since each helper refuses unless its own mode is enabled.
+    debug_config = DebugConfig(
+        enabled=True, mode=DebugMode.BOTH, port=5678, wait_for_attach=False, auto_port=False
+    )
+    await start_acp_server_debug(tmp_path / "acp.py", 8000, {}, debug_config)
+    await start_temporal_worker_debug(tmp_path / "run_worker.py", {}, debug_config)
+
+    assert seen == [SUBPROCESS_STREAM_LIMIT] * 4, f"a spawn is missing limit=: {seen}"
+    assert SUBPROCESS_STREAM_LIMIT > 64 * 1024, "asyncio's default is what breaks readline()"