P3 Final: Batch hydration, mypy fixes, hydration regression

openspace/skill_engine/analysis_store.py

@@ -22,7 +22,7 @@
 from datetime import datetime
 from functools import wraps
 from pathlib import Path
-from typing import Any, Dict, Generator, List, Optional
+from typing import Any, Callable, Dict, Generator, List, Optional
 
 from openspace.utils.logging import Logger
 
@@ -43,12 +43,12 @@ def _db_retry(
     max_retries: int = 5,
     initial_delay: float = 0.1,
     backoff: float = 2.0,
-):
+) -> Callable:
     """Retry on transient SQLite errors with exponential backoff."""
 
-    def decorator(func):
+    def decorator(func: Callable) -> Callable:
         @wraps(func)
-        def wrapper(*args, **kwargs):
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
             delay = initial_delay
             for attempt in range(max_retries):
                 try:
@@ -78,7 +78,7 @@ class AnalysisStore:
     Usage (standalone)::
 
         store = AnalysisStore(db_path=Path("analyses.db"))
-        store.record_execution_analysis(analysis)
+        store.record_execution_analysis_sync(analysis)
         recent = store.load_analyses(skill_id="some_skill")
         store.close()
 
@@ -314,9 +314,70 @@ def hydrate_recent_analyses(self, record: SkillRecord) -> SkillRecord:
             last_updated=record.last_updated,
             recent_analyses=recent_analyses,  # This is what we're hydrating
             tool_dependencies=record.tool_dependencies,
+            critical_tools=record.critical_tools,  # Fix: preserve critical_tools
             tags=record.tags,
         )
 
+    def batch_load_recent_analyses(
+        self, skill_ids: List[str], limit: int = 5
+    ) -> Dict[str, List[ExecutionAnalysis]]:
+        """Batch-load recent analyses for multiple skills.
+
+        Uses window functions to get top-N per skill in 2 queries total
+        (analyses + judgments), avoiding O(N*M) fan-out.
+        """
+        if not skill_ids:
+            return {}
+        placeholders = ",".join("?" * len(skill_ids))
+        with self._reader() as conn:
+            # 1 query: ranked analyses per skill (DISTINCT via window over unique ea.id per skill)
+            rows = conn.execute(
+                f"SELECT sub.* FROM ("
+                f"  SELECT ea.*, sj.skill_id AS source_skill_id, "
+                f"  ROW_NUMBER() OVER ("
+                f"    PARTITION BY sj.skill_id ORDER BY ea.timestamp DESC"
+                f"  ) AS rn "
+                f"  FROM execution_analyses ea "
+                f"  JOIN skill_judgments sj ON ea.id = sj.analysis_id "
+                f"  WHERE sj.skill_id IN ({placeholders})"
+                f") sub WHERE sub.rn <= ?",
+                [*skill_ids, limit],
+            ).fetchall()
+
+            # Collect analysis IDs and group by skill
+            filtered: Dict[str, list] = {}
+            analysis_ids: set = set()
+            for row in rows:
+                sid = row["source_skill_id"]
+                filtered.setdefault(sid, []).append(row)
+                analysis_ids.add(row["id"])
+
+            if not analysis_ids:
+                return {}
+
+            # 1 query: all judgments for matched analyses
+            aid_placeholders = ",".join("?" * len(analysis_ids))
+            judgment_rows = conn.execute(
+                f"SELECT analysis_id, skill_id, skill_applied, note "
+                f"FROM skill_judgments WHERE analysis_id IN ({aid_placeholders})",
+                list(analysis_ids),
+            ).fetchall()
+
+            judgments_by_analysis: Dict[str, list] = {}
+            for jr in judgment_rows:
+                judgments_by_analysis.setdefault(jr["analysis_id"], []).append(jr)
+
+            # Deserialize with preloaded judgments
+            result: Dict[str, List[ExecutionAnalysis]] = {}
+            for sid, ea_rows in filtered.items():
+                result[sid] = [
+                    self._to_analysis(
+                        conn, row, judgments_by_analysis.get(row["id"], [])
+                    )
+                    for row in ea_rows
+                ]
+            return result
+
     # ── Bulk Operations ────────────────────────────────────────────────
 
     @_db_retry()
@@ -495,6 +556,7 @@ def insert_analysis(self, a: ExecutionAnalysis) -> int:
             ),
         )
         analysis_id = cur.lastrowid
+        assert analysis_id is not None, "INSERT must set lastrowid"
 
         for j in a.skill_judgments:
             self._conn.execute(
@@ -505,14 +567,21 @@ def insert_analysis(self, a: ExecutionAnalysis) -> int:
         return analysis_id
 
     @staticmethod
-    def _to_analysis(conn: sqlite3.Connection, row: sqlite3.Row) -> ExecutionAnalysis:
+    def _to_analysis(
+        conn: sqlite3.Connection,
+        row: sqlite3.Row,
+        preloaded_judgments: Optional[List[sqlite3.Row]] = None,
+    ) -> ExecutionAnalysis:
         """Deserialize an execution_analyses row + judgments → ExecutionAnalysis."""
         analysis_id = row["id"]
 
-        judgment_rows = conn.execute(
-            "SELECT skill_id, skill_applied, note FROM skill_judgments WHERE analysis_id=?",
-            (analysis_id,),
-        ).fetchall()
+        if preloaded_judgments is not None:
+            judgment_rows = preloaded_judgments
+        else:
+            judgment_rows = conn.execute(
+                "SELECT skill_id, skill_applied, note FROM skill_judgments WHERE analysis_id=?",
+                (analysis_id,),
+            ).fetchall()
 
         suggestions: list[EvolutionSuggestion] = []
         raw_suggestions = row["evolution_suggestions"]

openspace/skill_engine/lineage_tracker.py

@@ -116,10 +116,7 @@ def __init__(
             if db_path is None:
                 raise ValueError("Either db_path or conn must be provided")
             self._db_path = Path(db_path)
-            self._conn = sqlite3.connect(str(db_path), timeout=30)
-            self._conn.row_factory = sqlite3.Row
-            self._conn.execute("PRAGMA journal_mode=WAL")
-            self._conn.execute("PRAGMA busy_timeout=30000")
+            self._conn = self._make_connection(read_only=False)
             # Standalone mode: ensure schema exists
             mm = MigrationManager(conn=self._conn, lock=self._mu)
             mm.initialize_schema()
@@ -480,6 +477,7 @@ def get_ancestors(
             List of ancestor :class:`SkillRecord` objects, sorted by
             generation (nearest first).
         """
+        max_depth = min(max_depth, 50)  # Safety clamp to prevent DoS
         with self._reader() as conn:
             visited: set[str] = {skill_id}  # Fix 1: Seed with starting skill_id to prevent cycles
             ancestors: List[SkillRecord] = []
@@ -551,6 +549,7 @@ def get_lineage_tree(
         Returns:
             Nested dict representing the lineage tree.
         """
+        max_depth = min(max_depth, 50)  # Safety clamp to prevent DoS
         with self._reader() as conn:
             return self._subtree(conn, skill_id, max_depth, set())
 

openspace/skill_engine/migration_manager.py

@@ -171,7 +171,7 @@ class MigrationManager:
 
         # SkillStore creates us internally:
         # store._migrations = MigrationManager(conn=store._conn, lock=store._mu)
-        # SkillStore calls manager.initialize_schema() during __init__
+        # SkillStore calls manager.ensure_current_schema() during __init__
     """
 
     def __init__(

openspace/skill_engine/store.py

@@ -29,6 +29,7 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import json
 import sqlite3
 import threading
@@ -681,14 +682,15 @@ def get_versions(self, name: str) -> List[SkillRecord]:
 
         Delegates to :class:`LineageTracker` (Epic 3.3).
         """
-        # Fix 5: Hydrate recent_analyses after delegation
+        # Fix: Fully hydrate records after delegation (tags, tool_deps, critical_tools, recent_analyses)
         records = self._lineage.get_evolution_chain(name)
-        return [self._hydrate_recent_analyses(record) for record in records]
+        return self._hydrate_records(records)
 
     @_db_retry()
     def load_by_category(self, category: SkillCategory, *, active_only: bool = True) -> List[SkillRecord]:
         """Delegate to SkillRepository for category queries."""
-        return self._repo.load_by_category(category, active_only=active_only)
+        records = self._repo.load_by_category(category, active_only=active_only)
+        return self._hydrate_records(records)
 
     @_db_retry()
     def load_analyses(
@@ -826,9 +828,9 @@ def get_ancestry(self, skill_id: str, max_depth: int = 10) -> List[SkillRecord]:
 
         Delegates to :class:`LineageTracker` (Epic 3.3).
         """
-        # Fix 5: Hydrate recent_analyses after delegation
+        # Fix: Fully hydrate records after delegation (tags, tool_deps, critical_tools, recent_analyses)
         records = self._lineage.get_ancestors(skill_id, max_depth=max_depth)
-        return [self._hydrate_recent_analyses(record) for record in records]
+        return self._hydrate_records(records)
 
     @_db_retry()
     def get_lineage_tree(self, skill_id: str, max_depth: int = 5) -> Dict[str, Any]:
@@ -992,6 +994,82 @@ def _hydrate_recent_analyses(self, record: SkillRecord) -> SkillRecord:
         """Hydrate recent_analyses for a SkillRecord from AnalysisStore delegation."""
         return self._analyses.hydrate_recent_analyses(record)
 
+    def _hydrate_record(self, record: SkillRecord) -> SkillRecord:
+        """Fully hydrate a SkillRecord with tags, tool_deps, critical_tools, and recent_analyses.
+
+        Used by facade methods that delegate to modules returning partially hydrated records.
+        """
+        with self._reader() as conn:
+            # Hydrate tool dependencies and critical tools
+            dep_rows = conn.execute(
+                "SELECT tool_key, critical FROM skill_tool_deps WHERE skill_id=?",
+                (record.skill_id,),
+            ).fetchall()
+
+            tool_dependencies = [r["tool_key"] for r in dep_rows]
+            critical_tools = [r["tool_key"] for r in dep_rows if r["critical"]]
+
+        # Hydrate tags using TagSearch
+        tags = self._tag_search.get_tags(record.skill_id)
+
+        # Hydrate recent_analyses using AnalysisStore
+        hydrated_record = self._analyses.hydrate_recent_analyses(record)
+
+        # Return a new record with all fields hydrated
+        return dataclasses.replace(
+            hydrated_record,
+            tags=tags,
+            tool_dependencies=tool_dependencies,
+            critical_tools=critical_tools,
+        )
+
+    def _hydrate_records(self, records: List[SkillRecord]) -> List[SkillRecord]:
+        """Batch hydrate multiple SkillRecords — O(1) queries instead of O(N).
+
+        Batches tag, tool_dep, and analysis queries across all records.
+        """
+        if not records:
+            return records
+
+        skill_ids = [r.skill_id for r in records]
+        placeholders = ",".join("?" * len(skill_ids))
+
+        with self._reader() as conn:
+            # Batch 1: tool_deps — 1 query
+            dep_rows = conn.execute(
+                f"SELECT skill_id, tool_key, critical FROM skill_tool_deps "
+                f"WHERE skill_id IN ({placeholders})",
+                skill_ids,
+            ).fetchall()
+
+        deps_by_skill: Dict[str, List[str]] = {}
+        critical_by_skill: Dict[str, List[str]] = {}
+        for row in dep_rows:
+            sid = row["skill_id"]
+            deps_by_skill.setdefault(sid, []).append(row["tool_key"])
+            if row["critical"]:
+                critical_by_skill.setdefault(sid, []).append(row["tool_key"])
+
+        # Batch 2: tags — 1 query (via TagSearch)
+        tags_by_skill = self._tag_search.get_tags_batch(skill_ids)
+
+        # Batch 3: analyses — 2 queries (via AnalysisStore)
+        analyses_by_skill = self._analyses.batch_load_recent_analyses(
+            skill_ids, SkillRecord.MAX_RECENT
+        )
+
+        # Assemble
+        return [
+            dataclasses.replace(
+                record,
+                tags=tags_by_skill.get(record.skill_id, []),
+                tool_dependencies=deps_by_skill.get(record.skill_id, []),
+                critical_tools=critical_by_skill.get(record.skill_id, []),
+                recent_analyses=analyses_by_skill.get(record.skill_id, []),
+            )
+            for record in records
+        ]
+
     # Deserialization
     def _to_record(self, conn: sqlite3.Connection, row: sqlite3.Row) -> SkillRecord:
         """Deserialize a skill_records row + related rows → SkillRecord."""
@@ -1101,6 +1179,10 @@ def get_tags(self, skill_id: str) -> List[str]:
         """Get all tags for a skill (facade to TagSearch)."""
         return self._tag_search.get_tags(skill_id)
 
+    def get_tags_batch(self, skill_ids: List[str]) -> Dict[str, List[str]]:
+        """Batch-get tags for multiple skills (facade to TagSearch)."""
+        return self._tag_search.get_tags_batch(skill_ids)
+
     def get_all_tags(self) -> List[Dict[str, Any]]:
         """Get all tags with usage counts (facade to TagSearch)."""
         return self._tag_search.get_all_tags()

openspace/skill_engine/tag_search.py

@@ -18,7 +18,7 @@
 from contextlib import contextmanager
 from functools import wraps
 from pathlib import Path
-from typing import Any, Dict, Generator, List, Optional
+from typing import Any, Callable, Dict, Generator, List, Optional
 
 from openspace.utils.logging import Logger
 
@@ -32,12 +32,12 @@ def _db_retry(
     max_retries: int = 5,
     initial_delay: float = 0.1,
     backoff: float = 2.0,
-):
+) -> Callable:
     """Retry on transient SQLite errors with exponential backoff."""
 
-    def decorator(func):
+    def decorator(func: Callable) -> Callable:
         @wraps(func)
-        def wrapper(*args, **kwargs):
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
             delay = initial_delay
             for attempt in range(max_retries):
                 try:
@@ -221,6 +221,23 @@ def get_tags(self, skill_id: str) -> List[str]:
             ).fetchall()
             return [r["tag"] for r in rows]
 
+    @_db_retry()
+    def get_tags_batch(self, skill_ids: List[str]) -> Dict[str, List[str]]:
+        """Batch-get tags for multiple skills. Returns {skill_id: [sorted tags]}."""
+        if not skill_ids:
+            return {}
+        placeholders = ",".join("?" * len(skill_ids))
+        with self._reader() as conn:
+            rows = conn.execute(
+                f"SELECT skill_id, tag FROM skill_tags "
+                f"WHERE skill_id IN ({placeholders}) ORDER BY skill_id, tag",
+                skill_ids,
+            ).fetchall()
+        result: Dict[str, List[str]] = {}
+        for row in rows:
+            result.setdefault(row["skill_id"], []).append(row["tag"])
+        return result
+
     @_db_retry()
     def find_skills_by_tags(
         self,
@@ -448,8 +465,8 @@ def search_skills(
             List of skill records matching the criteria.
         """
         with self._reader() as conn:
-            conditions = []
-            params = []
+            conditions: List[str] = []
+            params: List[Any] = []
 
             # Active filter
             if active_only: