diff --git a/src/agentkit/core/react.py b/src/agentkit/core/react.py index f7f14c9..e2dc0c6 100644 --- a/src/agentkit/core/react.py +++ b/src/agentkit/core/react.py @@ -1206,8 +1206,12 @@ class ReActEngine: tc.id, tool_result, compressor, tc.name ) conversation.append(tool_msg) - elif self._should_execute_parallel(response.tool_calls): + elif self._should_execute_parallel(response.tool_calls) and not ( + self._autonomy_mode and self._dangerous_tools_config is not None + ): # 并行执行多个工具调用 (parallel_tools=True) + # IQ-Boost/U2: autonomy mode forces serial execution so + # dangerous tools are always gated by _check_autonomy_gate. tool_results = await asyncio.gather( *[ self._execute_tool(tc.name, tc.arguments, tools) @@ -1220,6 +1224,10 @@ class ReActEngine: if isinstance(tool_result, Exception): tool_result = {"error": str(tool_result)} + # IQ-Boost/U3/R10: track failures for autonomy pause + # (no-op when autonomy mode is off). + self._track_tool_result_for_autonomy(tool_result) + yield ReActEvent( event_type="tool_call", step=step, @@ -1274,19 +1282,27 @@ class ReActEngine: # IQ-Boost/U3/R10: autonomy pause check (timeout/failures). # If paused and resume_handler returns False (cancel), # break out of the tool_calls loop. + # Split into detect→yield→await to avoid deadlock: the + # pause event must be yielded BEFORE awaiting resume. _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 + _pause_info = self._detect_autonomy_pause(step, _progress) + if _pause_info is not None: + _reason, _token, _edata = _pause_info + yield ReActEvent( + event_type="autonomy_paused", + step=step, + data=_edata, + ) + _should_continue = await self._await_autonomy_resume( + _token, _reason, resume_handler + ) + 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 @@ -2378,35 +2394,29 @@ class ReActEngine: }, ) - async def _check_autonomy_pause( + def _detect_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. + ) -> tuple[str, str, dict[str, object]] | None: + """IQ-Boost/U3/R10: detect autonomy pause conditions (non-blocking). - 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). + Returns ``(reason, resume_token, event_data)`` when a pause is + triggered, or ``None`` when no pause is needed. - 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. + The caller is responsible for: + 1. Yielding the ``autonomy_paused`` event (built from event_data) + **before** awaiting ``_await_autonomy_resume`` — otherwise the + frontend never receives the pause event and the resume handler + deadlocks. + 2. Awaiting ``_await_autonomy_resume(resume_token, reason, handler)`` + to block until the user resumes or cancels. - 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. + This split fixes the original deadlock where the event was buffered + inside this method and only yielded after the handler returned. """ - events: list[ReActEvent] = [] if not self._autonomy_mode or self._dangerous_tools_config is None: - return (True, events) + return None cfg = self._dangerous_tools_config reason: str | None = None @@ -2425,34 +2435,42 @@ class ReActEngine: reason = "consecutive_failures" if reason is None: - return (True, events) + return None 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 - ), - }, - ) - ) + event_data: dict[str, object] = { + "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 + ), + } + return (reason, resume_token, event_data) - # Wait for user resume or cancel + async def _await_autonomy_resume( + self, + resume_token: str, + reason: str, + resume_handler: Callable[..., Awaitable[object]] | None, + ) -> bool: + """Block until user resumes or cancels autonomy pause (U3/R10). + + Returns True on resume (counters reset, caller retries the tool), + False on cancel (caller breaks the loop). + + If ``resume_handler`` is None (non-WS callers, tests), auto-resume + immediately without blocking. + """ 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) + return True try: should_resume = await resume_handler(resume_token, reason) @@ -2466,9 +2484,9 @@ class ReActEngine: # Reset counters and continue self._autonomy_started_at = time.time() self._consecutive_failures = 0 - return (True, events) + return True # Cancelled by user — break the loop - return (False, events) + return False def _track_tool_result_for_autonomy(self, tool_result: object) -> None: """U3/R10: track tool failures for consecutive_failures threshold. diff --git a/src/agentkit/core/reflexion.py b/src/agentkit/core/reflexion.py index 5facdc3..10a1fb7 100644 --- a/src/agentkit/core/reflexion.py +++ b/src/agentkit/core/reflexion.py @@ -750,13 +750,20 @@ class ReflexionEngine: """检索历史 prompt 反思，返回最佳版本 (U6/R12, R13). Searches EpisodicMemory for similar task_input reflections with - score > min_score. Returns the highest-scored reflection as: + stored quality_score > min_score. Returns the highest-quality + reflection as: {improved_prompt, score, reflection, version, task_hash} or None if no episodic_memory / no results / all below threshold. KTD5: callers should only invoke this when a trigger condition is met (verify failure / schema failure / loop detection) to avoid pointless retrieval on every task. + + Note: ``item.score`` from ``EpisodicMemory.search`` is the hybrid + relevance score (cosine + time_decay), NOT the stored quality_score. + We read ``quality_score`` from ``item.value`` for filtering/ranking + so that a high-quality reflection (score=1.0) is preferred over a + low-quality one (score=0.0) regardless of textual similarity. """ if self._episodic_memory is None: return None @@ -772,24 +779,32 @@ class ReflexionEngine: if not results: return None - # Filter by min_score, pick the highest-scored + # Filter by stored quality_score (from value dict), pick the highest. + # Fallback to item.score (relevance) when quality_score is absent. best: MemoryItem | None = None + best_quality: float = 0.0 for item in results: - score = item.score or 0.0 - if score > min_score and (best is None or score > (best.score or 0.0)): + value = item.value if isinstance(item.value, dict) else {} + quality = float(value.get("quality_score", 0.0) or 0.0) + if quality <= 0.0: + # Fallback to relevance score if quality_score missing + quality = float(item.score or 0.0) + if quality > min_score and (best is None or quality > best_quality): best = item + best_quality = quality if best is None: return None - # Extract improved_prompt from metadata (output_summary field) + # Read fields from value dict (matching EpisodicMemory.search shape) + value = best.value if isinstance(best.value, dict) else {} + improved_prompt = value.get("output_summary", "") or "" + reflection_text = value.get("reflection", "") or "" metadata = best.metadata or {} - improved_prompt = metadata.get("output_summary", "") or metadata.get("improved_prompt", "") - reflection_text = metadata.get("reflection", "") or best.value or "" return { "improved_prompt": improved_prompt, - "score": best.score or 0.0, + "score": best_quality, "reflection": reflection_text, "version": metadata.get("version", 1), "task_hash": metadata.get("task_hash", ""), diff --git a/src/agentkit/memory/episodic.py b/src/agentkit/memory/episodic.py index 0a24559..3fc5a2b 100644 --- a/src/agentkit/memory/episodic.py +++ b/src/agentkit/memory/episodic.py @@ -569,9 +569,16 @@ class EpisodicMemory(Memory): ) return items - async def cleanup_expired(self, max_age_days: int = 30) -> int: + async def cleanup_expired( + self, max_age_days: int = 30, task_type: str | None = None + ) -> int: """删除超过 max_age_days 天的记录 (U5/R15 TTL). + Args: + max_age_days: Records older than this many days are deleted. + task_type: Optional filter — only delete records with this task_type + (e.g. ``"prompt_reflection"``). None = delete all task types. + Returns the number of deleted rows. 0 on failure (non-raising). """ from datetime import timedelta @@ -583,6 +590,8 @@ class EpisodicMemory(Memory): try: Model = self._episodic_model stmt = sql_delete(Model).where(Model.created_at < cutoff) + if task_type is not None: + stmt = stmt.where(Model.task_type == task_type) result = await db.execute(stmt) await db.commit() return result.rowcount or 0 diff --git a/tests/unit/test_autonomy_paused.py b/tests/unit/test_autonomy_paused.py index c4b6ef5..417befb 100644 --- a/tests/unit/test_autonomy_paused.py +++ b/tests/unit/test_autonomy_paused.py @@ -18,10 +18,37 @@ from unittest.mock import AsyncMock, MagicMock import pytest -from agentkit.core.react import ReActEngine +from agentkit.core.react import ReActEngine, ReActEvent from agentkit.server.config import DangerousToolsConfig +# --------------------------------------------------------------------------- +# Helper: simulates the caller pattern (detect → yield event → await resume) +# --------------------------------------------------------------------------- + + +async def _detect_and_await_pause( + engine: ReActEngine, + step: int, + progress: dict, + resume_handler=None, +) -> tuple[bool, list[ReActEvent]]: + """Simulate the react.py caller pattern for autonomy pause. + + Mirrors the production code: detect (non-blocking) → build event → + yield event → await resume (blocking). Returns (should_continue, events). + """ + pause_info = engine._detect_autonomy_pause(step, progress) + if pause_info is None: + return True, [] + reason, token, event_data = pause_info + event = ReActEvent(event_type="autonomy_paused", step=step, data=event_data) + should_continue = await engine._await_autonomy_resume( + token, reason, resume_handler + ) + return should_continue, [event] + + # --------------------------------------------------------------------------- # _track_tool_result_for_autonomy # --------------------------------------------------------------------------- @@ -90,12 +117,12 @@ class TestTrackToolResult: # --------------------------------------------------------------------------- -# _check_autonomy_pause — no pause conditions +# _detect_autonomy_pause + _await_autonomy_resume — no pause conditions # --------------------------------------------------------------------------- class TestAutonomyPauseNoTrigger: - """When conditions are not met, _check_autonomy_pause returns should_continue=True.""" + """When conditions are not met, no pause is triggered (should_continue=True).""" @pytest.mark.asyncio async def test_no_autonomy_mode_no_pause(self): @@ -104,8 +131,8 @@ class TestAutonomyPauseNoTrigger: dangerous_tools_config=DangerousToolsConfig(), autonomy_mode=False, ) - should_continue, events = await engine._check_autonomy_pause( - step=1, progress={}, resume_handler=None + should_continue, events = await _detect_and_await_pause( + engine, step=1, progress={}, resume_handler=None ) assert should_continue is True assert events == [] @@ -117,8 +144,8 @@ class TestAutonomyPauseNoTrigger: dangerous_tools_config=None, autonomy_mode=True, ) - should_continue, events = await engine._check_autonomy_pause( - step=1, progress={}, resume_handler=None + should_continue, events = await _detect_and_await_pause( + engine, step=1, progress={}, resume_handler=None ) assert should_continue is True assert events == [] @@ -135,8 +162,8 @@ class TestAutonomyPauseNoTrigger: ) 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 + should_continue, events = await _detect_and_await_pause( + engine, step=1, progress={}, resume_handler=None ) assert should_continue is True assert events == [] @@ -155,15 +182,15 @@ class TestAutonomyPauseNoTrigger: # 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 + should_continue, events = await _detect_and_await_pause( + engine, step=1, progress={}, resume_handler=None ) assert should_continue is True assert events == [] # --------------------------------------------------------------------------- -# _check_autonomy_pause — timeout trigger +# _detect_autonomy_pause + _await_autonomy_resume — timeout trigger # --------------------------------------------------------------------------- @@ -184,8 +211,8 @@ class TestAutonomyPauseTimeout: # 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 + should_continue, events = await _detect_and_await_pause( + engine, step=5, progress={"step": 5}, resume_handler=None ) # Auto-resume → should_continue=True @@ -212,8 +239,8 @@ class TestAutonomyPauseTimeout: 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 + should_continue, events = await _detect_and_await_pause( + engine, step=3, progress={}, resume_handler=resume_handler ) assert should_continue is True @@ -235,8 +262,8 @@ class TestAutonomyPauseTimeout: 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 + should_continue, events = await _detect_and_await_pause( + engine, step=3, progress={}, resume_handler=resume_handler ) assert should_continue is False # Cancelled @@ -244,7 +271,7 @@ class TestAutonomyPauseTimeout: # --------------------------------------------------------------------------- -# _check_autonomy_pause — consecutive failures trigger +# _detect_autonomy_pause + _await_autonomy_resume — consecutive failures trigger # --------------------------------------------------------------------------- @@ -265,8 +292,8 @@ class TestAutonomyPauseConsecutiveFailures: 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 + should_continue, events = await _detect_and_await_pause( + engine, step=5, progress={"step": 5}, resume_handler=resume_handler ) assert should_continue is True # Resumed @@ -290,8 +317,8 @@ class TestAutonomyPauseConsecutiveFailures: 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 + should_continue, events = await _detect_and_await_pause( + engine, step=1, progress={}, resume_handler=None ) assert should_continue is True @@ -313,8 +340,8 @@ class TestAutonomyPauseConsecutiveFailures: 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 + should_continue, events = await _detect_and_await_pause( + engine, step=1, progress={}, resume_handler=failing_handler ) assert should_continue is False # Cancelled due to exception diff --git a/tests/unit/test_lead_reflection_retrieval.py b/tests/unit/test_lead_reflection_retrieval.py index 263094f..1554358 100644 --- a/tests/unit/test_lead_reflection_retrieval.py +++ b/tests/unit/test_lead_reflection_retrieval.py @@ -35,16 +35,22 @@ def _make_memory_item( ) -> MemoryItem: from datetime import datetime, timezone + # Shape matches EpisodicMemory.list_prompt_reflections_by_hash return: + # output_summary/reflection/quality_score live in value dict, + # version/task_hash live in metadata. return MemoryItem( key="prompt_reflection:abc:1", - value={"task_input": "test", "reflection": reflection}, - metadata={ - "task_type": "prompt_reflection", + value={ + "input_summary": "test", "output_summary": output_summary, "reflection": reflection, + "quality_score": score, + }, + metadata={ + "task_type": "prompt_reflection", "version": version, "task_hash": "abc", - "quality_score": score, + "created_at": datetime.now(timezone.utc).isoformat(), }, score=score, created_at=datetime.now(timezone.utc),