From 4729e20bbf2a5729479e75c65288d23743e3b61b Mon Sep 17 00:00:00 2001 From: Chiguyong Date: Mon, 6 Jul 2026 13:23:06 +0800 Subject: [PATCH] feat(react): U2 PLAN_EXEC autonomy mode + dangerous-tools gate (R6-R9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ReActEngine now accepts dangerous_tools_config + autonomy_mode. In PLAN_EXEC autonomy mode, dangerous tools (per DangerousToolsConfig whitelist) trigger confirmation_request BEFORE first execution; non-dangerous tools execute directly without confirmation. _build_phase_engine injects the config and sets autonomy_mode=True when PLAN_EXEC is engaged. - _check_autonomy_gate: pre-execution gate (serial/parallel/text-parsed paths) - Approved dangerous tool re-executes with _skip_dangerous_check=True - Rejected/handler-failure → permission_denied result (exit_code=126) - Default autonomy_mode=False → transparent no-op (backward compat) - 19 new tests (14 autonomy gate + 5 PLAN_EXEC wiring) --- src/agentkit/core/react.py | 198 ++++++++++++++++- src/agentkit/server/routes/chat.py | 6 + tests/unit/test_plan_exec_autonomy.py | 123 +++++++++++ tests/unit/test_react_autonomy.py | 302 ++++++++++++++++++++++++++ 4 files changed, 619 insertions(+), 10 deletions(-) create mode 100644 tests/unit/test_plan_exec_autonomy.py create mode 100644 tests/unit/test_react_autonomy.py diff --git a/src/agentkit/core/react.py b/src/agentkit/core/react.py index f39991b..b3ef4eb 100644 --- a/src/agentkit/core/react.py +++ b/src/agentkit/core/react.py @@ -44,6 +44,11 @@ if TYPE_CHECKING: from agentkit.evolution.pitfall_detector import PitfallWarning from agentkit.memory.retriever import MemoryRetriever + # U2/R8: duck-typed at runtime to avoid core→server layering violation. + # Any object with ``is_dangerous(tool_name: str) -> bool`` + ``enabled: bool`` + # is accepted; server/config.DangerousToolsConfig is the canonical impl. + from agentkit.server.config import DangerousToolsConfig + logger = logging.getLogger(__name__) @@ -204,6 +209,15 @@ class ReActEngine: # max_reinjections) # Loop detector threshold raised from 2 to 3 (R10/RV22). phase_budgets: dict[str, int] | None = None, + # IQ-Boost/U2/R8: dangerous-tools whitelist for PLAN_EXEC autonomy. + # Duck-typed (see TYPE_CHECKING import). None = no autonomy gate + # (backward compat for DIRECT_CHAT/REACT and existing tests). + dangerous_tools_config: "DangerousToolsConfig | None" = None, + # IQ-Boost/U2/R6: autonomy mode flag. True = PLAN_EXEC plan confirmed, + # only dangerous tools (per dangerous_tools_config) trigger + # confirmation_request; non-dangerous tools execute directly. + # False = existing behavior (tool-side _is_dangerous drives confirmation). + autonomy_mode: bool = False, ): if max_steps < 1: raise ValueError(f"max_steps must be >= 1, got {max_steps}") @@ -298,6 +312,11 @@ class ReActEngine: # _execute_loop's finally block so subsequent execute() calls without a # restore still reset properly. self._state_restored: bool = False + # IQ-Boost/U2: autonomy mode state (R6/R7). When True, the engine gates + # dangerous tools (per dangerous_tools_config) with a pre-execution + # confirmation_request; non-dangerous tools execute without confirmation. + self._dangerous_tools_config = dangerous_tools_config + self._autonomy_mode: bool = autonomy_mode def reset(self) -> None: """Reset internal state for reuse across conversations. @@ -1239,13 +1258,38 @@ class ReActEngine: data={"tool_name": tc.name, "arguments": tc.arguments}, ) - tool_start = time.monotonic() - tool_result = await self._execute_tool(tc.name, tc.arguments, tools) - tool_duration_ms = int((time.monotonic() - tool_start) * 1000) + # IQ-Boost/U2/R7: autonomy mode pre-execution gate. + # Dangerous tools (per config) get a confirmation_request + # BEFORE first execution; non-dangerous tools fall through. + ( + should_exec, + autonomy_events, + autonomy_result, + ) = await self._check_autonomy_gate( + tc.name, + tc.arguments, + tools, + step, + confirmation_handler, + ) + for _aev in autonomy_events: + yield _aev - # Handle confirmation flow - if isinstance(tool_result, dict) and tool_result.get( - "needs_confirmation" + if not should_exec: + # Autonomy gate handled execution (approved+skip + # flag, or rejected). Use its result directly. + tool_result = autonomy_result + tool_duration_ms = 0 + else: + tool_start = time.monotonic() + tool_result = await self._execute_tool(tc.name, tc.arguments, tools) + tool_duration_ms = int((time.monotonic() - tool_start) * 1000) + + # Handle confirmation flow (tool-side _is_dangerous) + if ( + should_exec + and isinstance(tool_result, dict) + and tool_result.get("needs_confirmation") ): confirmation_id = tool_result["confirmation_id"] command = tool_result.get("command", "") @@ -1407,11 +1451,28 @@ class ReActEngine: step=step, data={"tool_name": pc["name"], "arguments": pc["arguments"]}, ) - tool_start = time.monotonic() - tool_result = await self._execute_tool( - pc["name"], pc["arguments"], tools + # IQ-Boost/U2: autonomy gate for text-parsed calls too + # (consistency — dangerous tools must be gated regardless + # of whether the LLM returned structured or text calls). + pc_args = pc["arguments"] if isinstance(pc["arguments"], dict) else {} + ( + should_exec, + autonomy_events, + autonomy_result, + ) = await self._check_autonomy_gate( + pc["name"], pc_args, tools, step, confirmation_handler ) - tool_duration_ms = int((time.monotonic() - tool_start) * 1000) + for _aev in autonomy_events: + yield _aev + if not should_exec: + tool_result = autonomy_result + tool_duration_ms = 0 + else: + tool_start = time.monotonic() + tool_result = await self._execute_tool( + pc["name"], pc["arguments"], tools + ) + tool_duration_ms = int((time.monotonic() - tool_start) * 1000) react_step = ReActStep( step=step, @@ -2170,6 +2231,113 @@ class ReActEngine: logger.warning(error_msg) return {"error": error_msg, "error_code": "tool_execution_failed"} + async def _check_autonomy_gate( + self, + tool_name: str, + tool_arguments: dict, + tools: list[Tool], + step: int, + confirmation_handler: Callable[..., Awaitable[object]] | None, + ) -> tuple[bool, list[ReActEvent], object]: + """IQ-Boost/U2/R7: pre-execution gate for PLAN_EXEC autonomy mode. + + When autonomy_mode is on and the tool matches the dangerous-tools + whitelist, emit a ``confirmation_request`` BEFORE the tool runs and + wait on ``confirmation_handler``. Approved → execute with + ``_skip_dangerous_check=True`` (bypass the tool's own _is_dangerous + re-confirmation). Rejected → return a permission_denied result. + + Non-dangerous tools and non-autonomy mode fall through (return + ``should_execute=True``) so the caller proceeds with the existing + ``_execute_tool`` → ``needs_confirmation`` flow unchanged. + + Returns ``(should_execute, events, pre_result)``: + - should_execute=True → caller runs _execute_tool; pre_result=None + - should_execute=False → caller uses pre_result as tool_result + """ + events: list[ReActEvent] = [] + if ( + not self._autonomy_mode + or self._dangerous_tools_config is None + or not self._dangerous_tools_config.is_dangerous(tool_name) + ): + return (True, events, None) + + # Dangerous tool in autonomy mode — gate before first execution. + confirmation_id = f"autonomy:{tool_name}:{step}" + command = f"{tool_name} {tool_arguments}" + reason = "危险操作（匹配 autonomy 白名单）需确认后执行" + events.append( + ReActEvent( + event_type="confirmation_request", + step=step, + data={ + "confirmation_id": confirmation_id, + "tool_name": tool_name, + "command": command, + "reason": reason, + }, + ) + ) + + approved = False + if confirmation_handler is not None: + try: + approved = await confirmation_handler(confirmation_id, command, reason) + except asyncio.CancelledError: + raise + except Exception as e: + logger.warning(f"Autonomy confirmation handler error: {e}") + + if approved: + tool = self._find_tool(tool_name, tools) + if tool is None: + events.append( + ReActEvent( + event_type="confirmation_result", + step=step, + data={"confirmation_id": confirmation_id, "approved": True}, + ) + ) + return (False, events, {"error": f"Tool '{tool_name}' not found"}) + clean_args = {k: v for k, v in tool_arguments.items() if not k.startswith("_")} + clean_args["_skip_dangerous_check"] = True + try: + result = await tool.safe_execute(**clean_args) + except (ToolValidationError, ValueError, TypeError, RuntimeError) as e: + result = { + "error": f"Tool '{tool_name}' execution failed: {e}", + "error_code": "tool_execution_failed", + } + events.append( + ReActEvent( + event_type="confirmation_result", + step=step, + data={"confirmation_id": confirmation_id, "approved": True}, + ) + ) + return (False, events, result) + + # Rejected by user (or handler timed out / returned False) + events.append( + ReActEvent( + event_type="confirmation_result", + step=step, + data={"confirmation_id": confirmation_id, "approved": False}, + ) + ) + return ( + False, + events, + { + "output": "", + "exit_code": 126, + "is_error": True, + "error_type": "permission_denied", + "message": f"用户拒绝执行危险操作: {tool_name}", + }, + ) + async def _execute_tool_with_confirmation( self, tc: object, @@ -2186,6 +2354,16 @@ class ReActEngine: Tuple of (tool_result, list of ReActEvents to yield) """ events: list[ReActEvent] = [] + # IQ-Boost/U2: autonomy gate runs first (pre-execution). If it handles + # the tool (dangerous in autonomy mode), return immediately with its + # result + events; otherwise fall through to the normal flow. + should_exec, gate_events, gate_result = await self._check_autonomy_gate( + tc.name, tc.arguments, tools, step, confirmation_handler + ) + if not should_exec: + return (gate_result, gate_events) + events.extend(gate_events) + tool_result = await self._execute_tool(tc.name, tc.arguments, tools) # Check if tool returned a confirmation request diff --git a/src/agentkit/server/routes/chat.py b/src/agentkit/server/routes/chat.py index 3df74d0..f07866a 100644 --- a/src/agentkit/server/routes/chat.py +++ b/src/agentkit/server/routes/chat.py @@ -695,9 +695,15 @@ def _build_phase_engine( ) return (None, None, f"phase policy error: {str(e)[:200]}") + # IQ-Boost/U2/R6-R8: PLAN_EXEC engages autonomy mode — plan confirmation + # is handled by the existing spec_review gate; after confirmation, only + # dangerous tools (per dangerous_tools_config) trigger confirmation_request. + dangerous_tools_cfg = getattr(server_config, "dangerous_tools", None) engine = ReActEngine( llm_gateway=llm_gateway, phase_policy=phase_policy, + dangerous_tools_config=dangerous_tools_cfg, + autonomy_mode=True, ) advance_phase_tool = AdvancePhaseTool(engine=engine) tools_with_advance_phase = list(base_tools) + [advance_phase_tool] diff --git a/tests/unit/test_plan_exec_autonomy.py b/tests/unit/test_plan_exec_autonomy.py new file mode 100644 index 0000000..86d0986 --- /dev/null +++ b/tests/unit/test_plan_exec_autonomy.py @@ -0,0 +1,123 @@ +"""Integration tests for PLAN_EXEC autonomy mode wiring (IQ-Boost U2). + +Verifies that ``_build_phase_engine`` in chat.py constructs a PLAN_EXEC +engine with ``autonomy_mode=True`` and ``dangerous_tools_config`` injected +from ``ServerConfig``. This is the integration seam between U1 (config) +and U2 (engine gate). +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +from agentkit.chat.skill_routing import ExecutionMode +from agentkit.core.react import ReActEngine +from agentkit.server.config import DangerousToolsConfig, ServerConfig + + +def _make_server_config(dangerous_tools: DangerousToolsConfig | None = None) -> ServerConfig: + """Build a minimal ServerConfig with the dangerous_tools section.""" + cfg = ServerConfig.from_dict({}) + if dangerous_tools is not None: + cfg.dangerous_tools = dangerous_tools + return cfg + + +class TestBuildPhaseEngineAutonomyWiring: + """_build_phase_engine injects autonomy config into the PLAN_EXEC engine.""" + + def test_plan_exec_engine_gets_autonomy_mode(self): + """When PLAN_EXEC is engaged, the engine has autonomy_mode=True.""" + # Import here to avoid module-level FastAPI app construction. + from agentkit.server.routes.chat import _build_phase_engine + + server_cfg = _make_server_config(DangerousToolsConfig()) + engine, tools, err = _build_phase_engine( + server_config=server_cfg, + llm_gateway=MagicMock(), + execution_mode=ExecutionMode.PLAN_EXEC, + base_tools=[], + session_id="test-session", + ) + assert err is None + assert engine is not None + assert isinstance(engine, ReActEngine) + assert engine._autonomy_mode is True + assert engine._dangerous_tools_config is not None + assert engine._dangerous_tools_config.is_dangerous("rm") is True + + def test_plan_exec_engine_gets_custom_dangerous_config(self): + """Custom dangerous_tools config (disabled) propagates to the engine.""" + from agentkit.server.routes.chat import _build_phase_engine + + server_cfg = _make_server_config(DangerousToolsConfig(enabled=False)) + engine, _, err = _build_phase_engine( + server_config=server_cfg, + llm_gateway=MagicMock(), + execution_mode=ExecutionMode.PLAN_EXEC, + base_tools=[], + session_id="test", + ) + assert err is None + assert engine is not None + assert engine._dangerous_tools_config.enabled is False + # Disabled whitelist → nothing is dangerous + assert engine._dangerous_tools_config.is_dangerous("rm") is False + + def test_non_plan_exec_mode_returns_none_engine(self): + """Non-PLAN_EXEC mode → (None, None, None) — no autonomy engine built.""" + from agentkit.server.routes.chat import _build_phase_engine + + server_cfg = _make_server_config(DangerousToolsConfig()) + engine, tools, err = _build_phase_engine( + server_config=server_cfg, + llm_gateway=MagicMock(), + execution_mode=ExecutionMode.REACT, + base_tools=[], + session_id="test", + ) + assert engine is None + assert tools is None + assert err is None + + def test_plan_exec_engine_gets_default_dangerous_config_when_server_config_missing( + self, + ): + """When server_config is None or lacks dangerous_tools, the engine + still gets autonomy_mode=True (config defaults to None — gate is a + no-op, but autonomy flag is set for future wiring).""" + from agentkit.server.routes.chat import _build_phase_engine + + # server_config=None simulates tests/misconfigured deployments. + engine, _, err = _build_phase_engine( + server_config=None, + llm_gateway=MagicMock(), + execution_mode=ExecutionMode.PLAN_EXEC, + base_tools=[], + session_id="test", + ) + assert err is None + assert engine is not None + assert engine._autonomy_mode is True + # dangerous_tools_config is None when server_config is None — gate + # becomes a transparent no-op (backward compat for tests without config). + assert engine._dangerous_tools_config is None + + def test_plan_exec_engine_includes_advance_phase_tool(self): + """PLAN_EXEC engine's tool list includes AdvancePhaseTool.""" + from agentkit.server.routes.chat import _build_phase_engine + + server_cfg = _make_server_config(DangerousToolsConfig()) + engine, tools, err = _build_phase_engine( + server_config=server_cfg, + llm_gateway=MagicMock(), + execution_mode=ExecutionMode.PLAN_EXEC, + base_tools=[], + session_id="test", + ) + assert err is None + assert engine is not None + assert tools is not None + # AdvancePhaseTool appended to the base tool list. + assert len(tools) == 1 + assert type(tools[0]).__name__ == "AdvancePhaseTool" diff --git a/tests/unit/test_react_autonomy.py b/tests/unit/test_react_autonomy.py new file mode 100644 index 0000000..bb56f56 --- /dev/null +++ b/tests/unit/test_react_autonomy.py @@ -0,0 +1,302 @@ +"""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