feat(react): U2 PLAN_EXEC autonomy mode + dangerous-tools gate (R6-R9)

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

@@ -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

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]

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"

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