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)