diff --git a/src/agentkit/core/react.py b/src/agentkit/core/react.py
index b3ef4eb..f7f14c9 100644
--- a/src/agentkit/core/react.py
+++ b/src/agentkit/core/react.py
@@ -147,7 +147,7 @@ class ReActResult:
 class ReActEvent:
     """ReAct 执行事件"""
 
-    event_type: str  # "thinking","token","tool_call","tool_result","confirmation_request","confirmation_result","phase_violation","step","final_answer","final_result","error"
+    event_type: str  # "thinking","token","tool_call","tool_result","confirmation_request","confirmation_result","phase_violation","step","final_answer","final_result","error","autonomy_paused","spec_review_request","spec_review_reply","expert_step","expert_result","team_synthesis","team_synthesis_chunk","phase_changed"
     step: int
     data: dict[str, object] = field(default_factory=dict)
     timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
@@ -317,6 +317,11 @@ class ReActEngine:
         # confirmation_request; non-dangerous tools execute without confirmation.
         self._dangerous_tools_config = dangerous_tools_config
         self._autonomy_mode: bool = autonomy_mode
+        # IQ-Boost/U3/R10: autonomy pause state. _autonomy_started_at is set in
+        # _execute_loop when autonomy mode engages; _consecutive_failures tracks
+        # tool errors. When either threshold is exceeded → autonomy_paused event.
+        self._autonomy_started_at: float = 0.0
+        self._consecutive_failures: int = 0
 
     def reset(self) -> None:
         """Reset internal state for reuse across conversations.
@@ -772,6 +777,10 @@ class ReActEngine:
         stream: bool = False,
         effective_timeout: float = 0.0,
         pitfall_warnings: "list[PitfallWarning] | None" = None,
+        # IQ-Boost/U3: autonomy pause resume handler. When autonomy_paused is
+        # triggered, the engine calls resume_handler(resume_token, reason) which
+        # blocks until the user sends ``resume`` (True) or cancels (False).
+        resume_handler: Callable[..., Awaitable[object]] | None = None,
     ) -> AsyncGenerator[ReActEvent, None]:
         """Unified ReAct loop — async generator yielding ReActEvent objects.
 
@@ -798,6 +807,10 @@ class ReActEngine:
         # below so the next execute() without a restore resets normally.
         if not self._state_restored:
             self.reset()
+        # IQ-Boost/U3: start autonomy timer when in autonomy mode.
+        if self._autonomy_mode:
+            self._autonomy_started_at = time.time()
+            self._consecutive_failures = 0
         tools = tools or []
         if tools:
             tools = self._maybe_add_tool_search(tools)
@@ -1258,6 +1271,23 @@ class ReActEngine:
                                 data={"tool_name": tc.name, "arguments": tc.arguments},
                             )
 
+                            # IQ-Boost/U3/R10: autonomy pause check (timeout/failures).
+                            # If paused and resume_handler returns False (cancel),
+                            # break out of the tool_calls loop.
+                            _progress = {
+                                "step": step,
+                                "tool_name": tc.name,
+                                "total_steps": len(trajectory),
+                            }
+                            should_continue, pause_events = await self._check_autonomy_pause(
+                                step, _progress, resume_handler
+                            )
+                            for _pev in pause_events:
+                                yield _pev
+                            if not should_continue:
+                                # User cancelled during pause — stop execution.
+                                break
+
                             # 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.
@@ -1381,6 +1411,9 @@ class ReActEngine:
 
                                 tool_duration_ms = int((time.monotonic() - tool_start) * 1000)
 
+                            # IQ-Boost/U3/R10: track tool failures for autonomy pause.
+                            self._track_tool_result_for_autonomy(tool_result)
+
                             react_step = ReActStep(
                                 step=step,
                                 action="tool_call",
@@ -1474,6 +1507,9 @@ class ReActEngine:
                                 )
                                 tool_duration_ms = int((time.monotonic() - tool_start) * 1000)
 
+                            # IQ-Boost/U3/R10: track tool failures for autonomy pause.
+                            self._track_tool_result_for_autonomy(tool_result)
+
                             react_step = ReActStep(
                                 step=step,
                                 action="tool_call",
@@ -1866,6 +1902,8 @@ class ReActEngine:
         timeout_seconds: float | None = None,
         confirmation_handler: Callable[..., Awaitable[object]] | None = None,
         pitfall_warnings: "list[PitfallWarning] | None" = None,
+        # IQ-Boost/U3: autonomy resume handler (see _execute_loop).
+        resume_handler: Callable[..., Awaitable[object]] | None = None,
     ) -> AsyncGenerator[ReActEvent, None]:
         """Execute ReAct loop, yielding ReActEvent objects.
 
@@ -1878,6 +1916,7 @@ class ReActEngine:
             compressor: 压缩策略，None 时使用实例默认压缩器
             timeout_seconds: 超时秒数，0 表示无超时，None 使用 default_timeout
             pitfall_warnings: U7/R12 — HIGH 级别避坑预警，注入 system prompt
+            resume_handler: U3/R10 — autonomy pause resume callback
         """
         effective_compressor = compressor if compressor is not None else self._compressor
         effective_timeout = (
@@ -1902,6 +1941,7 @@ class ReActEngine:
             stream=True,
             effective_timeout=effective_timeout,
             pitfall_warnings=pitfall_warnings,
+            resume_handler=resume_handler,
         ):
             yield event
 
@@ -2338,6 +2378,115 @@ class ReActEngine:
             },
         )
 
+    async def _check_autonomy_pause(
+        self,
+        step: int,
+        progress: dict[str, object],
+        resume_handler: Callable[..., Awaitable[object]] | None,
+    ) -> tuple[bool, list[ReActEvent]]:
+        """IQ-Boost/U3/R10: check autonomy pause conditions before tool execution.
+
+        Returns ``(should_continue, events)``:
+        - should_continue=True  → no pause needed; proceed with tool execution
+        - should_continue=False → autonomy_paused yielded; caller must break
+          the loop (resume_handler already returned False / cancelled) OR
+          retry (resume_handler returned True — counters reset, proceed).
+
+        When pause is triggered:
+        1. Yield ``autonomy_paused`` event with resume_token + reason + progress.
+        2. Call ``resume_handler(resume_token, reason)`` — blocks until user
+           sends ``resume`` (returns True) or cancels (returns False).
+        3. On resume: reset ``_autonomy_started_at`` + ``_consecutive_failures``,
+           return should_continue=True to let the caller retry the tool.
+        4. On cancel: return should_continue=False to break the loop.
+
+        If ``resume_handler`` is None (non-WS callers, tests), auto-resume
+        immediately (no blocking) — the pause event is still yielded for
+        observability, but the loop continues without waiting.
+        """
+        events: list[ReActEvent] = []
+        if not self._autonomy_mode or self._dangerous_tools_config is None:
+            return (True, events)
+
+        cfg = self._dangerous_tools_config
+        reason: str | None = None
+        # Check timeout (0 = disabled)
+        if (
+            cfg.autonomy_timeout_minutes > 0
+            and self._autonomy_started_at > 0
+            and time.time() - self._autonomy_started_at > cfg.autonomy_timeout_minutes * 60
+        ):
+            reason = "timeout"
+        # Check consecutive failures (0 = disabled)
+        elif (
+            cfg.max_consecutive_failures > 0
+            and self._consecutive_failures >= cfg.max_consecutive_failures
+        ):
+            reason = "consecutive_failures"
+
+        if reason is None:
+            return (True, events)
+
+        resume_token = f"autonomy_pause:{reason}:{step}"
+        events.append(
+            ReActEvent(
+                event_type="autonomy_paused",
+                step=step,
+                data={
+                    "reason": reason,
+                    "progress": progress,
+                    "resume_token": resume_token,
+                    "consecutive_failures": self._consecutive_failures,
+                    "elapsed_seconds": (
+                        time.time() - self._autonomy_started_at
+                        if self._autonomy_started_at > 0
+                        else 0
+                    ),
+                },
+            )
+        )
+
+        # Wait for user resume or cancel
+        if resume_handler is None:
+            # Non-WS caller (tests, REST) — auto-resume without blocking.
+            logger.warning("autonomy_paused (%s) with no resume_handler — auto-resuming", reason)
+            self._autonomy_started_at = time.time()
+            self._consecutive_failures = 0
+            return (True, events)
+
+        try:
+            should_resume = await resume_handler(resume_token, reason)
+        except asyncio.CancelledError:
+            raise
+        except Exception as e:
+            logger.warning(f"Autonomy resume handler error: {e}")
+            should_resume = False
+
+        if should_resume:
+            # Reset counters and continue
+            self._autonomy_started_at = time.time()
+            self._consecutive_failures = 0
+            return (True, events)
+        # Cancelled by user — break the loop
+        return (False, events)
+
+    def _track_tool_result_for_autonomy(self, tool_result: object) -> None:
+        """U3/R10: track tool failures for consecutive_failures threshold.
+
+        Called after each tool execution. If the result is an error dict,
+        increment the counter; otherwise reset to 0.
+        """
+        if not self._autonomy_mode:
+            return
+        if isinstance(tool_result, dict) and (
+            "error" in tool_result
+            or tool_result.get("is_error") is True
+            or tool_result.get("error_type") is not None
+        ):
+            self._consecutive_failures += 1
+        else:
+            self._consecutive_failures = 0
+
     async def _execute_tool_with_confirmation(
         self,
         tc: object,
diff --git a/src/agentkit/server/config.py b/src/agentkit/server/config.py
index ee34e06..d2d85be 100644
--- a/src/agentkit/server/config.py
+++ b/src/agentkit/server/config.py
@@ -65,6 +65,10 @@ class DangerousToolsConfig:
     confirmation request before execution. KTD2: conservative whitelist —
     ponytail ceiling = missed dangerous ops not in the list; upgrade path =
     LLM-assisted classification.
+
+    IQ-Boost/U3: also carries autonomy pause thresholds (timeout + consecutive
+    failures). Kept here to avoid a separate config section — all autonomy
+    behavior in one place.
     """
 
     enabled: bool = True
@@ -81,6 +85,9 @@ class DangerousToolsConfig:
             r"^payment_",
         ]
     )
+    # U3/R10: autonomy pause thresholds. 0 = disabled (no pause).
+    autonomy_timeout_minutes: int = 30
+    max_consecutive_failures: int = 3
 
     def is_dangerous(self, tool_name: str) -> bool:
         """Check if a tool name matches any dangerous pattern."""
@@ -96,6 +103,8 @@ class DangerousToolsConfig:
         return cls(
             enabled=data.get("enabled", True),
             tool_patterns=data.get("tool_patterns") or cls().tool_patterns,
+            autonomy_timeout_minutes=data.get("autonomy_timeout_minutes", 30),
+            max_consecutive_failures=data.get("max_consecutive_failures", 3),
         )
 
 
diff --git a/src/agentkit/server/routes/chat.py b/src/agentkit/server/routes/chat.py
index f07866a..089fce2 100644
--- a/src/agentkit/server/routes/chat.py
+++ b/src/agentkit/server/routes/chat.py
@@ -1019,6 +1019,9 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None:
     # U8/R8: pending spec-review futures keyed by spec_review_id. Resolved
     # by the spec_review_reply client message; cancelled on WS teardown.
     pending_spec_reviews: dict[str, asyncio.Future] = {}
+    # IQ-Boost/U3: pending autonomy-resume futures keyed by resume_token.
+    # Resolved by the ``resume`` client message; cancelled on WS teardown.
+    pending_autonomy_resumes: dict[str, asyncio.Future] = {}
     chat_manager.add(session_id, websocket, pending_replies)
 
     cancellation_token = CancellationToken()
@@ -1101,6 +1104,7 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None:
                         pending_replies,
                         pending_confirmations,
                         pending_spec_reviews,
+                        pending_autonomy_resumes,
                         model_override=model,
                     )
                 )
@@ -1156,6 +1160,22 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None:
                 cancellation_token.cancel()
                 await websocket.send_json({"type": "result", "data": {"status": "cancelled"}})
 
+            elif msg_type == "resume":
+                # IQ-Boost/U3/R10: Resume autonomy-paused execution. The client
+                # sends {resume_token}. An unknown token is logged + ignored.
+                resume_token = msg.get("resume_token")
+                logger.info(f"Received resume for token: {resume_token!r}")
+                if resume_token and resume_token in pending_autonomy_resumes:
+                    fut = pending_autonomy_resumes[resume_token]
+                    if not fut.done():
+                        fut.set_result(True)
+                    else:
+                        logger.warning(f"resume token {resume_token!r} already resolved")
+                else:
+                    logger.warning(
+                        f"resume token {resume_token!r} not found in pending_autonomy_resumes"
+                    )
+
             elif msg_type == "ping":
                 await websocket.send_json({"type": "pong"})
 
@@ -1180,6 +1200,10 @@ async def chat_websocket(websocket: WebSocket, session_id: str) -> None:
         for fut in pending_spec_reviews.values():
             if not fut.done():
                 fut.cancel()
+        # IQ-Boost/U3: cancel any pending autonomy-resume futures.
+        for fut in pending_autonomy_resumes.values():
+            if not fut.done():
+                fut.cancel()
         chat_manager.remove(session_id, websocket)
 
 
@@ -1192,6 +1216,8 @@ async def _handle_chat_message(
     pending_replies: dict[str, asyncio.Future],
     pending_confirmations: dict[str, asyncio.Future] | None = None,
     pending_spec_reviews: dict[str, asyncio.Future] | None = None,
+    # IQ-Boost/U3: pending autonomy-resume futures (keyed by resume_token).
+    pending_autonomy_resumes: dict[str, asyncio.Future] | None = None,
     model_override: str | None = None,
 ) -> None:
     """Handle a user message: append to session, execute Agent, stream events.
@@ -1583,6 +1609,35 @@ async def _handle_chat_message(
     if hasattr(react_engine, "_spec_review_handler"):
         react_engine._spec_review_handler = _spec_review_handler
 
+    # IQ-Boost/U3/R10: autonomy resume handler. When the engine triggers
+    # autonomy_paused (timeout/consecutive_failures), it calls this handler
+    # which blocks until the user sends ``resume`` (True) or the WS disconnects
+    # (Future cancelled → False). The engine already yielded the
+    # ``autonomy_paused`` event (forwarded to the frontend by the event loop
+    # below); this handler just provides the blocking mechanism.
+    _pending_autonomy_resumes = (
+        pending_autonomy_resumes if pending_autonomy_resumes is not None else {}
+    )
+
+    async def _resume_handler(resume_token: str, reason: str) -> bool:
+        """Block until the user sends ``resume`` for the given token."""
+        loop = asyncio.get_running_loop()
+        future: asyncio.Future[bool] = loop.create_future()
+        _pending_autonomy_resumes[resume_token] = future
+        logger.info(f"Autonomy paused ({reason}), waiting for resume: {resume_token}")
+        try:
+            # Wait up to 30 minutes for user resume (long task availability).
+            result = await asyncio.wait_for(future, timeout=1800.0)
+            return bool(result)
+        except asyncio.TimeoutError:
+            logger.warning(f"Autonomy resume {resume_token} timed out (30 min)")
+            return False
+        except asyncio.CancelledError:
+            logger.warning(f"Autonomy resume {resume_token} cancelled")
+            return False
+        finally:
+            _pending_autonomy_resumes.pop(resume_token, None)
+
     logger.info(
         f"Chat session {session_id}: executing with {len(routing.tools)} tools, model={routing.model}, skill={routing.skill_name}"
     )
@@ -1601,6 +1656,7 @@ async def _handle_chat_message(
             system_prompt=routing.system_prompt,
             cancellation_token=cancellation_token,
             confirmation_handler=_confirmation_handler,
+            resume_handler=_resume_handler,
         ):
             if event.event_type == "final_answer":
                 # Flush any buffered tokens as a single write
@@ -1658,6 +1714,17 @@ async def _handle_chat_message(
                         "data": event.data,
                     }
                 )
+            elif event.event_type == "autonomy_paused":
+                # IQ-Boost/U3/R10: forward autonomy pause to the frontend.
+                # The _resume_handler closure is already blocking the engine
+                # waiting for the user's ``resume`` message. The frontend
+                # shows a pause card with reason + progress + resume button.
+                await websocket.send_json(
+                    {
+                        "type": "autonomy_paused",
+                        "data": event.data,
+                    }
+                )
             elif event.event_type == "spec_review_request":
                 # U8/R8: the _spec_review_handler closure already sent this
                 # request directly to the frontend (it owns the spec_review_id
diff --git a/tests/unit/test_autonomy_paused.py b/tests/unit/test_autonomy_paused.py
new file mode 100644
index 0000000..c4b6ef5
--- /dev/null
+++ b/tests/unit/test_autonomy_paused.py
@@ -0,0 +1,366 @@
+"""Unit tests for autonomy_paused event + timeout/failure tracking (IQ-Boost U3, R10).
+
+Verifies:
+- Timeout triggers autonomy_paused with reason="timeout"
+- Consecutive failures (3x) trigger autonomy_paused with reason="consecutive_failures"
+- resume_handler returns True → counters reset, execution continues
+- resume_handler returns False → execution stops (cancel)
+- resume_handler=None → auto-resume (non-blocking, for tests/REST)
+- Non-autonomy mode → no pause (gate is a no-op)
+- Disabled thresholds (0) → no pause
+- _track_tool_result_for_autonomy increments on error, resets on success
+"""
+
+from __future__ import annotations
+
+import time
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from agentkit.core.react import ReActEngine
+from agentkit.server.config import DangerousToolsConfig
+
+
+# ---------------------------------------------------------------------------
+# _track_tool_result_for_autonomy
+# ---------------------------------------------------------------------------
+
+
+class TestTrackToolResult:
+    """Failure tracking increments on error, resets on success."""
+
+    def test_success_resets_counter(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(),
+            autonomy_mode=True,
+        )
+        engine._consecutive_failures = 2
+        engine._track_tool_result_for_autonomy({"output": "ok"})
+        assert engine._consecutive_failures == 0
+
+    def test_error_increments_counter(self):
+        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_is_error_flag_increments(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(),
+            autonomy_mode=True,
+        )
+        engine._track_tool_result_for_autonomy({"is_error": True, "output": ""})
+        assert engine._consecutive_failures == 1
+
+    def test_error_type_increments(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(),
+            autonomy_mode=True,
+        )
+        engine._track_tool_result_for_autonomy({"error_type": "permission_denied"})
+        assert engine._consecutive_failures == 1
+
+    def test_non_dict_result_resets(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(),
+            autonomy_mode=True,
+        )
+        engine._consecutive_failures = 2
+        engine._track_tool_result_for_autonomy("plain string result")
+        assert engine._consecutive_failures == 0
+
+    def test_no_tracking_when_not_autonomy(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(),
+            autonomy_mode=False,
+        )
+        engine._consecutive_failures = 0
+        engine._track_tool_result_for_autonomy({"error": "boom"})
+        assert engine._consecutive_failures == 0  # Unchanged
+
+
+# ---------------------------------------------------------------------------
+# _check_autonomy_pause — no pause conditions
+# ---------------------------------------------------------------------------
+
+
+class TestAutonomyPauseNoTrigger:
+    """When conditions are not met, _check_autonomy_pause returns should_continue=True."""
+
+    @pytest.mark.asyncio
+    async def test_no_autonomy_mode_no_pause(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(),
+            autonomy_mode=False,
+        )
+        should_continue, events = await engine._check_autonomy_pause(
+            step=1, progress={}, resume_handler=None
+        )
+        assert should_continue is True
+        assert events == []
+
+    @pytest.mark.asyncio
+    async def test_no_config_no_pause(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=None,
+            autonomy_mode=True,
+        )
+        should_continue, events = await engine._check_autonomy_pause(
+            step=1, progress={}, resume_handler=None
+        )
+        assert should_continue is True
+        assert events == []
+
+    @pytest.mark.asyncio
+    async def test_below_thresholds_no_pause(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=30,
+                max_consecutive_failures=3,
+            ),
+            autonomy_mode=True,
+        )
+        engine._autonomy_started_at = time.time()  # Just started
+        engine._consecutive_failures = 1  # Below threshold
+        should_continue, events = await engine._check_autonomy_pause(
+            step=1, progress={}, resume_handler=None
+        )
+        assert should_continue is True
+        assert events == []
+
+    @pytest.mark.asyncio
+    async def test_disabled_thresholds_no_pause(self):
+        """0 = disabled for both thresholds."""
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=0,
+                max_consecutive_failures=0,
+            ),
+            autonomy_mode=True,
+        )
+        # Even with expired time + many failures, disabled = no pause.
+        engine._autonomy_started_at = time.time() - 999999
+        engine._consecutive_failures = 99
+        should_continue, events = await engine._check_autonomy_pause(
+            step=1, progress={}, resume_handler=None
+        )
+        assert should_continue is True
+        assert events == []
+
+
+# ---------------------------------------------------------------------------
+# _check_autonomy_pause — timeout trigger
+# ---------------------------------------------------------------------------
+
+
+class TestAutonomyPauseTimeout:
+    """Timeout triggers autonomy_paused with reason='timeout'."""
+
+    @pytest.mark.asyncio
+    async def test_timeout_triggers_pause_auto_resume(self):
+        """resume_handler=None → auto-resume (non-blocking)."""
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=1,  # 1 minute
+                max_consecutive_failures=99,  # Disabled for this test
+            ),
+            autonomy_mode=True,
+        )
+        # Simulate started 2 minutes ago (exceeds 1-min timeout).
+        engine._autonomy_started_at = time.time() - 120
+
+        should_continue, events = await engine._check_autonomy_pause(
+            step=5, progress={"step": 5}, resume_handler=None
+        )
+
+        # Auto-resume → should_continue=True
+        assert should_continue is True
+        assert len(events) == 1
+        assert events[0].event_type == "autonomy_paused"
+        assert events[0].data["reason"] == "timeout"
+        assert "resume_token" in events[0].data
+        assert events[0].data["elapsed_seconds"] >= 120
+        # Counters reset after auto-resume
+        assert engine._consecutive_failures == 0
+        assert engine._autonomy_started_at > time.time() - 1  # Fresh timer
+
+    @pytest.mark.asyncio
+    async def test_timeout_triggers_pause_user_resumes(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=1,
+                max_consecutive_failures=99,
+            ),
+            autonomy_mode=True,
+        )
+        engine._autonomy_started_at = time.time() - 120
+
+        resume_handler = AsyncMock(return_value=True)
+        should_continue, events = await engine._check_autonomy_pause(
+            step=3, progress={}, resume_handler=resume_handler
+        )
+
+        assert should_continue is True
+        assert events[0].data["reason"] == "timeout"
+        resume_handler.assert_awaited_once()
+        # Counters reset
+        assert engine._consecutive_failures == 0
+
+    @pytest.mark.asyncio
+    async def test_timeout_triggers_pause_user_cancels(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=1,
+                max_consecutive_failures=99,
+            ),
+            autonomy_mode=True,
+        )
+        engine._autonomy_started_at = time.time() - 120
+
+        resume_handler = AsyncMock(return_value=False)
+        should_continue, events = await engine._check_autonomy_pause(
+            step=3, progress={}, resume_handler=resume_handler
+        )
+
+        assert should_continue is False  # Cancelled
+        assert events[0].data["reason"] == "timeout"
+
+
+# ---------------------------------------------------------------------------
+# _check_autonomy_pause — consecutive failures trigger
+# ---------------------------------------------------------------------------
+
+
+class TestAutonomyPauseConsecutiveFailures:
+    """3 consecutive failures trigger autonomy_paused."""
+
+    @pytest.mark.asyncio
+    async def test_consecutive_failures_triggers_pause(self):
+        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 = time.time()
+        engine._consecutive_failures = 3  # At threshold
+
+        resume_handler = AsyncMock(return_value=True)
+        should_continue, events = await engine._check_autonomy_pause(
+            step=5, progress={"step": 5}, resume_handler=resume_handler
+        )
+
+        assert should_continue is True  # Resumed
+        assert len(events) == 1
+        assert events[0].event_type == "autonomy_paused"
+        assert events[0].data["reason"] == "consecutive_failures"
+        assert events[0].data["consecutive_failures"] == 3
+        # Reset after resume
+        assert engine._consecutive_failures == 0
+
+    @pytest.mark.asyncio
+    async def test_below_failure_threshold_no_pause(self):
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=0,
+                max_consecutive_failures=3,
+            ),
+            autonomy_mode=True,
+        )
+        engine._autonomy_started_at = time.time()
+        engine._consecutive_failures = 2  # Below threshold
+
+        should_continue, events = await engine._check_autonomy_pause(
+            step=1, progress={}, resume_handler=None
+        )
+
+        assert should_continue is True
+        assert events == []
+
+    @pytest.mark.asyncio
+    async def test_resume_handler_exception_cancels(self):
+        """If resume_handler raises, treat as cancel (should_continue=False)."""
+        engine = ReActEngine(
+            llm_gateway=MagicMock(),
+            dangerous_tools_config=DangerousToolsConfig(
+                autonomy_timeout_minutes=0,
+                max_consecutive_failures=3,
+            ),
+            autonomy_mode=True,
+        )
+        engine._consecutive_failures = 3
+
+        async def failing_handler(token, reason):
+            raise RuntimeError("handler crashed")
+
+        should_continue, events = await engine._check_autonomy_pause(
+            step=1, progress={}, resume_handler=failing_handler
+        )
+
+        assert should_continue is False  # Cancelled due to exception
+        assert events[0].data["reason"] == "consecutive_failures"
+
+
+# ---------------------------------------------------------------------------
+# Config parsing
+# ---------------------------------------------------------------------------
+
+
+class TestAutonomyConfigParsing:
+    """DangerousToolsConfig parses autonomy thresholds from dict."""
+
+    def test_defaults(self):
+        cfg = DangerousToolsConfig()
+        assert cfg.autonomy_timeout_minutes == 30
+        assert cfg.max_consecutive_failures == 3
+
+    def test_custom_values_from_dict(self):
+        cfg = DangerousToolsConfig.from_dict(
+            {
+                "enabled": True,
+                "autonomy_timeout_minutes": 60,
+                "max_consecutive_failures": 5,
+            }
+        )
+        assert cfg.autonomy_timeout_minutes == 60
+        assert cfg.max_consecutive_failures == 5
+
+    def test_disabled_values_from_dict(self):
+        cfg = DangerousToolsConfig.from_dict(
+            {
+                "autonomy_timeout_minutes": 0,
+                "max_consecutive_failures": 0,
+            }
+        )
+        assert cfg.autonomy_timeout_minutes == 0
+        assert cfg.max_consecutive_failures == 0
+
+    def test_partial_config_uses_defaults(self):
+        cfg = DangerousToolsConfig.from_dict({"enabled": True})
+        assert cfg.autonomy_timeout_minutes == 30
+        assert cfg.max_consecutive_failures == 3
+
+    def test_empty_dict_uses_defaults(self):
+        cfg = DangerousToolsConfig.from_dict({})
+        assert cfg.autonomy_timeout_minutes == 30
+        assert cfg.max_consecutive_failures == 3