mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-03 02:11:48 +00:00
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""Runtime tests for tool-call loop guardrails."""
|
|
|
|
import json
|
|
import uuid
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from run_agent import AIAgent
|
|
|
|
|
|
def _make_tool_defs(*names: str) -> list[dict]:
|
|
return [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": name,
|
|
"description": f"{name} tool",
|
|
"parameters": {"type": "object", "properties": {}},
|
|
},
|
|
}
|
|
for name in names
|
|
]
|
|
|
|
|
|
def _mock_tool_call(name="web_search", arguments="{}", call_id=None):
|
|
return SimpleNamespace(
|
|
id=call_id or f"call_{uuid.uuid4().hex[:8]}",
|
|
type="function",
|
|
function=SimpleNamespace(name=name, arguments=arguments),
|
|
)
|
|
|
|
|
|
def _mock_response(content="Hello", finish_reason="stop", tool_calls=None):
|
|
msg = SimpleNamespace(content=content, tool_calls=tool_calls)
|
|
choice = SimpleNamespace(message=msg, finish_reason=finish_reason)
|
|
return SimpleNamespace(choices=[choice], model="test/model", usage=None)
|
|
|
|
|
|
def _make_agent(*tool_names: str, max_iterations: int = 10, config: dict | None = None) -> AIAgent:
|
|
with (
|
|
patch("run_agent.get_tool_definitions", return_value=_make_tool_defs(*tool_names)),
|
|
patch("run_agent.check_toolset_requirements", return_value={}),
|
|
patch("hermes_cli.config.load_config", return_value=config or {}),
|
|
patch("run_agent.OpenAI"),
|
|
):
|
|
agent = AIAgent(
|
|
api_key="test-key-1234567890",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
max_iterations=max_iterations,
|
|
quiet_mode=True,
|
|
skip_context_files=True,
|
|
skip_memory=True,
|
|
)
|
|
agent.client = MagicMock()
|
|
agent._cached_system_prompt = "You are helpful."
|
|
agent._use_prompt_caching = False
|
|
agent.tool_delay = 0
|
|
agent.compression_enabled = False
|
|
agent.save_trajectories = False
|
|
return agent
|
|
|
|
|
|
def _seed_exact_failures(agent: AIAgent, tool_name: str, args: dict, count: int = 2) -> None:
|
|
for _ in range(count):
|
|
agent._tool_guardrails.after_call(
|
|
tool_name,
|
|
args,
|
|
json.dumps({"error": "boom"}),
|
|
failed=True,
|
|
)
|
|
|
|
|
|
def _hard_stop_config(**overrides) -> dict:
|
|
cfg = {
|
|
"tool_loop_guardrails": {
|
|
"warnings_enabled": True,
|
|
"hard_stop_enabled": True,
|
|
"hard_stop_after": {
|
|
"exact_failure": 2,
|
|
"same_tool_failure": 8,
|
|
"idempotent_no_progress": 5,
|
|
},
|
|
}
|
|
}
|
|
cfg["tool_loop_guardrails"].update(overrides)
|
|
return cfg
|
|
|
|
|
|
def test_default_sequential_path_warns_repeated_exact_failure_without_blocking_execution():
|
|
agent = _make_agent("web_search")
|
|
args = {"query": "same"}
|
|
_seed_exact_failures(agent, "web_search", args)
|
|
starts = []
|
|
progress = []
|
|
agent.tool_start_callback = lambda *a, **k: starts.append((a, k))
|
|
agent.tool_progress_callback = lambda *a, **k: progress.append((a, k))
|
|
tc = _mock_tool_call("web_search", json.dumps(args), "c-soft")
|
|
msg = SimpleNamespace(content="", tool_calls=[tc])
|
|
messages = []
|
|
|
|
with patch("run_agent.handle_function_call", return_value=json.dumps({"error": "boom"})) as mock_hfc:
|
|
agent._execute_tool_calls_sequential(msg, messages, "task-1")
|
|
|
|
mock_hfc.assert_called_once()
|
|
assert len(starts) == 1
|
|
assert any(event[0][0] == "tool.completed" for event in progress)
|
|
assert len(messages) == 1
|
|
assert messages[0]["role"] == "tool"
|
|
assert messages[0]["tool_call_id"] == "c-soft"
|
|
assert "repeated_exact_failure_warning" in messages[0]["content"]
|
|
assert "repeated_exact_failure_block" not in messages[0]["content"]
|
|
assert agent._tool_guardrail_halt_decision is None
|
|
|
|
|
|
def test_config_enabled_hard_stop_blocks_repeated_exact_failure_before_execution():
|
|
agent = _make_agent("web_search", config=_hard_stop_config())
|
|
args = {"query": "same"}
|
|
_seed_exact_failures(agent, "web_search", args)
|
|
starts = []
|
|
progress = []
|
|
agent.tool_start_callback = lambda *a, **k: starts.append((a, k))
|
|
agent.tool_progress_callback = lambda *a, **k: progress.append((a, k))
|
|
tc = _mock_tool_call("web_search", json.dumps(args), "c-block")
|
|
msg = SimpleNamespace(content="", tool_calls=[tc])
|
|
messages = []
|
|
|
|
with patch("run_agent.handle_function_call", return_value="SHOULD_NOT_RUN") as mock_hfc:
|
|
agent._execute_tool_calls_sequential(msg, messages, "task-1")
|
|
|
|
mock_hfc.assert_not_called()
|
|
assert starts == []
|
|
assert progress == []
|
|
assert len(messages) == 1
|
|
assert messages[0]["role"] == "tool"
|
|
assert messages[0]["tool_call_id"] == "c-block"
|
|
assert "repeated_exact_failure_block" in messages[0]["content"]
|
|
|
|
|
|
def test_sequential_after_call_appends_guidance_to_tool_result_without_extra_messages():
|
|
agent = _make_agent("web_search")
|
|
args = {"query": "same"}
|
|
_seed_exact_failures(agent, "web_search", args, count=1)
|
|
tc = _mock_tool_call("web_search", json.dumps(args), "c-warn")
|
|
msg = SimpleNamespace(content="", tool_calls=[tc])
|
|
messages = []
|
|
|
|
with patch("run_agent.handle_function_call", return_value=json.dumps({"error": "boom"})):
|
|
agent._execute_tool_calls_sequential(msg, messages, "task-1")
|
|
|
|
assert [m["role"] for m in messages] == ["tool"]
|
|
assert messages[0]["tool_call_id"] == "c-warn"
|
|
assert "Tool loop warning" in messages[0]["content"]
|
|
assert "repeated_exact_failure_warning" in messages[0]["content"]
|
|
|
|
|
|
def test_config_enabled_hard_stop_concurrent_path_does_not_submit_blocked_calls_and_preserves_result_order():
|
|
agent = _make_agent("web_search", config=_hard_stop_config())
|
|
blocked_args = {"query": "blocked"}
|
|
allowed_args = {"query": "allowed"}
|
|
_seed_exact_failures(agent, "web_search", blocked_args)
|
|
starts = []
|
|
progress_events = []
|
|
agent.tool_start_callback = lambda tool_call_id, name, args: starts.append((tool_call_id, name, args))
|
|
agent.tool_progress_callback = lambda event, name, preview, args, **kw: progress_events.append((event, name, args, kw))
|
|
calls = [
|
|
_mock_tool_call("web_search", json.dumps(blocked_args), "c-block"),
|
|
_mock_tool_call("web_search", json.dumps(allowed_args), "c-allow"),
|
|
]
|
|
msg = SimpleNamespace(content="", tool_calls=calls)
|
|
messages = []
|
|
executed = []
|
|
|
|
def fake_handle(name, args, task_id, **kwargs):
|
|
executed.append((name, args, kwargs["tool_call_id"]))
|
|
return json.dumps({"ok": args["query"]})
|
|
|
|
with patch("run_agent.handle_function_call", side_effect=fake_handle):
|
|
agent._execute_tool_calls_concurrent(msg, messages, "task-1")
|
|
|
|
assert executed == [("web_search", allowed_args, "c-allow")]
|
|
assert [m["tool_call_id"] for m in messages] == ["c-block", "c-allow"]
|
|
assert "repeated_exact_failure_block" in messages[0]["content"]
|
|
assert json.loads(messages[1]["content"]) == {"ok": "allowed"}
|
|
assert starts == [("c-allow", "web_search", allowed_args)]
|
|
started_events = [event for event in progress_events if event[0] == "tool.started"]
|
|
completed_events = [event for event in progress_events if event[0] == "tool.completed"]
|
|
assert started_events == [("tool.started", "web_search", allowed_args, {})]
|
|
assert len(completed_events) == 1
|
|
assert completed_events[0][1] == "web_search"
|
|
|
|
|
|
def test_plugin_pre_tool_block_wins_without_counting_as_toolguard_block():
|
|
agent = _make_agent("web_search")
|
|
args = {"query": "same"}
|
|
tc = _mock_tool_call("web_search", json.dumps(args), "c-plugin")
|
|
msg = SimpleNamespace(content="", tool_calls=[tc])
|
|
messages = []
|
|
|
|
with (
|
|
patch("hermes_cli.plugins.get_pre_tool_call_block_message", return_value="plugin policy"),
|
|
patch("run_agent.handle_function_call", return_value="SHOULD_NOT_RUN") as mock_hfc,
|
|
):
|
|
agent._execute_tool_calls_sequential(msg, messages, "task-1")
|
|
|
|
mock_hfc.assert_not_called()
|
|
assert "plugin policy" in messages[0]["content"]
|
|
assert agent._tool_guardrails.before_call("web_search", args).action == "allow"
|
|
|
|
|
|
def test_default_run_conversation_warns_without_guardrail_halt():
|
|
agent = _make_agent("web_search", max_iterations=10)
|
|
same_args = {"query": "same"}
|
|
responses = [
|
|
_mock_response(
|
|
content="",
|
|
finish_reason="tool_calls",
|
|
tool_calls=[_mock_tool_call("web_search", json.dumps(same_args), f"c{i}")],
|
|
)
|
|
for i in range(1, 4)
|
|
]
|
|
responses.append(_mock_response(content="done", finish_reason="stop", tool_calls=None))
|
|
agent.client.chat.completions.create.side_effect = responses
|
|
|
|
with (
|
|
patch("run_agent.handle_function_call", return_value=json.dumps({"error": "boom"})) as mock_hfc,
|
|
patch.object(agent, "_persist_session"),
|
|
patch.object(agent, "_save_trajectory"),
|
|
patch.object(agent, "_cleanup_task_resources"),
|
|
):
|
|
result = agent.run_conversation("search repeatedly")
|
|
|
|
assert mock_hfc.call_count == 3
|
|
assert result["turn_exit_reason"].startswith("text_response")
|
|
assert "guardrail" not in result
|
|
assert result["final_response"] == "done"
|
|
tool_contents = [m["content"] for m in result["messages"] if m.get("role") == "tool"]
|
|
assert any("repeated_exact_failure_warning" in content for content in tool_contents)
|
|
|
|
|
|
def test_config_enabled_hard_stop_run_conversation_returns_controlled_guardrail_halt_without_top_level_error():
|
|
agent = _make_agent("web_search", max_iterations=10, config=_hard_stop_config())
|
|
same_args = {"query": "same"}
|
|
responses = [
|
|
_mock_response(
|
|
content="",
|
|
finish_reason="tool_calls",
|
|
tool_calls=[_mock_tool_call("web_search", json.dumps(same_args), f"c{i}")],
|
|
)
|
|
for i in range(1, 10)
|
|
]
|
|
agent.client.chat.completions.create.side_effect = responses
|
|
|
|
with (
|
|
patch("run_agent.handle_function_call", return_value=json.dumps({"error": "boom"})) as mock_hfc,
|
|
patch.object(agent, "_persist_session"),
|
|
patch.object(agent, "_save_trajectory"),
|
|
patch.object(agent, "_cleanup_task_resources"),
|
|
):
|
|
result = agent.run_conversation("search repeatedly")
|
|
|
|
assert mock_hfc.call_count == 2
|
|
assert result["api_calls"] == 3
|
|
assert result["api_calls"] < agent.max_iterations
|
|
assert result["turn_exit_reason"] == "guardrail_halt"
|
|
assert "error" not in result
|
|
assert result["completed"] is True
|
|
assert "stopped retrying" in result["final_response"]
|
|
assert result["guardrail"]["code"] == "repeated_exact_failure_block"
|
|
assert result["guardrail"]["tool_name"] == "web_search"
|
|
|
|
assistant_tool_calls = [m for m in result["messages"] if m.get("role") == "assistant" and m.get("tool_calls")]
|
|
for assistant_msg in assistant_tool_calls:
|
|
call_ids = [tc["id"] for tc in assistant_msg["tool_calls"]]
|
|
following_results = [m for m in result["messages"] if m.get("role") == "tool" and m.get("tool_call_id") in call_ids]
|
|
assert len(following_results) == len(call_ids)
|