github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-27 05:30:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(acp): apply invocation timeout to initialization and session creation (#6292)

Browse Source 
* fix(acp): apply invocation timeout to initialization and session creation

* fix(acp): preserve SDK timeout errors before the invocation deadline
This commit is contained in:
quifox 2026-10-05 14:39:59 +08:00 committed by GitHub
parent b85e4ae74b
commit 420b39ed90
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 138 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -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

9
backend/packages/harness/deerflow/config/acp_config.py
View File

@ -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)."
),
)

1
backend/packages/harness/deerflow/tools/AGENTS.md
View File

@ -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`

32
backend/packages/harness/deerflow/tools/builtins/invoke_acp_agent_tool.py
View File

@ -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

175
backend/tests/test_invoke_acp_agent_tool.py
View File

@ -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