mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
fix(harness): add timeout to invoke_acp_agent to prevent indefinite hangs (#4238)
invoke_acp_agent had no timeout anywhere in its call path, and ACPAgentConfig had no timeout field. If the ACP agent subprocess answers initialize/new_session correctly but then hangs inside prompt(), the tool call - and therefore the whole agent turn - blocks indefinitely, with the child process left running. MCP stdio servers already guard against this class of hang via tool_call_timeout; ACP agent invocations had no equivalent. Add ACPAgentConfig.timeout_seconds (default 1800, ge=1), mirroring the shape/default of subagents.timeout_seconds, and wrap the conn.prompt() call in asyncio.wait_for(). On TimeoutError, return a clear error instead of hanging; exiting the spawn_agent_process context block triggers the ACP library's own graceful-then-forceful subprocess cleanup, so the hung process is actually terminated.
This commit is contained in:
parent
01a89f2379
commit
09d9cf53d2
@ -24,6 +24,17 @@ class ACPAgentConfig(BaseModel):
|
|||||||
"are denied — the agent must be configured to operate without requesting permissions."
|
"are denied — the agent must be configured to operate without requesting permissions."
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
timeout_seconds: int = Field(
|
||||||
|
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."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
_acp_agents: dict[str, ACPAgentConfig] = {}
|
_acp_agents: dict[str, ACPAgentConfig] = {}
|
||||||
|
|||||||
@ -1,5 +1,6 @@
|
|||||||
"""Built-in tool for invoking external ACP-compatible agents."""
|
"""Built-in tool for invoking external ACP-compatible agents."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import shutil
|
import shutil
|
||||||
@ -237,10 +238,25 @@ def build_invoke_acp_agent_tool(agents: dict) -> BaseTool:
|
|||||||
if agent_config.model:
|
if agent_config.model:
|
||||||
session_kwargs["model"] = agent_config.model
|
session_kwargs["model"] = agent_config.model
|
||||||
session = await conn.new_session(**session_kwargs)
|
session = await conn.new_session(**session_kwargs)
|
||||||
await conn.prompt(
|
try:
|
||||||
session_id=session.session_id,
|
await asyncio.wait_for(
|
||||||
prompt=[text_block(prompt)],
|
conn.prompt(
|
||||||
)
|
session_id=session.session_id,
|
||||||
|
prompt=[text_block(prompt)],
|
||||||
|
),
|
||||||
|
timeout=agent_config.timeout_seconds,
|
||||||
|
)
|
||||||
|
except TimeoutError:
|
||||||
|
logger.error(
|
||||||
|
"ACP agent '%s' timed out after %s seconds without responding to 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 "
|
||||||
|
f"long-running tasks, increase acp_agents.{agent}.timeout_seconds in config.yaml."
|
||||||
|
)
|
||||||
result = client.collected_text
|
result = client.collected_text
|
||||||
logger.info("ACP agent '%s' returned %s", agent, result[:1000])
|
logger.info("ACP agent '%s' returned %s", agent, result[:1000])
|
||||||
logger.info("ACP agent '%s' returned %d characters", agent, len(result))
|
logger.info("ACP agent '%s' returned %d characters", agent, len(result))
|
||||||
|
|||||||
@ -105,6 +105,38 @@ def test_acp_agent_config_auto_approve_permissions():
|
|||||||
assert cfg.auto_approve_permissions is True
|
assert cfg.auto_approve_permissions is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_acp_agent_config_timeout_seconds_default():
|
||||||
|
"""Default mirrors subagents.timeout_seconds (1800s = 30 min) so an ACP agent
|
||||||
|
subprocess that hangs after initialize/new_session cannot block the tool call
|
||||||
|
(and therefore the whole agent turn) indefinitely."""
|
||||||
|
cfg = ACPAgentConfig(command="my-agent", description="desc")
|
||||||
|
assert cfg.timeout_seconds == 1800
|
||||||
|
|
||||||
|
|
||||||
|
def test_acp_agent_config_timeout_seconds_override():
|
||||||
|
cfg = ACPAgentConfig(command="my-agent", description="desc", timeout_seconds=45)
|
||||||
|
assert cfg.timeout_seconds == 45
|
||||||
|
|
||||||
|
|
||||||
|
def test_acp_agent_config_timeout_seconds_rejects_non_positive():
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
ACPAgentConfig(command="my-agent", description="desc", timeout_seconds=0)
|
||||||
|
|
||||||
|
|
||||||
|
def test_load_acp_config_preserves_timeout_seconds():
|
||||||
|
load_acp_config_from_dict(
|
||||||
|
{
|
||||||
|
"codex": {
|
||||||
|
"command": "codex-acp",
|
||||||
|
"args": [],
|
||||||
|
"description": "Codex CLI",
|
||||||
|
"timeout_seconds": 60,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
)
|
||||||
|
assert get_acp_agents()["codex"].timeout_seconds == 60
|
||||||
|
|
||||||
|
|
||||||
def test_acp_agent_config_missing_command_raises():
|
def test_acp_agent_config_missing_command_raises():
|
||||||
with pytest.raises(ValidationError):
|
with pytest.raises(ValidationError):
|
||||||
ACPAgentConfig(description="No command provided")
|
ACPAgentConfig(description="No command provided")
|
||||||
|
|||||||
@ -1,6 +1,9 @@
|
|||||||
"""Tests for the built-in ACP invocation tool."""
|
"""Tests for the built-in ACP invocation tool."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import contextlib
|
||||||
import sys
|
import sys
|
||||||
|
import time
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@ -813,3 +816,103 @@ def test_get_available_tools_uses_explicit_app_config_for_acp_agents(monkeypatch
|
|||||||
|
|
||||||
assert captured["agents"] is explicit_agents
|
assert captured["agents"] is explicit_agents
|
||||||
assert "invoke_acp_agent" in [tool.name for tool in tools]
|
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 = """\
|
||||||
|
import asyncio
|
||||||
|
|
||||||
|
import acp
|
||||||
|
from acp.schema import InitializeResponse, NewSessionResponse
|
||||||
|
|
||||||
|
|
||||||
|
class _HungAgent:
|
||||||
|
async def initialize(self, protocol_version, client_capabilities=None, client_info=None, **kwargs):
|
||||||
|
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")
|
||||||
|
|
||||||
|
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()
|
||||||
|
|
||||||
|
|
||||||
|
asyncio.run(acp.run_agent(_HungAgent()))
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
|
@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.
|
||||||
|
"""
|
||||||
|
import acp as acp_module
|
||||||
|
|
||||||
|
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 / "hung_acp_agent.py"
|
||||||
|
script_path.write_text(_HUNG_ACP_AGENT_SCRIPT, encoding="utf-8")
|
||||||
|
|
||||||
|
captured: dict[str, object] = {}
|
||||||
|
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):
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
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"
|
||||||
|
assert "timed out" in result.lower()
|
||||||
|
assert "hung" in result
|
||||||
|
|
||||||
|
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"
|
||||||
|
|||||||
@ -1445,6 +1445,7 @@ sandbox:
|
|||||||
# description: Claude Code for implementation, refactoring, and debugging
|
# description: Claude Code for implementation, refactoring, and debugging
|
||||||
# model: null
|
# model: null
|
||||||
# # auto_approve_permissions: false # Set to true to auto-approve ACP permission requests
|
# # auto_approve_permissions: false # Set to true to auto-approve ACP permission requests
|
||||||
|
# # timeout_seconds: 1800 # Abort + kill the subprocess if it doesn't respond in time (default: 1800 = 30 min)
|
||||||
# # env: # Optional: inject environment variables into the agent subprocess
|
# # env: # Optional: inject environment variables into the agent subprocess
|
||||||
# # ANTHROPIC_API_KEY: $ANTHROPIC_API_KEY # $VAR resolves from host environment
|
# # ANTHROPIC_API_KEY: $ANTHROPIC_API_KEY # $VAR resolves from host environment
|
||||||
#
|
#
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user