"""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)