From 19537fddfa78f88f603f9e13cd485bb27005a3b0 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Sat, 5 Sep 2026 11:30:50 -0400 Subject: [PATCH 1/6] fix(cli): keep agent output streaming alive on an unreadable line stream_process_output is the only reader of a child's stdout pipe. When readline() raised on an over-limit line, or the utf-8 decode failed, the loop exited, nothing drained the pipe, and the child blocked forever inside write() once 64 KiB accumulated. The agent presented as a silent freeze: 0% CPU, no further logs, health check dead, no traceback. Reproduced with an agent emitting an 81,988 character log line against asyncio's 65,536 byte StreamReader default: ValueError('Separator is found, but chunk is longer than limit') Two changes: - Handle readline() and decode failures per line rather than per loop. The line is dropped with a warning and streaming continues. readline() already removes the offending line, or clears the buffer, and resumes the transport before it raises, so continuing is safe and always makes progress. - Pass limit=8 MiB when spawning the ACP server and the Temporal worker, so ordinary large lines stream through instead of being dropped. The pre-existing handler logged this at debug level, but make_logger pins every logger to INFO with no env override, so that message could never be emitted. The one signal that would have explained the freeze was unreachable by construction. It is now a warning naming the exception. The debug spawn helpers in cli/debug/debug_handlers.py stream through this same function, so they can no longer deadlock either. They still use asyncio's default limit; raising it there needs the constant to live somewhere both modules can import, which run_handlers cannot provide without a cycle. Adds a regression test that fails against the previous loop: the child never exits because the reader stops draining its pipe. --- src/agentex/lib/cli/handlers/run_handlers.py | 44 ++++++++++-- tests/lib/cli/test_run_handlers_streaming.py | 72 ++++++++++++++++++++ 2 files changed, 112 insertions(+), 4 deletions(-) create mode 100644 tests/lib/cli/test_run_handlers_streaming.py diff --git a/src/agentex/lib/cli/handlers/run_handlers.py b/src/agentex/lib/cli/handlers/run_handlers.py index 3a43e95dd..6bc60ef97 100644 --- a/src/agentex/lib/cli/handlers/run_handlers.py +++ b/src/agentex/lib/cli/handlers/run_handlers.py @@ -23,6 +23,11 @@ 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 + class RunError(Exception): """An error occurred during agent run""" @@ -215,6 +220,7 @@ async def start_acp_server( env=env, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, + limit=SUBPROCESS_STREAM_LIMIT, ) @@ -234,23 +240,53 @@ 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 while True: - line = await process.stdout.readline() + try: + line = await process.stdout.readline() + except ValueError as e: + # Line longer than the stream limit. readline() has already + # dropped it (or cleared the buffer) and resumed the transport, + # so continuing is safe and always makes progress. + logger.warning( + f"Dropped an oversized log line from {prefix} ({e}); " + f"raise limit= on this process's create_subprocess_exec if it recurs." + ) + continue + 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}") + # 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. + 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): diff --git a/tests/lib/cli/test_run_handlers_streaming.py b/tests/lib/cli/test_run_handlers_streaming.py new file mode 100644 index 000000000..461cf82a3 --- /dev/null +++ b/tests/lib/cli/test_run_handlers_streaming.py @@ -0,0 +1,72 @@ +"""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 agentex.lib.cli.handlers.run_handlers import ( + SUBPROCESS_STREAM_LIMIT, + stream_process_output, +) + +# Emits a line 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. +CHILD_SCRIPT = """ +import sys +print("before") +print("X" * {oversized}) +for i in range(2000): + print("after", i, "y" * 60) +print("done") +""" + + +async def _drain(limit: int, oversized: int) -> int | None: + process = await asyncio.create_subprocess_exec( + sys.executable, + "-c", + CHILD_SCRIPT.format(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() -> 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 + returncode = await _drain(limit=limit, oversized=limit + 16_000) + + assert returncode == 0, "child did not exit: the reader stopped draining its pipe" + + +async def test_large_line_within_limit_is_streamed() -> None: + """A line larger than asyncio's 64 KiB default still streams under our limit.""" + returncode = await _drain(limit=SUBPROCESS_STREAM_LIMIT, oversized=82_000) + + assert returncode == 0 + + +async def test_subprocess_stream_limit_exceeds_asyncio_default() -> None: + """The whole point of the constant: asyncio's default is what breaks readline().""" + assert SUBPROCESS_STREAM_LIMIT > 64 * 1024 From df7f0ff5c55e23a8c370757bdfba89b1850f45c2 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Sat, 5 Sep 2026 11:35:50 -0400 Subject: [PATCH 2/6] Address greptile: assert large output is actually streamed The within-limit test only checked that the child exited, which the dropped case satisfies too, so it would still have passed if the line were discarded or if the limit= arguments were removed from the spawn helpers. - Count marker characters in the captured console output rather than matching the line, since rich wraps long output at terminal width. - Assert the oversized case emits none of them and still streams what follows. - Add a test that both spawn helpers pass limit=SUBPROCESS_STREAM_LIMIT, so the production wiring cannot regress unnoticed. --- tests/lib/cli/test_run_handlers_streaming.py | 70 +++++++++++++++----- 1 file changed, 55 insertions(+), 15 deletions(-) diff --git a/tests/lib/cli/test_run_handlers_streaming.py b/tests/lib/cli/test_run_handlers_streaming.py index 461cf82a3..7aa7c7f81 100644 --- a/tests/lib/cli/test_run_handlers_streaming.py +++ b/tests/lib/cli/test_run_handlers_streaming.py @@ -10,19 +10,26 @@ import sys import asyncio +from typing import Any +import pytest + +from agentex.lib.cli.handlers import run_handlers from agentex.lib.cli.handlers.run_handlers import ( SUBPROCESS_STREAM_LIMIT, + start_acp_server, + start_temporal_worker, stream_process_output, ) -# Emits a line 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. +# 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 = """ -import sys print("before") -print("X" * {oversized}) +print("{marker}" * {oversized}) for i in range(2000): print("after", i, "y" * 60) print("done") @@ -30,10 +37,11 @@ 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(oversized=oversized), + CHILD_SCRIPT.format(marker=MARKER, oversized=oversized), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, limit=limit, @@ -48,25 +56,57 @@ async def _drain(limit: int, oversized: int) -> int | None: return process.returncode -async def test_oversized_line_is_skipped_without_stalling_the_child() -> None: - """A line past the reader's limit is dropped and streaming continues. +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 - returncode = await _drain(limit=limit, oversized=limit + 16_000) + 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. -async def test_large_line_within_limit_is_streamed() -> None: - """A line larger than asyncio's 64 KiB default still streams under our limit.""" - returncode = await _drain(limit=SUBPROCESS_STREAM_LIMIT, oversized=82_000) + 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" + + +async def test_agent_subprocesses_are_spawned_with_the_larger_limit( + monkeypatch: pytest.MonkeyPatch, tmp_path: Any +) -> None: + """The helpers must pass limit=, or large lines are dropped in production.""" + 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) -async def test_subprocess_stream_limit_exceeds_asyncio_default() -> None: - """The whole point of the constant: asyncio's default is what breaks readline().""" - assert SUBPROCESS_STREAM_LIMIT > 64 * 1024 + assert seen == [SUBPROCESS_STREAM_LIMIT, SUBPROCESS_STREAM_LIMIT] + assert SUBPROCESS_STREAM_LIMIT > 64 * 1024, "asyncio's default is what breaks readline()" From ef6ff4652d3012bf0cabcae35b478310951a8e49 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Sat, 5 Sep 2026 11:50:22 -0400 Subject: [PATCH 3/6] fix(logging): honor LOG_LEVEL instead of pinning every logger to INFO make_logger hardcoded logging.INFO and read no override, so the SDK's log level could not be changed by any configuration. That is not only a missing knob: it made diagnostics already written into the SDK unreachable. The handler that explains why agent output streaming stopped logged at debug level, so the one message that would have identified a frozen worker could never be emitted. Read LOG_LEVEL from the environment, defaulting to INFO so nothing changes for anyone who does not set it. Unprefixed to match the SDK's other variables (ENVIRONMENT, REDIS_URL, AGENT_NAME), and read directly rather than through EnvVarKeys, because environment_variables imports this module. getLevelName returns the string "Level FOO" for an unrecognized name, so an unusable value falls back to INFO rather than being handed to setLevel, where a typo would silently disable logging. --- src/agentex/lib/utils/logging.py | 21 ++++++++- tests/lib/utils/test_logging_level.py | 66 +++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 1 deletion(-) create mode 100644 tests/lib/utils/test_logging_level.py diff --git a/src/agentex/lib/utils/logging.py b/src/agentex/lib/utils/logging.py index 5bbaf61ac..a0d39331b 100644 --- a/src/agentex/lib/utils/logging.py +++ b/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": diff --git a/tests/lib/utils/test_logging_level.py b/tests/lib/utils/test_logging_level.py new file mode 100644 index 000000000..16b171e33 --- /dev/null +++ b/tests/lib/utils/test_logging_level.py @@ -0,0 +1,66 @@ +"""Tests for log level resolution in agentex.lib.utils.logging. + +The level used to be pinned to INFO with no override, so a debug() call could +never be emitted on any configuration. That is not just a missing feature: it +made diagnostics that were already written into the SDK unreachable. +""" + +from __future__ import annotations + +import logging + +import pytest + +from agentex.lib.utils.logging import ( + DEFAULT_LOG_LEVEL, + make_logger, + resolve_log_level, +) + + +def test_defaults_to_info_when_unset(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("LOG_LEVEL", raising=False) + + assert resolve_log_level() == DEFAULT_LOG_LEVEL == logging.INFO + + +@pytest.mark.parametrize( + ("configured", "expected"), + [ + ("DEBUG", logging.DEBUG), + ("debug", logging.DEBUG), + (" WaRnInG ", logging.WARNING), + ("ERROR", logging.ERROR), + ("CRITICAL", logging.CRITICAL), + ], +) +def test_reads_level_from_env( + monkeypatch: pytest.MonkeyPatch, configured: str, expected: int +) -> None: + monkeypatch.setenv("LOG_LEVEL", configured) + + assert resolve_log_level() == expected + + +@pytest.mark.parametrize("configured", ["", " ", "VERBOSE", "10x", "TRUE"]) +def test_falls_back_to_info_on_an_unusable_value( + monkeypatch: pytest.MonkeyPatch, configured: str +) -> None: + """A typo must not silently disable logging. + + logging.getLevelName returns the string "Level FOO" for anything it does not + recognise, which would otherwise be handed straight to setLevel. + """ + monkeypatch.setenv("LOG_LEVEL", configured) + + assert resolve_log_level() == logging.INFO + + +def test_make_logger_applies_the_configured_level(monkeypatch: pytest.MonkeyPatch) -> None: + """The regression that mattered: a debug() call must be able to emit.""" + monkeypatch.setenv("LOG_LEVEL", "DEBUG") + + logger = make_logger("agentex.tests.level_from_env") + + assert logger.level == logging.DEBUG + assert logger.isEnabledFor(logging.DEBUG) From 35a0e5e5cc144530034034a2a2bcbd81a45b2fe5 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Sat, 5 Sep 2026 11:51:49 -0400 Subject: [PATCH 4/6] fix(cli): do not assume why a line was unreadable, and bound the retries The previous commit caught ValueError from readline() and asserted it was a limit overrun. It only knew an exception had been raised. That mattered beyond the message. Skipping the line is safe only for the overrun case, where readline() has already discarded the line and resumed the transport before raising. readline() flattens LimitOverrunError into a bare ValueError, so the two are indistinguishable at the call site, and any other ValueError that consumes nothing would have spun the loop forever at 100% CPU while still not draining the pipe. That is worse than the freeze being fixed. Now: report the actual exception with repr() and no assumed cause, and bound consecutive failures at MAX_CONSECUTIVE_READ_ERRORS (100). Past that, re-raise so the outer handler reports that nothing is draining the child's stdout. The outer try/except is load-bearing, not vestigial: it is the escalation path for that re-raise, and it still covers console.print failures and non-ValueError transport errors. It does not swallow cancellation, since CancelledError derives from BaseException, so the auto-reload path that cancels these tasks is unaffected (verified). (cherry picked from commit 13fa1f5c28f1687f1f90469c22eba9bccf1bbc82) --- src/agentex/lib/cli/handlers/run_handlers.py | 27 ++++++++++++++++---- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/src/agentex/lib/cli/handlers/run_handlers.py b/src/agentex/lib/cli/handlers/run_handlers.py index 6bc60ef97..12adf091f 100644 --- a/src/agentex/lib/cli/handlers/run_handlers.py +++ b/src/agentex/lib/cli/handlers/run_handlers.py @@ -28,6 +28,11 @@ # 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. +MAX_CONSECUTIVE_READ_ERRORS = 100 + class RunError(Exception): """An error occurred during agent run""" @@ -255,19 +260,31 @@ async def stream_process_output(process: asyncio.subprocess.Process, prefix: str try: if process.stdout is None: return + consecutive_read_errors = 0 while True: try: line = await process.stdout.readline() except ValueError as e: - # Line longer than the stream limit. readline() has already - # dropped it (or cleared the buffer) and resumed the transport, - # so continuing is safe and always makes progress. + # 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"Dropped an oversized log line from {prefix} ({e}); " - f"raise limit= on this process's create_subprocess_exec if it recurs." + 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 From a4cd8722bbe962c8cd3e19de1e6572818e8ddc16 Mon Sep 17 00:00:00 2001 From: Rishav C Date: Sat, 5 Sep 2026 12:14:44 -0400 Subject: [PATCH 5/6] 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. --- src/agentex/lib/cli/handlers/run_handlers.py | 9 ++-- tests/lib/cli/test_run_handlers_streaming.py | 51 ++++++++++++++++++++ 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/src/agentex/lib/cli/handlers/run_handlers.py b/src/agentex/lib/cli/handlers/run_handlers.py index 12adf091f..a72deda29 100644 --- a/src/agentex/lib/cli/handlers/run_handlers.py +++ b/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." diff --git a/tests/lib/cli/test_run_handlers_streaming.py b/tests/lib/cli/test_run_handlers_streaming.py index 7aa7c7f81..16e8ed1a8 100644 --- a/tests/lib/cli/test_run_handlers_streaming.py +++ b/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: From d0a1c2ea15445fa180b2d5918bc09d253c36246c Mon Sep 17 00:00:00 2001 From: Rishav C Date: Sat, 5 Sep 2026 12:23:58 -0400 Subject: [PATCH 6/6] 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. --- src/agentex/lib/cli/debug/debug_handlers.py | 3 +++ src/agentex/lib/cli/handlers/run_handlers.py | 6 +---- src/agentex/lib/cli/utils/cli_utils.py | 12 ++++++++++ tests/lib/cli/test_run_handlers_streaming.py | 23 +++++++++++++++++--- 4 files changed, 36 insertions(+), 8 deletions(-) diff --git a/src/agentex/lib/cli/debug/debug_handlers.py b/src/agentex/lib/cli/debug/debug_handlers.py index 98746387f..a27d682cd 100644 --- a/src/agentex/lib/cli/debug/debug_handlers.py +++ b/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, ) diff --git a/src/agentex/lib/cli/handlers/run_handlers.py b/src/agentex/lib/cli/handlers/run_handlers.py index a72deda29..18ee84e93 100644 --- a/src/agentex/lib/cli/handlers/run_handlers.py +++ b/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. diff --git a/src/agentex/lib/cli/utils/cli_utils.py b/src/agentex/lib/cli/utils/cli_utils.py index 43b3fba62..4238e8fd9 100644 --- a/src/agentex/lib/cli/utils/cli_utils.py +++ b/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" diff --git a/tests/lib/cli/test_run_handlers_streaming.py b/tests/lib/cli/test_run_handlers_streaming.py index 16e8ed1a8..8f0ab13b5 100644 --- a/tests/lib/cli/test_run_handlers_streaming.py +++ b/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()"