refactor(iq): ce-simplify-code fixes — P1 bug + P2 quality

P1 (bug fix):
- store_prompt_reflection: bypass store() and write ORM entry directly
  so task_hash/version/score land in metadata_ JSONB column. Previously
  store() dropped these keys, breaking list_prompt_reflections_by_hash
  filtering (U7 would always return empty).

P2 (quality):
- DangerousToolsConfig: pre-compile regex patterns in __post_init__
  (is_dangerous runs on per-tool hot path in autonomy mode).
- ABTester.compare_prompt_versions: extract _no_data_result helper,
  remove redundant isinstance check after value type guard.
- orchestrator: fix misleading "transient TeamPlan" comment.

Tests updated to match new store_prompt_reflection implementation
(direct ORM write instead of store() delegation). 137 tests pass.

src/agentkit/evolution/ab_tester.py

@@ -226,6 +226,17 @@ class ABTester:
 
     # ── IQ-Boost/U7: Prompt-version offline comparison (R14) ────────────
 
+    @staticmethod
+    def _no_data_result(task_hash: str) -> dict[str, object]:
+        """Build the empty-result payload for compare_prompt_versions."""
+        return {
+            "task_hash": task_hash,
+            "versions": [],
+            "best_version": None,
+            "recommendation": "no_data",
+            "total_versions": 0,
+        }
+
     async def compare_prompt_versions(self, task_hash: str) -> dict[str, object]:
         """离线对比同一 task_hash 的多个 prompt 版本效果 (U7/R14).
 
@@ -246,34 +257,16 @@ class ABTester:
             无 episodic_memory 或无记录时返回 "no_data" recommendation。
         """
         if self._episodic_memory is None:
-            return {
-                "task_hash": task_hash,
-                "versions": [],
-                "best_version": None,
-                "recommendation": "no_data",
-                "total_versions": 0,
-            }
+            return self._no_data_result(task_hash)
 
         try:
             items = await self._episodic_memory.list_prompt_reflections_by_hash(task_hash)
         except (DBAPIError, RuntimeError, ValueError, KeyError, OSError) as e:
             logger.warning(f"U7: compare_prompt_versions retrieval failed: {e}")
-            return {
-                "task_hash": task_hash,
-                "versions": [],
-                "best_version": None,
-                "recommendation": "no_data",
-                "total_versions": 0,
-            }
+            return self._no_data_result(task_hash)
 
         if not items:
-            return {
-                "task_hash": task_hash,
-                "versions": [],
-                "best_version": None,
-                "recommendation": "no_data",
-                "total_versions": 0,
-            }
+            return self._no_data_result(task_hash)
 
         # Sort by score descending (best first)
         sorted_items = sorted(items, key=lambda it: it.score, reverse=True)
@@ -282,8 +275,8 @@ class ABTester:
         for item in sorted_items:
             meta = item.metadata or {}
             value = item.value if isinstance(item.value, dict) else {}
-            reflection_text = value.get("reflection", "") if isinstance(value, dict) else ""
-            improved_prompt = value.get("output_summary", "") if isinstance(value, dict) else ""
+            reflection_text = value.get("reflection", "")
+            improved_prompt = value.get("output_summary", "")
             versions.append(
                 {
                     "version": meta.get("version", 1),

src/agentkit/experts/orchestrator.py

@@ -559,8 +559,7 @@ class TeamOrchestrator(
 
         Returns the original phases if count is within limit or retry fails.
         """
-        # Count independent subtasks via a transient TeamPlan (avoids mutating
-        # the real plan before execute() finalizes phase list).
+        # Count phases with no depends_on (independent subtasks).
         independent_count = sum(1 for ph in phases if not ph.depends_on)
         if independent_count <= self.MAX_INDEPENDENT_SUBTASKS:
             return phases

src/agentkit/memory/episodic.py

@@ -425,14 +425,21 @@ class EpisodicMemory(Memory):
     ) -> str | None:
         """持久化 prompt 反思到 EpisodicMemory，支持跨任务检索 (U5/R11).
 
-        Reuses the existing ``store()`` path with ``task_type="prompt_reflection"``
-        as the discriminator. The ORM row's fields map:
+        Writes directly to the ORM row (bypassing ``store()``) so that
+        ``task_hash``/``version``/``score`` land in the ``metadata_`` JSONB
+        column — ``store()`` only maps a fixed set of metadata keys to ORM
+        columns and drops the rest, which would break
+        ``list_prompt_reflections_by_hash`` filtering.
+
+        The ORM row's fields map:
             input_summary  ← task_input (truncated)
             output_summary ← improved_prompt (truncated)
             reflection     ← reflection text
             quality_score  ← score (0.0=failed, 1.0=verified)
             outcome        ← "reflection"
             agent_name     ← agent_name
+            task_type      ← "prompt_reflection"
+            metadata_      ← {task_hash, version, score, timestamp}
 
         Returns the storage key ``"prompt_reflection:{task_hash}:{version}"``
         on success, or None on failure (non-raising — callers continue without).
@@ -443,29 +450,42 @@ class EpisodicMemory(Memory):
             task_hash = hashlib.sha256(task_input.encode("utf-8")).hexdigest()[:16]
         key = f"prompt_reflection:{task_hash}:{version}"
 
-        value = {
-            "task_input": task_input[:500],
-            "reflection": reflection,
-            "improved_prompt": improved_prompt,
-            "score": score,
-            "version": version,
+        # Generate embedding from task_input for semantic search compatibility.
+        embedding = None
+        if self._embedder:
+            try:
+                embedding = await self._embedder.embed(task_input)
+            except (RuntimeError, ValueError, OSError) as e:
+                logger.warning(f"U5: embedder failed for prompt reflection: {e}")
+
+        extra_meta = {
             "task_hash": task_hash,
+            "version": version,
+            "score": score,
             "timestamp": datetime.now(timezone.utc).isoformat(),
         }
-        metadata = {
-            "agent_name": agent_name,
-            "task_type": "prompt_reflection",
-            "output_summary": improved_prompt[:500],
-            "outcome": "reflection",
-            "quality_score": score,
-            "reflection": reflection,
-        }
-        try:
-            await self.store(key=key, value=value, metadata=metadata)
-            return key
-        except (DBAPIError, ValueError, KeyError, RuntimeError, OSError) as e:
-            logger.warning(f"U5: failed to persist prompt reflection: {e}")
-            return None
+
+        async with self._session_factory() as db:
+            try:
+                Model = self._episodic_model
+                entry = Model(
+                    agent_name=agent_name,
+                    task_type="prompt_reflection",
+                    input_summary=task_input[:500],
+                    output_summary=improved_prompt[:500],
+                    outcome="reflection",
+                    quality_score=score,
+                    reflection=reflection,
+                    embedding=embedding,
+                    metadata_=extra_meta,
+                )
+                db.add(entry)
+                await db.commit()
+                return key
+            except (DBAPIError, ValueError, KeyError, RuntimeError, OSError) as e:
+                await db.rollback()
+                logger.warning(f"U5: failed to persist prompt reflection: {e}")
+                return None
 
     async def search_prompt_reflections(
         self,

src/agentkit/server/config.py

@@ -88,12 +88,20 @@ class DangerousToolsConfig:
     # U3/R10: autonomy pause thresholds. 0 = disabled (no pause).
     autonomy_timeout_minutes: int = 30
     max_consecutive_failures: int = 3
+    # Pre-compiled patterns — populated by __post_init__, reused by is_dangerous
+    # on the per-tool hot path. ponytail: ceiling = rebuild when tool_patterns
+    # mutated after construction (rare; config-time only). Upgrade path =
+    # cached_property with invalidation on tool_patterns setter.
+    _compiled_patterns: list[re.Pattern] = field(default_factory=list, repr=False)
+
+    def __post_init__(self) -> None:
+        self._compiled_patterns = [re.compile(p) for p in self.tool_patterns]
 
     def is_dangerous(self, tool_name: str) -> bool:
         """Check if a tool name matches any dangerous pattern."""
         if not self.enabled:
             return False
-        return any(re.match(p, tool_name) for p in self.tool_patterns)
+        return any(p.match(tool_name) for p in self._compiled_patterns)
 
     @classmethod
     def from_dict(cls, data: dict | None) -> "DangerousToolsConfig":

tests/unit/test_reflexion_persist.py

@@ -58,15 +58,29 @@ def _make_llm_gateway_mock(reflection_text: str = "reflection text") -> MagicMoc
 
 
 class TestStorePromptReflection:
-    """U5/R11: store_prompt_reflection persists via store() with
-    task_type='prompt_reflection' discriminator."""
+    """U5/R11: store_prompt_reflection persists via direct ORM write with
+    task_type='prompt_reflection' discriminator and metadata_ JSONB."""
+
+    def _make_persistable_mock(self) -> MagicMock:
+        """Build an EpisodicMemory mock with _embedder + _session_factory
+        configured so store_prompt_reflection can run its real code path."""
+        mem = MagicMock(spec=EpisodicMemory)
+        mem._embedder = None  # skip embedding generation in test
+        mock_db = AsyncMock()
+        mock_db.add = MagicMock()
+        mock_db.commit = AsyncMock()
+        mock_db.rollback = AsyncMock()
+        mem._session_factory = MagicMock()
+        mem._session_factory.return_value.__aenter__ = AsyncMock(return_value=mock_db)
+        mem._session_factory.return_value.__aexit__ = AsyncMock(return_value=None)
+        mem._episodic_model = MagicMock()
+        # Track the ORM entry passed to db.add for assertions
+        mem._mock_db = mock_db
+        return mem
 
     @pytest.mark.asyncio
-    async def test_store_calls_underlying_store_with_correct_metadata(self):
-        mem = MagicMock(spec=EpisodicMemory)
-        mem.store = AsyncMock(return_value=None)
-        # We need to call the real method, not a mock — patch store only
-        # Use the unbound method pattern
+    async def test_store_writes_orm_entry_with_correct_fields(self):
+        mem = self._make_persistable_mock()
         await EpisodicMemory.store_prompt_reflection(
             mem,
             task_input="test task",
@@ -77,30 +91,26 @@ class TestStorePromptReflection:
             agent_name="test_agent",
         )
 
-        mem.store.assert_awaited_once()
-        call_kwargs = mem.store.await_args.kwargs
-        assert "key" in call_kwargs
-        assert call_kwargs["key"].startswith("prompt_reflection:")
-        assert ":1" in call_kwargs["key"]
-
-        value = call_kwargs["value"]
-        assert value["task_input"] == "test task"
-        assert value["reflection"] == "reflection text"
-        assert value["improved_prompt"] == "improved prompt"
-        assert value["score"] == 0.5
-        assert value["version"] == 1
-        assert "timestamp" in value
-
-        metadata = call_kwargs["metadata"]
-        assert metadata["task_type"] == "prompt_reflection"
-        assert metadata["agent_name"] == "test_agent"
-        assert metadata["quality_score"] == 0.5
-        assert metadata["reflection"] == "reflection text"
+        # Verify ORM entry created via model constructor + db.add + commit
+        mem._episodic_model.assert_called_once()
+        call_kwargs = mem._episodic_model.call_args.kwargs
+        assert call_kwargs["task_type"] == "prompt_reflection"
+        assert call_kwargs["agent_name"] == "test_agent"
+        assert call_kwargs["input_summary"] == "test task"
+        assert call_kwargs["output_summary"] == "improved prompt"
+        assert call_kwargs["reflection"] == "reflection text"
+        assert call_kwargs["quality_score"] == 0.5
+        assert call_kwargs["outcome"] == "reflection"
+        # metadata_ JSONB carries task_hash + version + score
+        assert "task_hash" in call_kwargs["metadata_"]
+        assert call_kwargs["metadata_"]["version"] == 1
+        assert call_kwargs["metadata_"]["score"] == 0.5
+        mem._mock_db.add.assert_called_once()
+        mem._mock_db.commit.assert_awaited_once()
 
     @pytest.mark.asyncio
     async def test_store_returns_key_on_success(self):
-        mem = MagicMock(spec=EpisodicMemory)
-        mem.store = AsyncMock(return_value=None)
+        mem = self._make_persistable_mock()
         key = await EpisodicMemory.store_prompt_reflection(
             mem,
             task_input="test",
@@ -116,8 +126,10 @@ class TestStorePromptReflection:
     async def test_store_returns_none_on_failure(self):
         from sqlalchemy.exc import DBAPIError
 
-        mem = MagicMock(spec=EpisodicMemory)
-        mem.store = AsyncMock(side_effect=DBAPIError("stmt", params={}, orig=Exception("db down")))
+        mem = self._make_persistable_mock()
+        mem._mock_db.commit = AsyncMock(
+            side_effect=DBAPIError("stmt", params={}, orig=Exception("db down"))
+        )
         key = await EpisodicMemory.store_prompt_reflection(
             mem,
             task_input="test",
@@ -125,11 +137,11 @@ class TestStorePromptReflection:
             improved_prompt="p",
         )
         assert key is None
+        mem._mock_db.rollback.assert_awaited_once()
 
     @pytest.mark.asyncio
     async def test_store_uses_provided_task_hash(self):
-        mem = MagicMock(spec=EpisodicMemory)
-        mem.store = AsyncMock(return_value=None)
+        mem = self._make_persistable_mock()
         key = await EpisodicMemory.store_prompt_reflection(
             mem,
             task_input="test",
@@ -141,8 +153,7 @@ class TestStorePromptReflection:
 
     @pytest.mark.asyncio
     async def test_store_generates_task_hash_from_input(self):
-        mem = MagicMock(spec=EpisodicMemory)
-        mem.store = AsyncMock(return_value=None)
+        mem = self._make_persistable_mock()
         key1 = await EpisodicMemory.store_prompt_reflection(
             mem, task_input="same task", reflection="r", improved_prompt="p"
         )
@@ -401,8 +412,9 @@ class TestMultiVersionCoexistence:
 
     @pytest.mark.asyncio
     async def test_multiple_versions_stored_with_incrementing_version(self):
-        mem = MagicMock(spec=EpisodicMemory)
-        mem.store = AsyncMock(return_value=None)
+        # Reuse the persistable mock helper from TestStorePromptReflection.
+        helper = TestStorePromptReflection()
+        mem = helper._make_persistable_mock()
 
         # Store v1, v2, v3 for same task
         key1 = await EpisodicMemory.store_prompt_reflection(
@@ -419,5 +431,6 @@ class TestMultiVersionCoexistence:
         assert key1 != key2 != key3
         assert ":1" in key1 and ":2" in key2 and ":3" in key3
 
-        # All three store() calls made
-        assert mem.store.await_count == 3
+        # All three ORM entries created via db.add
+        assert mem._mock_db.add.call_count == 3
+        assert mem._mock_db.commit.await_count == 3