fix(3.2): address /8eyes findings — delegation scope, shared lock, LIKE escape, JSON safety

F1: Revert save_record/load_record/load_all delegation from SkillStore to
    SkillRepository — SkillStore._upsert and _to_record persist/hydrate
    recent_analyses which SkillRepository intentionally omits (Epic 3.4).
    Simple CRUD (delete, count, exists) remains delegated.

F2: SkillRepository now accepts an optional lock= parameter. SkillStore
    passes its own _mu so both share one mutex over the shared connection.

F3: search() escapes %, _, \\ in the name filter before LIKE via ESCAPE.

F4: json.loads(raw_snapshot) in both _to_record methods wrapped in
    try/except json.JSONDecodeError with fallback to {}.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Brian Krafft

openspace/skill_engine/skill_repository.py

@@ -136,16 +136,20 @@ class SkillRepository:
         db_path: Path to the SQLite database file.
         conn: Optional existing SQLite connection (for embedding in SkillStore).
                If provided, the repository will NOT own or close this connection.
+        lock: Optional :class:`threading.Lock` to use instead of creating a
+              private one.  When sharing a connection with another component
+              (e.g. ``SkillStore``), pass the same lock to avoid dual-mutex.
     """
 
     def __init__(
         self,
         db_path: Optional[Path] = None,
         conn: Optional[sqlite3.Connection] = None,
+        lock: Optional[threading.Lock] = None,
     ) -> None:
         self._owns_conn = conn is None
         self._closed = False
-        self._mu = threading.Lock()
+        self._mu = lock if lock is not None else threading.Lock()
 
         if conn is not None:
             self._conn = conn
@@ -349,8 +353,9 @@ def search(
                 conditions.append("sr.is_active = 1")
 
             if name is not None:
-                conditions.append("sr.name LIKE ?")
-                params.append(f"%{name}%")
+                escaped = name.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+                conditions.append("sr.name LIKE ? ESCAPE '\\'")
+                params.append(f"%{escaped}%")
 
             if category is not None:
                 conditions.append("sr.category = ?")
@@ -533,7 +538,10 @@ def _to_record(conn: sqlite3.Connection, row: sqlite3.Row) -> SkillRecord:
         ]
 
         raw_snapshot = row["lineage_content_snapshot"] or "{}"
-        snapshot: Dict[str, str] = json.loads(raw_snapshot)
+        try:
+            snapshot: Dict[str, str] = json.loads(raw_snapshot)
+        except json.JSONDecodeError:
+            snapshot = {}
 
         lineage = SkillLineage(
             origin=SkillOrigin(row["lineage_origin"]),

openspace/skill_engine/store.py

@@ -193,8 +193,9 @@ def __init__(self, db_path: Optional[Path] = None) -> None:
         self._conn = self._make_connection(read_only=False)
         self._init_db()
 
-        # CRUD repository (Epic 3.2) — delegates data-access operations
-        self._repo = SkillRepository(conn=self._conn)
+        # CRUD repository (Epic 3.2) — delegates simple CRUD operations.
+        # Shares our lock to avoid dual-mutex on the same connection.
+        self._repo = SkillRepository(conn=self._conn, lock=self._mu)
 
         logger.debug(f"SkillStore ready at {self._db_path}")
 
@@ -504,13 +505,38 @@ async def delete_record(self, skill_id: str) -> bool:
     # Sync write implementations (thread-safe via self._mu)
     @_db_retry()
     def _save_record_sync(self, record: SkillRecord) -> None:
+        """Persist a SkillRecord including its recent_analyses.
+
+        NOT delegated to SkillRepository — analyses persistence is
+        handled by SkillStore._upsert until Epic 3.4 (AnalysisStore).
+        """
         self._ensure_open()
-        self._repo.save(record)
+        with self._mu:
+            self._conn.execute("BEGIN")
+            try:
+                self._upsert(record)
+                self._conn.commit()
+            except Exception:
+                self._conn.rollback()
+                raise
 
     @_db_retry()
     def _save_records_sync(self, records: List[SkillRecord]) -> None:
+        """Batch persist SkillRecords including their recent_analyses.
+
+        NOT delegated to SkillRepository — analyses persistence is
+        handled by SkillStore._upsert until Epic 3.4 (AnalysisStore).
+        """
         self._ensure_open()
-        self._repo.save_many(records)
+        with self._mu:
+            self._conn.execute("BEGIN")
+            try:
+                for r in records:
+                    self._upsert(r)
+                self._conn.commit()
+            except Exception:
+                self._conn.rollback()
+                raise
 
     @_db_retry()
     def _record_analysis_sync(self, analysis: ExecutionAnalysis) -> None:
@@ -619,19 +645,41 @@ def _delete_record_sync(self, skill_id: str) -> bool:
     # Read API (sync, each call opens its own read-only conn)
     @_db_retry()
     def load_record(self, skill_id: str) -> Optional[SkillRecord]:
-        """Load a single :class:`SkillRecord` by id."""
-        return self._repo.get(skill_id)
+        """Load a single :class:`SkillRecord` by id, including recent_analyses.
+
+        NOT delegated to SkillRepository — analysis hydration is
+        handled by SkillStore._to_record until Epic 3.4 (AnalysisStore).
+        """
+        with self._reader() as conn:
+            row = conn.execute(
+                "SELECT * FROM skill_records WHERE skill_id=?",
+                (skill_id,),
+            ).fetchone()
+            return self._to_record(conn, row) if row else None
 
     @_db_retry()
     def load_all(self, *, active_only: bool = False) -> Dict[str, SkillRecord]:
-        """Load skill records, keyed by ``skill_id``.
+        """Load skill records, keyed by ``skill_id``, including recent_analyses.
+
+        NOT delegated to SkillRepository — analysis hydration is
+        handled by SkillStore._to_record until Epic 3.4 (AnalysisStore).
 
         Args:
             active_only: If True, only return records with ``is_active=True``.
         """
-        result = self._repo.list_all(active_only=active_only)
-        logger.info(f"Loaded {len(result)} skill records (active_only={active_only})")
-        return result
+        with self._reader() as conn:
+            if active_only:
+                rows = conn.execute(
+                    "SELECT * FROM skill_records WHERE is_active=1"
+                ).fetchall()
+            else:
+                rows = conn.execute("SELECT * FROM skill_records").fetchall()
+            result: Dict[str, SkillRecord] = {}
+            for row in rows:
+                rec = self._to_record(conn, row)
+                result[rec.skill_id] = rec
+            logger.info(f"Loaded {len(result)} skill records (active_only={active_only})")
+            return result
 
     @_db_retry()
     def load_active(self) -> Dict[str, SkillRecord]:
@@ -1240,7 +1288,10 @@ def _to_record(self, conn: sqlite3.Connection, row: sqlite3.Row) -> SkillRecord:
         # Deserialize content_snapshot: stored as JSON dict
         # mapping relative file paths to their text content
         raw_snapshot = row["lineage_content_snapshot"] or "{}"
-        snapshot: Dict[str, str] = json.loads(raw_snapshot)
+        try:
+            snapshot: Dict[str, str] = json.loads(raw_snapshot)
+        except json.JSONDecodeError:
+            snapshot = {}
 
         lineage = SkillLineage(
             origin=SkillOrigin(row["lineage_origin"]),

tests/test_skill_repository.py

@@ -388,3 +388,134 @@ def test_close_prevents_further_operations(self, repo):
         repo.close()
         with pytest.raises(RuntimeError):
             repo.get("anything")
+
+
+# ═══════════════════════════════════════════════════════════════════════
+#  F2: Shared Lock (dual-mutex fix)
+# ═══════════════════════════════════════════════════════════════════════
+
+
+class TestSharedLock:
+    """When given an external lock, SkillRepository should use it."""
+
+    def test_uses_provided_lock(self, tmp_path):
+        from openspace.skill_engine.skill_repository import SkillRepository
+        import threading
+
+        external_lock = threading.Lock()
+        db_path = tmp_path / "shared_lock.db"
+        repo = SkillRepository(db_path=db_path, lock=external_lock)
+        assert repo._mu is external_lock
+        repo.close()
+
+    def test_creates_own_lock_when_none_provided(self, tmp_path):
+        from openspace.skill_engine.skill_repository import SkillRepository
+        import threading
+
+        db_path = tmp_path / "own_lock.db"
+        repo = SkillRepository(db_path=db_path)
+        assert isinstance(repo._mu, type(threading.Lock()))
+        repo.close()
+
+    def test_shared_conn_and_lock(self, tmp_path):
+        """Shared connection + shared lock should work without deadlock."""
+        from openspace.skill_engine.skill_repository import SkillRepository
+        import sqlite3
+        import threading
+
+        db_path = tmp_path / "shared.db"
+        conn = sqlite3.connect(str(db_path), check_same_thread=False)
+        conn.row_factory = sqlite3.Row
+        lock = threading.Lock()
+
+        repo = SkillRepository(conn=conn, lock=lock)
+        # Init tables manually since shared conn skips _init_db
+        repo._owns_conn = True  # temporarily to init
+        repo._db_path = db_path
+        repo._init_db()
+        repo._owns_conn = False
+        repo._db_path = Path(":shared:")
+
+        record = _make_record()
+        repo.save(record)
+        loaded = repo.get(record.skill_id)
+        assert loaded is not None
+        assert loaded.skill_id == record.skill_id
+
+        repo.close()
+        conn.close()
+
+
+# ═══════════════════════════════════════════════════════════════════════
+#  F3: LIKE Wildcard Escaping
+# ═══════════════════════════════════════════════════════════════════════
+
+
+class TestSearchEscaping:
+    """search() must escape SQL LIKE wildcards in user input."""
+
+    def test_percent_in_name_does_not_match_broadly(self, repo):
+        repo.save_many([
+            _make_record(skill_id="s1", name="100%_done"),
+            _make_record(skill_id="s2", name="100_other_tasks"),
+            _make_record(skill_id="s3", name="something_else"),
+        ])
+
+        results = repo.search(name="100%_done")
+        assert len(results) == 1
+        assert results[0].skill_id == "s1"
+
+    def test_underscore_in_name_is_literal(self, repo):
+        repo.save_many([
+            _make_record(skill_id="s1", name="my_skill"),
+            _make_record(skill_id="s2", name="myXskill"),
+        ])
+
+        # Without escaping, "_" would match any single char including "X"
+        results = repo.search(name="my_skill")
+        assert len(results) == 1
+        assert results[0].skill_id == "s1"
+
+    def test_backslash_in_name_is_literal(self, repo):
+        repo.save(_make_record(skill_id="s1", name="path\\to\\skill"))
+        results = repo.search(name="path\\to")
+        assert len(results) == 1
+
+
+# ═══════════════════════════════════════════════════════════════════════
+#  F4: JSON Safety in _to_record
+# ═══════════════════════════════════════════════════════════════════════
+
+
+class TestJsonSafety:
+    """_to_record should survive malformed JSON in lineage_content_snapshot."""
+
+    def test_malformed_snapshot_falls_back_to_empty_dict(self, repo):
+        """Insert a record, corrupt the JSON, then load it."""
+        record = _make_record(skill_id="corrupt1", name="corrupt_test")
+        repo.save(record)
+
+        # Directly corrupt the snapshot column
+        repo._conn.execute(
+            "UPDATE skill_records SET lineage_content_snapshot = ? WHERE skill_id = ?",
+            ("NOT VALID JSON {{{", "corrupt1"),
+        )
+        repo._conn.commit()
+
+        loaded = repo.get("corrupt1")
+        assert loaded is not None
+        assert loaded.lineage.content_snapshot == {}
+
+    def test_empty_snapshot_falls_back_to_empty_dict(self, repo):
+        record = _make_record(skill_id="empty1", name="empty_test")
+        repo.save(record)
+
+        repo._conn.execute(
+            "UPDATE skill_records SET lineage_content_snapshot = '' WHERE skill_id = ?",
+            ("empty1",),
+        )
+        repo._conn.commit()
+
+        loaded = repo.get("empty1")
+        assert loaded is not None
+        assert loaded.lineage.content_snapshot == {}