510 lines
20 KiB
Python
510 lines
20 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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U7: _track_tool_result_for_autonomy denied parameter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestTrackToolResultDenied:
|
|
"""U7: user denial (denied=True) does NOT increment the failure counter.
|
|
|
|
Verifies:
|
|
- denied=True leaves _consecutive_failures unchanged
|
|
- denied=False with an error result still increments (backward compat)
|
|
- denied=False with a permission_denied result still increments
|
|
- denied=True with a permission_denied result does NOT increment
|
|
"""
|
|
|
|
def test_denied_does_not_increment(self):
|
|
"""denied=True with an error result leaves the counter unchanged."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._consecutive_failures = 1
|
|
engine._track_tool_result_for_autonomy({"error": "boom", "is_error": True}, denied=True)
|
|
assert engine._consecutive_failures == 1 # Unchanged
|
|
|
|
def test_denied_permission_denied_does_not_increment(self):
|
|
"""denied=True with a permission_denied result leaves the counter
|
|
unchanged — this is the exact scenario U7 fixes (user rejected
|
|
confirmation)."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._consecutive_failures = 2
|
|
engine._track_tool_result_for_autonomy(
|
|
{
|
|
"output": "",
|
|
"exit_code": 126,
|
|
"is_error": True,
|
|
"error_type": "permission_denied",
|
|
"message": "用户拒绝执行命令",
|
|
},
|
|
denied=True,
|
|
)
|
|
assert engine._consecutive_failures == 2 # Unchanged
|
|
|
|
def test_not_denied_error_increments(self):
|
|
"""denied=False (default) with an error result still increments
|
|
(backward compat)."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._consecutive_failures = 1
|
|
engine._track_tool_result_for_autonomy({"error": "boom"})
|
|
assert engine._consecutive_failures == 2
|
|
|
|
def test_not_denied_permission_denied_increments(self):
|
|
"""denied=False with a permission_denied result increments — without
|
|
the denied flag, a permission_denied result is treated as a failure.
|
|
This confirms the denied flag is required to skip counting."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._consecutive_failures = 0
|
|
engine._track_tool_result_for_autonomy(
|
|
{"error_type": "permission_denied", "is_error": True}
|
|
)
|
|
assert engine._consecutive_failures == 1
|
|
|
|
def test_denied_does_not_reset_counter(self):
|
|
"""denied=True leaves the counter at whatever value it had — it
|
|
neither increments nor resets. This matters when denials interleave
|
|
with real failures."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=True,
|
|
)
|
|
# One real failure, then a denial.
|
|
engine._consecutive_failures = 0
|
|
engine._track_tool_result_for_autonomy({"error": "boom"})
|
|
assert engine._consecutive_failures == 1
|
|
engine._track_tool_result_for_autonomy({"error_type": "permission_denied"}, denied=True)
|
|
assert engine._consecutive_failures == 1 # Still 1, not reset to 0
|
|
|
|
def test_denied_no_effect_when_not_autonomy(self):
|
|
"""denied=True is a no-op when autonomy mode is off (same as the
|
|
base behavior — tracking is skipped entirely)."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(),
|
|
autonomy_mode=False,
|
|
)
|
|
engine._consecutive_failures = 0
|
|
engine._track_tool_result_for_autonomy({"error": "boom"}, denied=True)
|
|
assert engine._consecutive_failures == 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# U7: denial does not trigger autonomy_paused
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDenialDoesNotTriggerPause:
|
|
"""U7: a sequence of user denials does NOT trigger autonomy_paused
|
|
(reason=consecutive_failures), while a sequence of real failures does.
|
|
|
|
Uses _detect_autonomy_pause to check whether a pause would fire.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_three_denials_no_pause(self):
|
|
"""3 consecutive denials (denied=True) do NOT trigger
|
|
autonomy_paused — the counter never reaches the threshold."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(
|
|
autonomy_timeout_minutes=0, # Disabled
|
|
max_consecutive_failures=3,
|
|
),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._autonomy_started_at = 0 # Disable timeout check
|
|
denied_result = {
|
|
"error_type": "permission_denied",
|
|
"is_error": True,
|
|
"output": "",
|
|
}
|
|
# Simulate 3 denials — counter must stay at 0.
|
|
for _ in range(3):
|
|
engine._track_tool_result_for_autonomy(denied_result, denied=True)
|
|
assert engine._consecutive_failures == 0
|
|
# No pause should fire.
|
|
pause_info = engine._detect_autonomy_pause(step=1, progress={})
|
|
assert pause_info is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_three_failures_trigger_pause(self):
|
|
"""3 consecutive failures (denied=False) DO trigger autonomy_paused
|
|
— confirms the threshold still works for real failures."""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(
|
|
autonomy_timeout_minutes=0, # Disabled
|
|
max_consecutive_failures=3,
|
|
),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._autonomy_started_at = 0 # Disable timeout check
|
|
error_result = {"error": "tool crashed"}
|
|
# Simulate 3 real failures.
|
|
for _ in range(3):
|
|
engine._track_tool_result_for_autonomy(error_result)
|
|
assert engine._consecutive_failures == 3
|
|
# Pause should fire with reason=consecutive_failures.
|
|
pause_info = engine._detect_autonomy_pause(step=1, progress={})
|
|
assert pause_info is not None
|
|
reason, _token, _data = pause_info
|
|
assert reason == "consecutive_failures"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mixed_denials_and_failures_no_false_pause(self):
|
|
"""Interleaved denials and failures: denials don't increment, so
|
|
2 failures + 5 denials + 1 failure = 3 failures total → pause.
|
|
|
|
Confirms denials neither help nor hurt the failure count — only
|
|
real failures count toward the threshold.
|
|
"""
|
|
engine = ReActEngine(
|
|
llm_gateway=MagicMock(),
|
|
dangerous_tools_config=DangerousToolsConfig(
|
|
autonomy_timeout_minutes=0,
|
|
max_consecutive_failures=3,
|
|
),
|
|
autonomy_mode=True,
|
|
)
|
|
engine._autonomy_started_at = 0
|
|
error_result = {"error": "crash"}
|
|
denied_result = {"error_type": "permission_denied", "is_error": True}
|
|
|
|
# 2 real failures.
|
|
engine._track_tool_result_for_autonomy(error_result)
|
|
engine._track_tool_result_for_autonomy(error_result)
|
|
assert engine._consecutive_failures == 2
|
|
|
|
# 5 denials — counter must stay at 2.
|
|
for _ in range(5):
|
|
engine._track_tool_result_for_autonomy(denied_result, denied=True)
|
|
assert engine._consecutive_failures == 2
|
|
# No pause yet (below threshold).
|
|
assert engine._detect_autonomy_pause(step=1, progress={}) is None
|
|
|
|
# 1 more real failure → threshold reached.
|
|
engine._track_tool_result_for_autonomy(error_result)
|
|
assert engine._consecutive_failures == 3
|
|
pause_info = engine._detect_autonomy_pause(step=1, progress={})
|
|
assert pause_info is not None
|
|
reason, _token, _data = pause_info
|
|
assert reason == "consecutive_failures"
|