fix(iq): apply code review fixes — P0 deadlock + P1 gate/tracking/filter

P0 #1: Fix autonomy_paused event deadlock — split _check_autonomy_pause
into _detect_autonomy_pause (non-blocking) + _await_autonomy_resume
(blocking). Caller now yields the pause event BEFORE awaiting resume,
so the frontend receives it and the resume handler doesn't deadlock.

P1 #2+#6: Fix retrieve_prompt_reflection field mapping — read
output_summary/reflection/quality_score from MemoryItem.value dict
(matching EpisodicMemory.search shape) instead of metadata. Score
filtering uses stored quality_score, not search relevance score.

P1 #3: Add optional task_type filter to cleanup_expired so
prompt_reflection TTL cleanup doesn't delete all episodic records.

P1 #4: Disable parallel tool execution in autonomy mode — dangerous
tools must go through _check_autonomy_gate, which only runs in the
serial path.

P1 #5: Add _track_tool_result_for_autonomy to parallel result loop
so tool failures are counted toward the consecutive_failures threshold.

Tests: adapt test_autonomy_paused.py to new detect/await interface;
fix test_lead_reflection_retrieval.py mock shape (fields in value dict).
137 IQ-boost tests pass, ruff clean.

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.

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", ""),

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

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

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),