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 3a43e95dd..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,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): 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/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/cli/test_run_handlers_streaming.py b/tests/lib/cli/test_run_handlers_streaming.py new file mode 100644 index 000000000..8f0ab13b5 --- /dev/null +++ b/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()" 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)