feat(react): U3 PLAN_EXEC autonomy pause + autonomy_paused event (R10)
Add autonomy pause mechanism for PLAN_EXEC mode: when execution exceeds timeout (default 30min) or hits consecutive failures (default 3), engine emits `autonomy_paused` event and awaits user resume via resume_handler. - DangerousToolsConfig: add autonomy_timeout_minutes + max_consecutive_failures - ReActEngine: add _check_autonomy_pause() + _track_tool_result_for_autonomy() Called in serial + text-parsed tool paths (between tool_call event and autonomy gate). resume_handler=None → auto-resume (REST/testing). - chat.py: WebSocket `resume` message handler + _resume_handler closure (Future-based, 30min hard timeout). pending_autonomy_resumes cleanup in finally block. autonomy_paused event forwarded to frontend. Tests: 21 new tests (track_result + no_trigger + timeout + failures + config parsing + handler exception). All pass; U1/U2 no regression.
This commit is contained in:
Chiguyong
parent
4729e20bbf
commit
7627403a8a
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
Loading…
Reference in New Issue