From 195bf358ca70cbb5431a3da328e5d0238b830e0e Mon Sep 17 00:00:00 2001 From: Evan <1178327328@qq.com> Date: Fri, 23 Oct 2026 10:30:26 -0400 Subject: [PATCH] fix(acp): fail fast when an ACP agent exits and drain its stderr (#7354) * fix(acp): fail fast when an ACP agent exits and drain its stderr An ACP agent that exited before answering (missing credentials, a wrong argument) left invoke_acp_agent waiting for the whole timeout_seconds budget (30 minutes by default) and then suggested raising the timeout. The pinned SDK never fails pending requests when the agent's stdout reaches EOF, so nothing noticed the process was gone. The SDK also opens the agent's stderr as a pipe and never reads it. An agent that logged more than asyncio's pipe buffering absorbs blocked on its next stderr write and stalled the invocation the same way. Race the ACP exchange against the agent process: when it exits (or a newer SDK reports the connection closed), allow a short grace for responses still buffered in the pipe, then report the exit code. Drain stderr continuously into a bounded tail that is logged on failure and kept out of model-visible text. * fix(acp): log the full stderr tail on timeout and pin tail retention Log the timeout tail after the SDK context has terminated and reaped the agent, so lines still in flight when it stalled are included. Make the exit test write more stderr than the retained tail, so keeping the start of the stream instead of its end fails the assertion on the fatal line. --- README.md | 1 + .../packages/harness/deerflow/tools/AGENTS.md | 2 + .../tools/builtins/invoke_acp_agent_tool.py | 133 ++++++++++++++--- .../blocking_io/test_invoke_acp_agent_tool.py | 3 +- backend/tests/test_invoke_acp_agent_tool.py | 135 +++++++++++++++++- 5 files changed, 249 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index 40a648af54..93b78f53e6 100644 --- a/README.md +++ b/README.md @@ -417,6 +417,7 @@ streams simulate chunks from a non-streaming response, while no-tool streams sta - ACP agent entries are separate from model providers — if you configure `acp_agents.codex`, point it at a Codex ACP adapter such as `npx -y @zed-industries/codex-acp` - A bare or absolute ACP agent `command` is resolved before the agent starts, so npm-installed launchers work on Windows too: `npx` or `mcode` is spawned as its `npx.cmd`/`mcode.cmd` shim instead of failing with a command-not-found error, and an absolute path such as `C:\Users\me\AppData\Roaming\npm\mcode` resolves to the `mcode.cmd` shim beside npm's extensionless shell script instead of failing with `WinError 193`. The lookup uses the `PATH` the agent subprocess will actually see (`acp_agents..env.PATH` when set, otherwise the Gateway's) and hands the spawn an absolute path, so a relative `PATH` entry cannot be re-interpreted inside the agent's own workspace. A relative `command` (for example `bin/mcode`) is used as configured, and a command that cannot be resolved is still reported with the install guidance below. - Each ACP agent's `timeout_seconds` (default: 1800) is one shared budget for initialization, session creation, and the prompt, starting after the subprocess launches. On timeout, DeerFlow aborts the invocation and closes the subprocess before returning an error. Workspace/MCP preparation and subprocess cleanup are outside this budget. A `TimeoutError` raised by the SDK before this deadline expires retains its own error message. + - If an ACP agent exits before answering (for example because it is not authenticated or a configured argument is wrong), DeerFlow reports its exit code within a few seconds instead of waiting for `timeout_seconds`, and records the end of the agent's stderr in the Gateway log. Agent stderr is read continuously, so an agent that logs heavily cannot stall on a full pipe. - MiniMax Code speaks ACP directly. Install and authenticate it, then add it as an ACP agent: ```bash diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md index 17b14987a7..8161c8edfb 100644 --- a/backend/packages/harness/deerflow/tools/AGENTS.md +++ b/backend/packages/harness/deerflow/tools/AGENTS.md @@ -119,6 +119,8 @@ E2B output sync records remote file versions and actual host file metadata in a - ACP launchers must be real ACP adapters. The standard `codex` CLI is not ACP-compatible by itself; configure a wrapper such as `npx -y @zed-industries/codex-acp` or an installed `codex-acp` binary - MiniMax Code speaks ACP directly: configure `command: mcode` with `args: ["acp"]`. It receives DeerFlow's enabled MCP servers and uses the per-thread ACP workspace; the Gateway process must have an authenticated `mcode` executable on `PATH` - `timeout_seconds` is one deadline shared by initialization, session creation, and prompt after process launch. Keep the SDK process context outside the timeout so teardown finishes before the tool returns; workspace/MCP preparation and teardown are outside the protocol budget. Real-process regressions in `tests/test_invoke_acp_agent_tool.py` cover every hung phase, the shared budget, success, and caller cancellation. Only report budget expiry when the timeout context's `expired()` is true; otherwise propagate SDK `TimeoutError` to `_format_invocation_error`. +- `_StderrTail` drains the agent's stderr for the whole invocation, keeping the last 8 KiB. The SDK opens stderr as a pipe and never reads it; without draining, an agent that logs past asyncio's pipe buffering blocks on its next write and the call stalls until `timeout_seconds`. Cancel the drain only after the SDK context has reaped the process +- `_await_while_agent_runs` races the ACP exchange against `proc.wait()`. SDK 0.9 never fails pending requests on stdout EOF and newer releases raise `ConnectionError`; both become `_AgentExitedError` after `_EXIT_DRAIN_SECONDS`, which lets responses still buffered behind the exit complete normally. The result reports only the exit code; the stderr tail goes to the Gateway log, never to model-visible text. Other SDK exceptions keep their existing handling - ACP results collect only `agent_message_chunk` text. Thought chunks remain internal and must not be concatenated into the tool result - Missing ACP executables now return an actionable error message instead of a raw `[Errno 2]` - A configured `command` that is a bare name or an absolute path is resolved before the spawn (`_resolve_agent_command`), off the event loop because the lookup stats the filesystem. The ACP SDK's stdio transport calls `asyncio.create_subprocess_exec`, which performs no `PATHEXT` lookup on Windows, so a bare npm launcher (`npx`, `mcode`) would raise `FileNotFoundError`, and an absolute path without an extension (`C:\npm\mcode`, npm's shell script beside its `.cmd` shim) would die with `WinError 193`, although the shim exists. Resolution consults the `PATH` the agent subprocess will actually see (`acp_agents..env.PATH` when the agent config sets one, the Gateway's otherwise) and returns an absolute path, so a relative `PATH` entry cannot be re-interpreted inside the ACP workspace; an absolute command is resolved inside its own directory. A relative `command` goes to the spawn unchanged, so the spawn — not `shutil.which` — decides what it refers to, and `_format_invocation_error` always receives the configured value so its remediation still matches `config.yaml` diff --git a/backend/packages/harness/deerflow/tools/builtins/invoke_acp_agent_tool.py b/backend/packages/harness/deerflow/tools/builtins/invoke_acp_agent_tool.py index 17415e6e5b..c7577d50fb 100644 --- a/backend/packages/harness/deerflow/tools/builtins/invoke_acp_agent_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/invoke_acp_agent_tool.py @@ -12,6 +12,12 @@ from pydantic import BaseModel, Field logger = logging.getLogger(__name__) +# Bytes of agent stderr kept for the Gateway log when an invocation fails. +_STDERR_TAIL_BYTES = 8 * 1024 +# After the agent process exits, how long the SDK receive loop may keep reading +# responses still buffered in the stdout pipe before the call is declared failed. +_EXIT_DRAIN_SECONDS = 2.0 + class _InvokeACPAgentInput(BaseModel): agent: str = Field(description="Name of the ACP agent to invoke") @@ -185,6 +191,70 @@ def _resolve_agent_command(command: str, path: str | None = None) -> str: return resolved if os.path.isabs(resolved) else os.path.abspath(resolved) +class _StderrTail: + """Drain an agent's stderr pipe, keeping only a bounded tail. + + The SDK spawns agents with ``stderr=PIPE`` but never reads it. Once an agent + has logged more than asyncio's pipe buffering absorbs, its next stderr write + blocks, the agent stops answering, and the invocation stalls until the + timeout. Draining continuously keeps verbose agents running. + """ + + def __init__(self, stream: asyncio.StreamReader | None) -> None: + self._data = bytearray() + self._task = asyncio.create_task(self._drain(stream)) if stream is not None else None + + async def _drain(self, stream: asyncio.StreamReader) -> None: + while chunk := await stream.read(65536): + self._data += chunk + del self._data[:-_STDERR_TAIL_BYTES] + + async def text(self, *, wait: float = 0.0) -> str: + """Return the tail, waiting up to ``wait`` seconds for the pipe to reach EOF.""" + if self._task is not None and wait > 0: + await asyncio.wait({self._task}, timeout=wait) + return self._data.decode("utf-8", errors="replace").strip() + + async def aclose(self) -> None: + if self._task is not None: + # A descendant can keep the pipe open after the agent itself is reaped. + self._task.cancel() + await asyncio.gather(self._task, return_exceptions=True) + + +class _AgentExitedError(Exception): + """The agent process ended before answering an outstanding ACP request.""" + + def __init__(self, returncode: int | None) -> None: + super().__init__(returncode) + self.returncode = returncode + + +async def _await_while_agent_runs(awaitable: Any, proc: asyncio.subprocess.Process) -> Any: + """Await an ACP exchange, failing fast if the agent process goes away. + + Older SDK releases never fail a pending request when the agent closes its + stdout, so a crash at startup would otherwise wait out the whole timeout. + Newer releases raise ``ConnectionError`` instead; both are reported as the + agent having exited, which is what the operator needs to know. + """ + exchange = asyncio.ensure_future(awaitable) + exited = asyncio.ensure_future(proc.wait()) + try: + await asyncio.wait({exchange, exited}, return_when=asyncio.FIRST_COMPLETED) + if not exchange.done(): + # A response may still sit in the stdout pipe behind the exit. + await asyncio.wait({exchange}, timeout=_EXIT_DRAIN_SECONDS) + if exchange.done() and not isinstance(exchange.exception(), ConnectionError): + return exchange.result() + await asyncio.wait({exited}, timeout=_EXIT_DRAIN_SECONDS) + raise _AgentExitedError(proc.returncode) + finally: + exchange.cancel() + exited.cancel() + await asyncio.gather(exchange, exited, return_exceptions=True) + + def _format_invocation_error(agent: str, cmd: str, exc: Exception) -> str: """Return a user-facing ACP invocation error with actionable remediation.""" if not isinstance(exc, FileNotFoundError): @@ -293,39 +363,63 @@ def build_invoke_acp_agent_tool(agents: dict) -> BaseTool: ) mcp_servers = [] + async def _converse(conn) -> None: + await conn.initialize( + protocol_version=PROTOCOL_VERSION, + client_capabilities=ClientCapabilities(), + client_info=Implementation(name="deerflow", title="DeerFlow", version="0.1.0"), + ) + session_kwargs: dict[str, Any] = {"cwd": physical_cwd, "mcp_servers": mcp_servers} + if agent_config.model: + session_kwargs["model"] = agent_config.model + session = await conn.new_session(**session_kwargs) + await conn.prompt( + session_id=session.session_id, + prompt=[text_block(prompt)], + ) + + stderr_tail: _StderrTail | None = None + timed_out = False try: from acp import spawn_agent_process async with spawn_agent_process(client, cmd, *args, env=agent_env, cwd=physical_cwd) as (conn, proc): + stderr_tail = _StderrTail(proc.stderr) logger.info("Spawning ACP agent '%s' with command '%s' and args %s in cwd %s", agent, cmd, args, physical_cwd) try: async with asyncio.timeout(agent_config.timeout_seconds) as deadline: - await conn.initialize( - protocol_version=PROTOCOL_VERSION, - client_capabilities=ClientCapabilities(), - client_info=Implementation(name="deerflow", title="DeerFlow", version="0.1.0"), - ) - session_kwargs: dict[str, Any] = {"cwd": physical_cwd, "mcp_servers": mcp_servers} - if agent_config.model: - session_kwargs["model"] = agent_config.model - session = await conn.new_session(**session_kwargs) - await conn.prompt( - session_id=session.session_id, - prompt=[text_block(prompt)], - ) + await _await_while_agent_runs(_converse(conn), proc) except TimeoutError: if not deadline.expired(): raise + # Leave the SDK context first: it terminates and reaps the + # agent, so the lines a stalled agent wrote last reach the tail. + timed_out = True + except _AgentExitedError as exc: + status = f"exited with code {exc.returncode}" if exc.returncode is not None else "closed its ACP connection" logger.error( - "ACP agent '%s' timed out after %s seconds during initialization, session creation, or prompt; terminating subprocess", + "ACP agent '%s' %s before completing the ACP invocation. stderr tail:\n%s", agent, - agent_config.timeout_seconds, + status, + await stderr_tail.text(wait=_EXIT_DRAIN_SECONDS), ) return ( - f"Error: ACP agent '{agent}' timed out after {agent_config.timeout_seconds} seconds " - "without completing the ACP invocation. The agent subprocess has been terminated. If this agent handles " - f"long-running tasks, increase acp_agents.{agent}.timeout_seconds in config.yaml." + f"Error: ACP agent '{agent}' {status} before completing the ACP invocation. " + f"Check that the agent is installed and authenticated and that acp_agents.{agent} in config.yaml is correct; " + "the agent's stderr is recorded in the Gateway log." ) + if timed_out: + logger.error( + "ACP agent '%s' timed out after %s seconds during initialization, session creation, or prompt; subprocess terminated. stderr tail:\n%s", + agent, + agent_config.timeout_seconds, + await stderr_tail.text(wait=_EXIT_DRAIN_SECONDS), + ) + return ( + f"Error: ACP agent '{agent}' timed out after {agent_config.timeout_seconds} seconds " + "without completing the ACP invocation. The agent subprocess has been terminated. If this agent handles " + f"long-running tasks, increase acp_agents.{agent}.timeout_seconds in config.yaml." + ) result = client.collected_text logger.info("ACP agent '%s' returned %s", agent, result[:1000]) logger.info("ACP agent '%s' returned %d characters", agent, len(result)) @@ -335,6 +429,9 @@ def build_invoke_acp_agent_tool(agents: dict) -> BaseTool: # Report the configured command, not the resolved path: remediation # text has to match what the operator wrote in config.yaml. return await asyncio.to_thread(_format_invocation_error, agent, agent_config.command, e) + finally: + if stderr_tail is not None: + await stderr_tail.aclose() return StructuredTool.from_function( name="invoke_acp_agent", diff --git a/backend/tests/blocking_io/test_invoke_acp_agent_tool.py b/backend/tests/blocking_io/test_invoke_acp_agent_tool.py index b949de1f1e..6b7e0c50fc 100644 --- a/backend/tests/blocking_io/test_invoke_acp_agent_tool.py +++ b/backend/tests/blocking_io/test_invoke_acp_agent_tool.py @@ -62,7 +62,8 @@ async def test_invoke_acp_agent_setup_does_not_block_event_loop(monkeypatch, tmp @contextlib.asynccontextmanager async def fake_spawn_agent_process(client, command, *args, env=None, cwd=None): captured["cwd"] = cwd - yield _Connection(), SimpleNamespace() + # An agent process that stays alive for the whole exchange. + yield _Connection(), SimpleNamespace(stderr=None, wait=asyncio.Event().wait) monkeypatch.setattr(acp, "spawn_agent_process", fake_spawn_agent_process) diff --git a/backend/tests/test_invoke_acp_agent_tool.py b/backend/tests/test_invoke_acp_agent_tool.py index 54a32cb7d3..9292f68a72 100644 --- a/backend/tests/test_invoke_acp_agent_tool.py +++ b/backend/tests/test_invoke_acp_agent_tool.py @@ -2,6 +2,7 @@ import asyncio import contextlib +import logging import os import sys import time @@ -26,6 +27,16 @@ from deerflow.tools.builtins.invoke_acp_agent_tool import ( from deerflow.tools.tools import get_available_tools +class _LiveAgentProcess: + """Process double for an agent that stays alive for the whole ACP exchange.""" + + stderr = None + returncode = None + + async def wait(self) -> int: + await asyncio.Event().wait() + + def test_build_mcp_servers_filters_disabled_and_maps_transports(): set_extensions_config(ExtensionsConfig(mcp_servers={"stale": McpServerConfig(enabled=True, type="stdio", command="echo")}, skills={})) fresh_config = ExtensionsConfig( @@ -259,7 +270,7 @@ async def test_invoke_acp_agent_uses_fixed_acp_workspace(monkeypatch, tmp_path): captured["spawn"] = {"cmd": cmd, "args": list(args), "cwd": cwd} async def __aenter__(self): - return DummyConn(), object() + return DummyConn(), _LiveAgentProcess() async def __aexit__(self, exc_type, exc, tb): return False @@ -384,7 +395,7 @@ async def test_invoke_acp_agent_uses_per_thread_workspace_when_thread_id_in_conf captured["cwd"] = cwd async def __aenter__(self): - return DummyConn(), object() + return DummyConn(), _LiveAgentProcess() async def __aexit__(self, exc_type, exc, tb): return False @@ -477,7 +488,7 @@ async def test_invoke_acp_agent_passes_env_to_spawn(monkeypatch, tmp_path): captured["env"] = env async def __aenter__(self): - return DummyConn(), object() + return DummyConn(), _LiveAgentProcess() async def __aexit__(self, exc_type, exc, tb): return False @@ -570,7 +581,7 @@ async def test_invoke_acp_agent_skips_invalid_mcp_servers(monkeypatch, tmp_path, captured["spawn"] = {"cmd": cmd, "args": list(args), "env": env, "cwd": cwd} async def __aenter__(self): - return DummyConn(), object() + return DummyConn(), _LiveAgentProcess() async def __aexit__(self, exc_type, exc, tb): return False @@ -657,7 +668,7 @@ async def test_invoke_acp_agent_passes_none_env_when_not_configured(monkeypatch, captured["env"] = env async def __aenter__(self): - return DummyConn(), object() + return DummyConn(), _LiveAgentProcess() async def __aexit__(self, exc_type, exc, tb): return False @@ -770,7 +781,7 @@ def test_get_available_tools_sync_invoke_acp_agent_preserves_thread_workspace(mo captured["cwd"] = cwd async def __aenter__(self): - return DummyConn(), object() + return DummyConn(), _LiveAgentProcess() async def __aexit__(self, exc_type, exc, tb): return False @@ -1003,6 +1014,118 @@ async def test_invoke_acp_agent_preserves_sdk_timeout_errors(acp_subprocess_tool assert acp_subprocess_tool.captured["proc"].returncode is not None +_STDERR_ACP_AGENT_SCRIPT = """\ +import asyncio +import os +import sys +import time + +import acp +from acp.schema import InitializeResponse, NewSessionResponse, PromptResponse + +# argv: <"exit" or "hang"> + + +class _TestAgent: + def on_connect(self, conn): + self._conn = conn + + def _enter_phase(self, phase): + line = f"{phase}: " + "x" * 200 + "\\n" + for _ in range(int(sys.argv[2]) // len(line) + 1): + sys.stderr.write(line) + if phase == sys.argv[1] and sys.argv[3] == "exit": + sys.stderr.write(f"fatal: cannot {phase} without credentials\\n") + sys.stderr.flush() + os._exit(3) + if phase == sys.argv[1]: + sys.stderr.write(f"stalled: waiting for {phase} login\\n") + sys.stderr.flush() + time.sleep(3600) + sys.stderr.flush() + + async def initialize(self, protocol_version, client_capabilities=None, client_info=None, **kwargs): + self._enter_phase("initialize") + return InitializeResponse(protocol_version=protocol_version) + + async def new_session(self, cwd, additional_directories=None, mcp_servers=None, **kwargs): + self._enter_phase("new_session") + return NewSessionResponse(session_id="test-session") + + async def prompt(self, session_id, prompt, **kwargs): + self._enter_phase("prompt") + await self._conn.session_update(session_id=session_id, update=acp.update_agent_message_text("done")) + return PromptResponse(stop_reason="end_turn") + + +asyncio.run(acp.run_agent(_TestAgent())) +""" + + +@pytest.fixture +def acp_stderr_tool(monkeypatch, tmp_path): + from deerflow.config import paths as paths_module + + monkeypatch.setattr(paths_module, "get_paths", lambda: paths_module.Paths(base_dir=tmp_path)) + monkeypatch.setattr( + "deerflow.config.extensions_config.ExtensionsConfig.from_file", + classmethod(lambda cls: ExtensionsConfig(mcp_servers={}, skills={})), + ) + script_path = tmp_path / "stderr_acp_agent.py" + script_path.write_text(_STDERR_ACP_AGENT_SCRIPT, encoding="utf-8") + + def build_tool(stop_phase="", stderr_bytes=0, stop="exit", timeout_seconds=60): + return build_invoke_acp_agent_tool( + { + "test": ACPAgentConfig( + command=sys.executable, + args=[str(script_path), stop_phase, str(stderr_bytes), stop], + description="Test ACP agent", + timeout_seconds=timeout_seconds, + ) + } + ) + + return build_tool + + +@pytest.mark.anyio +@pytest.mark.parametrize("phase", ["initialize", "new_session", "prompt"]) +async def test_invoke_acp_agent_reports_an_agent_that_exits_mid_invocation(acp_stderr_tool, caplog, phase): + # More stderr than the retained tail, so the fatal line (written last) is + # only logged if the tail keeps the end of the stream rather than its start. + tool = acp_stderr_tool(stop_phase=phase, stderr_bytes=40 * 1024) + start = time.monotonic() + with caplog.at_level(logging.ERROR, logger="deerflow.tools.builtins.invoke_acp_agent_tool"): + # The tool's own 60s budget must not be what ends the call. + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=30) + assert time.monotonic() - start < 15 + assert result.startswith("Error: ACP agent 'test' exited with code 3 before completing the ACP invocation.") + assert "timeout_seconds" not in result + # The agent's diagnostics reach the operator log, not the model. + assert f"fatal: cannot {phase} without credentials" in caplog.text + assert "credentials" not in result + + +@pytest.mark.anyio +async def test_invoke_acp_agent_logs_the_stderr_of_a_stalled_agent(acp_stderr_tool, caplog): + tool = acp_stderr_tool(stop_phase="prompt", stderr_bytes=40 * 1024, stop="hang", timeout_seconds=3) + with caplog.at_level(logging.ERROR, logger="deerflow.tools.builtins.invoke_acp_agent_tool"): + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=30) + assert "timed out after 3 seconds" in result + # The last line the agent wrote before stalling is the one an operator needs. + assert "stalled: waiting for prompt login" in caplog.text + assert "login" not in result + + +@pytest.mark.anyio +async def test_invoke_acp_agent_drains_verbose_stderr(acp_stderr_tool): + # Far beyond what an unread stderr pipe absorbs before the agent's writes block. + tool = acp_stderr_tool(stderr_bytes=1024 * 1024, timeout_seconds=20) + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=40) + assert result == "done" + + def _write_launcher_shim(directory: Path, name: str, command: str) -> Path: """Write an executable launcher shim for ``name`` under ``directory``.""" directory.mkdir(parents=True, exist_ok=True)