303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""Unit tests for ReActEngine autonomy mode (IQ-Boost U2, R6-R9).
|
|
|
|
Verifies the PLAN_EXEC autonomy gate:
|
|
- Dangerous tools (per DangerousToolsConfig) trigger confirmation_request
|
|
BEFORE first execution in autonomy mode.
|
|
- Non-dangerous tools execute directly without confirmation.
|
|
- Non-autonomy mode preserves existing behavior (gate is a no-op).
|
|
- Disabled whitelist (enabled=False) allows all tools.
|
|
- Approved dangerous tool re-executes with _skip_dangerous_check=True.
|
|
- Rejected dangerous tool returns permission_denied result.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
|
|
from agentkit.core.react import ReActEngine
|
|
from agentkit.server.config import DangerousToolsConfig
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _check_autonomy_gate — backward compatibility (no-op when disabled)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutonomyGateBackwardCompat:
|
|
"""When autonomy_mode=False or dangerous_tools_config=None, the gate is a
|
|
transparent no-op — existing callers see no behavior change."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_autonomy_mode_returns_should_execute(self):
|
|
"""Default engine (autonomy_mode=False) → gate passes through."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=False,
|
|
)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/tmp/x"}, [], step=1, confirmation_handler=None
|
|
)
|
|
assert should_exec is True
|
|
assert events == []
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_dangerous_config_returns_should_execute(self):
|
|
"""autonomy_mode=True but no config → gate passes through (no whitelist
|
|
to check against)."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=None,
|
|
autonomy_mode=True,
|
|
)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/tmp/x"}, [], step=1, confirmation_handler=None
|
|
)
|
|
assert should_exec is True
|
|
assert events == []
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_whitelist_allows_all(self):
|
|
"""enabled=False → is_dangerous always returns False → all tools pass."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(enabled=False),
|
|
autonomy_mode=True,
|
|
)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/etc"}, [], step=1, confirmation_handler=None
|
|
)
|
|
assert should_exec is True
|
|
assert events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _check_autonomy_gate — non-dangerous tool in autonomy mode
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutonomyGateNonDangerous:
|
|
"""In autonomy mode, non-dangerous tools execute without confirmation."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_non_dangerous_tool_passes_through(self):
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"read_file", {"path": "/tmp/x"}, [], step=1, confirmation_handler=None
|
|
)
|
|
assert should_exec is True
|
|
assert events == []
|
|
assert result is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_custom_pattern_non_match_passes(self):
|
|
"""Tool not matching any custom pattern passes through."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(tool_patterns=[r"^rm$", r"^kubectl"]),
|
|
autonomy_mode=True,
|
|
)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"search", {"query": "foo"}, [], step=1, confirmation_handler=None
|
|
)
|
|
assert should_exec is True
|
|
assert events == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _check_autonomy_gate — dangerous tool triggers confirmation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutonomyGateDangerous:
|
|
"""Dangerous tools in autonomy mode trigger confirmation_request before
|
|
execution."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_emits_confirmation_request(self):
|
|
"""rm matches ^rm$ pattern → confirmation_request event emitted."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
fake_tool = MagicMock()
|
|
fake_tool.safe_execute = AsyncMock(return_value={"output": "deleted"})
|
|
engine._find_tool = lambda name, tools: fake_tool
|
|
|
|
handler = AsyncMock(return_value=True)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/tmp/x"}, [fake_tool], step=1, confirmation_handler=handler
|
|
)
|
|
|
|
assert should_exec is False # Gate handled execution
|
|
assert len(events) == 2 # confirmation_request + confirmation_result
|
|
assert events[0].event_type == "confirmation_request"
|
|
assert events[0].data["tool_name"] == "rm"
|
|
assert events[0].data["confirmation_id"].startswith("autonomy:rm:")
|
|
assert events[1].event_type == "confirmation_result"
|
|
assert events[1].data["approved"] is True
|
|
# Tool executed with _skip_dangerous_check=True
|
|
fake_tool.safe_execute.assert_awaited_once_with(path="/tmp/x", _skip_dangerous_check=True)
|
|
assert result == {"output": "deleted"}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_rejected_returns_permission_denied(self):
|
|
"""User rejects → permission_denied result, tool NOT executed."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
fake_tool = MagicMock()
|
|
fake_tool.safe_execute = AsyncMock(return_value={"output": "should not run"})
|
|
engine._find_tool = lambda name, tools: fake_tool
|
|
|
|
handler = AsyncMock(return_value=False)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/tmp/x"}, [fake_tool], step=1, confirmation_handler=handler
|
|
)
|
|
|
|
assert should_exec is False
|
|
assert len(events) == 2
|
|
assert events[1].data["approved"] is False
|
|
fake_tool.safe_execute.assert_not_awaited()
|
|
assert result["error_type"] == "permission_denied"
|
|
assert result["exit_code"] == 126
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_no_handler_auto_rejects(self):
|
|
"""No confirmation_handler → auto-reject (safe default)."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._find_tool = lambda name, tools: MagicMock()
|
|
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"kubectl_delete", {"ns": "prod"}, [], step=2, confirmation_handler=None
|
|
)
|
|
|
|
assert should_exec is False
|
|
assert events[1].data["approved"] is False
|
|
assert result["error_type"] == "permission_denied"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_handler_exception_auto_rejects(self):
|
|
"""Handler raises → auto-reject (does not crash the engine)."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._find_tool = lambda name, tools: MagicMock()
|
|
|
|
async def failing_handler(cid, cmd, reason):
|
|
raise RuntimeError("handler crashed")
|
|
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"deploy", {"env": "prod"}, [], step=3, confirmation_handler=failing_handler
|
|
)
|
|
|
|
assert should_exec is False
|
|
assert events[1].data["approved"] is False
|
|
assert result["error_type"] == "permission_denied"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_payment_glob_pattern_matches(self):
|
|
"""payment_* pattern matches payment_charge tool."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
fake_tool = MagicMock()
|
|
fake_tool.safe_execute = AsyncMock(return_value={"ok": True})
|
|
engine._find_tool = lambda name, tools: fake_tool
|
|
|
|
handler = AsyncMock(return_value=True)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"payment_charge",
|
|
{"amount": 100},
|
|
[fake_tool],
|
|
step=1,
|
|
confirmation_handler=handler,
|
|
)
|
|
|
|
assert should_exec is False
|
|
assert events[0].event_type == "confirmation_request"
|
|
assert result == {"ok": True}
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_not_found_returns_error(self):
|
|
"""Approved but tool not found → error result."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._find_tool = lambda name, tools: None
|
|
|
|
handler = AsyncMock(return_value=True)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/x"}, [], step=1, confirmation_handler=handler
|
|
)
|
|
|
|
assert should_exec is False
|
|
assert "error" in result
|
|
assert "not found" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dangerous_tool_execution_failure_returns_error(self):
|
|
"""Approved but tool.safe_execute raises → error dict."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
fake_tool = MagicMock()
|
|
fake_tool.safe_execute = AsyncMock(side_effect=RuntimeError("boom"))
|
|
engine._find_tool = lambda name, tools: fake_tool
|
|
|
|
handler = AsyncMock(return_value=True)
|
|
should_exec, events, result = await engine._check_autonomy_gate(
|
|
"rm", {"path": "/x"}, [fake_tool], step=1, confirmation_handler=handler
|
|
)
|
|
|
|
assert should_exec is False
|
|
assert "error" in result
|
|
assert "tool_execution_failed" in result.get("error_code", "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Constructor wiring
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAutonomyConstructorWiring:
|
|
"""Verify __init__ stores autonomy state correctly."""
|
|
|
|
def test_defaults_are_off(self):
|
|
engine = ReActEngine(llm_gateway=MagicMock())
|
|
assert engine._autonomy_mode is False
|
|
assert engine._dangerous_tools_config is None
|
|
|
|
def test_autonomy_mode_stored(self):
|
|
cfg = DangerousToolsConfig()
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=cfg,
|
|
autonomy_mode=True,
|
|
)
|
|
assert engine._autonomy_mode is True
|
|
assert engine._dangerous_tools_config is cfg
|