Address greptile: give the debug spawns the same stream limit

The retry bound counts every consecutive ValueError, including limit overruns,
which do make progress. That is harmless where the limit is 8 MiB, but the
debug spawns were left on asyncio's 64 KiB default, so an agent emitting enough
consecutive large lines under --debug exhausts the bound, the reader re-raises,
and the child deadlocks on a full pipe. The two changes were individually
defensible and together reintroduced the bug being fixed.

Move SUBPROCESS_STREAM_LIMIT to cli/utils/cli_utils.py, which imports only
typer and rich, so both handlers can share it. It could not live in either
handler: run_handlers imports cli.debug, so the reverse import would cycle.

Extends the wiring test to all four spawn sites rather than two, so a new one
cannot be added on the default limit without failing.

Author: chakrris

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,11 +24,6 @@
 logger = make_logger(__name__)
 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 by validation errors), so give the reader room before it has to drop one.
-SUBPROCESS_STREAM_LIMIT = 8 * 1024 * 1024
-
 # 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.

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"

tests/lib/cli/test_run_handlers_streaming.py

@@ -14,7 +14,12 @@
 
 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,
@@ -144,10 +149,15 @@ async def readline(self) -> bytes:
         await task
 
 
-async def test_agent_subprocesses_are_spawned_with_the_larger_limit(
+async def test_every_spawn_uses_the_larger_limit(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Any
 ) -> None:
-    """The helpers must pass limit=, or large lines are dropped in production."""
+    """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:
@@ -159,5 +169,12 @@ async def fake_exec(*_args: Any, **kwargs: Any) -> None:
     await start_acp_server(tmp_path / "acp.py", 8000, {}, tmp_path)
     await start_temporal_worker(tmp_path / "run_worker.py", {}, tmp_path)
 
-    assert seen == [SUBPROCESS_STREAM_LIMIT, SUBPROCESS_STREAM_LIMIT]
+    # 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()"