From 420b39ed903dc7ccd868ac6321a610df7fc4f407 Mon Sep 17 00:00:00 2001 From: quifox <289420841+quifox@users.noreply.github.com> Date: Mon, 5 Oct 2026 14:39:59 +0800 Subject: [PATCH] fix(acp): apply invocation timeout to initialization and session creation (#6292) * fix(acp): apply invocation timeout to initialization and session creation * fix(acp): preserve SDK timeout errors before the invocation deadline --- README.md | 1 + .../harness/deerflow/config/acp_config.py | 9 +- .../packages/harness/deerflow/tools/AGENTS.md | 1 + .../tools/builtins/invoke_acp_agent_tool.py | 32 ++-- backend/tests/test_invoke_acp_agent_tool.py | 175 ++++++++++++------ 5 files changed, 138 insertions(+), 80 deletions(-) diff --git a/README.md b/README.md index e153384814..5c7918a9fb 100644 --- a/README.md +++ b/README.md @@ -312,6 +312,7 @@ For Google's official Gemini OpenAI-compatible endpoint, use the - The Codex model provider returns completed responses without waiting for the SSE connection to close. Failed or incomplete responses report the provider's error or reason; partial output is not returned as a successful answer. Non-object error details are reported as text. - Claude Code accepts `CLAUDE_CODE_OAUTH_TOKEN`, `ANTHROPIC_AUTH_TOKEN`, `CLAUDE_CODE_CREDENTIALS_PATH`, or `~/.claude/.credentials.json` - 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` + - 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. - MiniMax Code speaks ACP directly. Install and authenticate it, then add it as an ACP agent: ```bash diff --git a/backend/packages/harness/deerflow/config/acp_config.py b/backend/packages/harness/deerflow/config/acp_config.py index 7b2021e6f5..03376363a5 100644 --- a/backend/packages/harness/deerflow/config/acp_config.py +++ b/backend/packages/harness/deerflow/config/acp_config.py @@ -30,11 +30,10 @@ class ACPAgentConfig(BaseModel): default=1800, ge=1, description=( - "Maximum time in seconds to wait for the agent to respond to a single invoke_acp_agent " - "call before the invocation is aborted and the subprocess is terminated. Mirrors " - "subagents.timeout_seconds (default: 1800 = 30 minutes) — without this backstop, an ACP " - "agent subprocess that hangs after initialize/new_session blocks the tool call, and " - "therefore the whole agent turn, indefinitely." + "Shared time budget in seconds for initialization, session creation, and prompt during one " + "invoke_acp_agent call, starting after the subprocess is launched. On timeout, the invocation " + "is aborted and the subprocess is terminated. Workspace/MCP preparation and subprocess cleanup " + "are outside this budget. Mirrors subagents.timeout_seconds (default: 1800 = 30 minutes)." ), ) diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md index f196be86ca..acb347035c 100644 --- a/backend/packages/harness/deerflow/tools/AGENTS.md +++ b/backend/packages/harness/deerflow/tools/AGENTS.md @@ -74,6 +74,7 @@ E2B output sync records remote file versions and actual host file metadata in a - `invoke_acp_agent` - Invokes external ACP-compatible agents from `config.yaml` - 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`. - 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]` - Each ACP agent uses a per-thread workspace at `{base_dir}/users/{user_id}/threads/{thread_id}/acp-workspace/`. The workspace is accessible to the lead agent via the virtual path `/mnt/acp-workspace/` (read-only). In docker sandbox mode, the directory is volume-mounted into the container at `/mnt/acp-workspace` (read-only); in local sandbox mode, path translation is handled by `tools.py` 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 d7e32314e6..472a0f32da 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 @@ -234,32 +234,32 @@ def build_invoke_acp_agent_tool(agents: dict) -> BaseTool: async with spawn_agent_process(client, cmd, *args, env=agent_env, cwd=physical_cwd) as (conn, proc): logger.info("Spawning ACP agent '%s' with command '%s' and args %s in cwd %s", agent, cmd, args, physical_cwd) - 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) try: - await asyncio.wait_for( - conn.prompt( + 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)], - ), - timeout=agent_config.timeout_seconds, - ) + ) except TimeoutError: + if not deadline.expired(): + raise logger.error( - "ACP agent '%s' timed out after %s seconds without responding to prompt; terminating subprocess", + "ACP agent '%s' timed out after %s seconds during initialization, session creation, or prompt; terminating subprocess", agent, agent_config.timeout_seconds, ) return ( f"Error: ACP agent '{agent}' timed out after {agent_config.timeout_seconds} seconds " - "without responding. The agent subprocess has been terminated. If this agent handles " + "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 diff --git a/backend/tests/test_invoke_acp_agent_tool.py b/backend/tests/test_invoke_acp_agent_tool.py index 186e7b339d..5754fd5b7e 100644 --- a/backend/tests/test_invoke_acp_agent_tool.py +++ b/backend/tests/test_invoke_acp_agent_tool.py @@ -837,55 +837,50 @@ def test_get_available_tools_uses_explicit_app_config_for_acp_agents(monkeypatch assert "invoke_acp_agent" in [tool.name for tool in tools] -# --------------------------------------------------------------------------- -# Regression: invoke_acp_agent must not hang forever on a stuck prompt() call -# --------------------------------------------------------------------------- -# -# A minimal, real ACP agent subprocess that answers `initialize`/`new_session` -# correctly but then hangs forever inside `prompt()` (never responds). This -# reproduces an agent that has finished the ACP handshake but then wedges — -# e.g. stuck on a runaway internal step — instead of a mocked `acp` module, -# so the regression test below exercises the real spawn/kill subprocess path. -_HUNG_ACP_AGENT_SCRIPT = """\ +# A real ACP subprocess can stop responding at any protocol phase. Keep the +# SDK's transport and cleanup intact so these tests detect leaked child processes. +_ACP_AGENT_SCRIPT = """\ import asyncio +import sys +from pathlib import Path import acp -from acp.schema import InitializeResponse, NewSessionResponse +from acp.schema import InitializeResponse, NewSessionResponse, PromptResponse -class _HungAgent: +class _TestAgent: + def on_connect(self, conn): + self._conn = conn + + async def _enter_phase(self, phase): + Path(sys.argv[3]).write_text(phase, encoding="utf-8") + if phase == "prompt": + await self._conn.session_update(session_id="test-session", update=acp.update_agent_thought_text(phase)) + if phase == sys.argv[1]: + await asyncio.Event().wait() + delay = float(sys.argv[2]) + if delay: + await asyncio.sleep(delay) + async def initialize(self, protocol_version, client_capabilities=None, client_info=None, **kwargs): + await self._enter_phase("initialize") return InitializeResponse(protocol_version=protocol_version) async def new_session(self, cwd, additional_directories=None, mcp_servers=None, **kwargs): - return NewSessionResponse(session_id="hung-session") + await self._enter_phase("new_session") + return NewSessionResponse(session_id="test-session") async def prompt(self, session_id, prompt, **kwargs): - # Deliberately never respond: simulates an ACP agent that completes the - # handshake but then hangs instead of answering session/prompt. - await asyncio.Event().wait() + await self._enter_phase("prompt") + return PromptResponse(stop_reason="end_turn") -asyncio.run(acp.run_agent(_HungAgent())) +asyncio.run(acp.run_agent(_TestAgent())) """ -@pytest.mark.anyio -async def test_invoke_acp_agent_times_out_and_kills_hung_subprocess(monkeypatch, tmp_path): - """invoke_acp_agent must time out and kill the subprocess instead of hanging - forever when the agent answers initialize/new_session but then never - responds to session/prompt. - - Before the timeout_seconds fix, neither this tool nor ACPAgentConfig had - any timeout, so this exact scenario blocked the tool call — and therefore - the whole agent turn — indefinitely, with the child process left running. - - `timeout_seconds` is configured small (2s) so the pass-after run completes - in a couple of seconds. The outer `asyncio.wait_for(..., timeout=20)` is - only a test-level safety net: it must never fire in the pass-after case - (elapsed stays well under it), but bounds this test to ~20s instead of - hanging the whole suite forever if the fix regresses. - """ +@pytest.fixture +def acp_subprocess_tool(monkeypatch, tmp_path): import acp as acp_module from deerflow.config import paths as paths_module @@ -895,43 +890,105 @@ async def test_invoke_acp_agent_times_out_and_kills_hung_subprocess(monkeypatch, "deerflow.config.extensions_config.ExtensionsConfig.from_file", classmethod(lambda cls: ExtensionsConfig(mcp_servers={}, skills={})), ) - - script_path = tmp_path / "hung_acp_agent.py" - script_path.write_text(_HUNG_ACP_AGENT_SCRIPT, encoding="utf-8") - - captured: dict[str, object] = {} + script_path = tmp_path / "test_acp_agent.py" + script_path.write_text(_ACP_AGENT_SCRIPT, encoding="utf-8") + phase_path = tmp_path / "phase.txt" + captured = {} + phase_entered = asyncio.Event() real_spawn_agent_process = acp_module.spawn_agent_process - # Spy on the real spawn_agent_process (not a fake) so we can inspect the - # actual asyncio.subprocess.Process afterwards, while every bit of real - # spawn/handshake/cleanup behavior stays exactly as production uses it. @contextlib.asynccontextmanager async def _spying_spawn_agent_process(client, cmd, *args, env=None, cwd=None): + original_session_update = client.session_update + + async def _session_update(session_id, update, **kwargs): + if getattr(update, "session_update", None) == "agent_thought_chunk": + phase_entered.set() + await original_session_update(session_id, update, **kwargs) + + client.session_update = _session_update async with real_spawn_agent_process(client, cmd, *args, env=env, cwd=cwd) as (conn, proc): captured["proc"] = proc yield conn, proc monkeypatch.setattr(acp_module, "spawn_agent_process", _spying_spawn_agent_process) - tool = build_invoke_acp_agent_tool( - { - "hung": ACPAgentConfig( - command=sys.executable, - args=[str(script_path)], - description="Hung test agent", - timeout_seconds=2, - ) - } - ) + def build_tool(hang_phase="", phase_delay=0, timeout_seconds=2): + tool = build_invoke_acp_agent_tool( + { + "test": ACPAgentConfig( + command=sys.executable, + args=[str(script_path), hang_phase, str(phase_delay), str(phase_path)], + description="Test ACP agent", + timeout_seconds=timeout_seconds, + ) + } + ) + return tool + return SimpleNamespace(build_tool=build_tool, captured=captured, phase_entered=phase_entered, phase_path=phase_path) + + +@pytest.mark.anyio +@pytest.mark.parametrize("phase", ["initialize", "new_session", "prompt"]) +async def test_invoke_acp_agent_times_out_and_kills_hung_subprocess(acp_subprocess_tool, phase): + tool = acp_subprocess_tool.build_tool(hang_phase=phase) start = time.monotonic() - result = await asyncio.wait_for(tool.coroutine(agent="hung", prompt="do work"), timeout=20) - elapsed = time.monotonic() - start - - assert elapsed < 10, f"expected the configured 2s timeout to fire quickly, took {elapsed:.1f}s" + # The outer timeout only bounds a regression; it must not supply the tool's deadline. + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=15) + assert time.monotonic() - start < 10 assert "timed out" in result.lower() - assert "hung" in result + assert "test" in result + assert acp_subprocess_tool.phase_path.read_text(encoding="utf-8") == phase + assert acp_subprocess_tool.captured["proc"].returncode is not None - proc = captured.get("proc") - assert proc is not None, "spawn_agent_process spy did not capture the subprocess" - assert proc.returncode is not None, "subprocess must be terminated, not left running after a timeout" + +@pytest.mark.anyio +async def test_invoke_acp_agent_shares_one_timeout_budget(acp_subprocess_tool): + # Each phase completes within 3s, but their combined 3.6s exceeds one invocation budget. + tool = acp_subprocess_tool.build_tool(phase_delay=1.2, timeout_seconds=3) + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=15) + assert "timed out" in result.lower() + assert acp_subprocess_tool.captured["proc"].returncode is not None + + +@pytest.mark.anyio +async def test_invoke_acp_agent_completes_within_timeout(acp_subprocess_tool): + tool = acp_subprocess_tool.build_tool() + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=15) + assert result == "(no response)" + assert acp_subprocess_tool.phase_path.read_text(encoding="utf-8") == "prompt" + assert acp_subprocess_tool.captured["proc"].returncode is not None + + +@pytest.mark.anyio +async def test_invoke_acp_agent_cancellation_closes_subprocess(acp_subprocess_tool): + tool = acp_subprocess_tool.build_tool(hang_phase="prompt", timeout_seconds=60) + invocation = asyncio.create_task(tool.coroutine(agent="test", prompt="do work")) + try: + await asyncio.wait_for(acp_subprocess_tool.phase_entered.wait(), timeout=15) + invocation.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(invocation, timeout=15) + assert acp_subprocess_tool.captured["proc"].returncode is not None + finally: + if not invocation.done(): + invocation.cancel() + with contextlib.suppress(asyncio.CancelledError): + await invocation + + +@pytest.mark.anyio +@pytest.mark.parametrize("phase", ["initialize", "new_session", "prompt"]) +async def test_invoke_acp_agent_preserves_sdk_timeout_errors(acp_subprocess_tool, monkeypatch, phase): + from acp.client.connection import ClientSideConnection + + async def raise_sdk_timeout(self, *args, **kwargs): + raise TimeoutError("SDK transport timed out") + + monkeypatch.setattr(ClientSideConnection, phase, raise_sdk_timeout) + tool = acp_subprocess_tool.build_tool(timeout_seconds=60) + result = await asyncio.wait_for(tool.coroutine(agent="test", prompt="do work"), timeout=15) + assert result == "Error invoking ACP agent 'test': SDK transport timed out" + assert "timeout_seconds" not in result + assert acp_subprocess_tool.captured["proc"].returncode is not None