fix(3.4): address /8eyes R1 — duplicate rows, atomic bulk, layering, json safety

- Finding 1: Add DISTINCT to load_analyses/load_recent_analyses_for_skill to prevent duplicate rows from JOIN
- Finding 2: Validate ALL analyses before inserting any in shared-conn mode bulk_upsert for atomicity
- Finding 3: Make _insert_analysis public (insert_analysis) - fix store.py layering violation
- Finding 4: Add _safe_json_loads helper for json.loads with error handling - prevent crashes on corrupt DB data
- Finding 5: Fix docstring async/sync mismatch - remove await from examples

Tests: All 18 analysis store tests pass + 3 new tests for duplicate prevention, atomicity, public API
Full suite: 1231 passed (baseline maintained)

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

Author: Brian Krafft

openspace/skill_engine/analysis_store.py

@@ -108,7 +108,7 @@ class AnalysisStore:
     Usage (standalone)::
 
         store = AnalysisStore(db_path=Path("analyses.db"))
-        await store.record_execution_analysis(analysis)
+        store.record_execution_analysis(analysis)
         recent = store.load_analyses(skill_id="some_skill")
         store.close()
 
@@ -221,7 +221,7 @@ def record_execution_analysis_sync(self, analysis: ExecutionAnalysis) -> None:
         with self._mu:
             self._conn.execute("BEGIN")
             try:
-                analysis_id = self._insert_analysis(analysis)
+                analysis_id = self.insert_analysis(analysis)
                 self._conn.commit()
             except Exception:
                 self._conn.rollback()
@@ -243,8 +243,9 @@ def load_analyses(
         """
         with self._reader() as conn:
             if skill_id is not None:
+                # Use DISTINCT to avoid duplicate analyses when multiple judgments match
                 rows = conn.execute(
-                    "SELECT ea.* FROM execution_analyses ea "
+                    "SELECT DISTINCT ea.* FROM execution_analyses ea "
                     "JOIN skill_judgments sj ON ea.id = sj.analysis_id "
                     "WHERE sj.skill_id = ? "
                     "ORDER BY ea.timestamp DESC LIMIT ?",
@@ -300,8 +301,9 @@ def load_recent_analyses_for_skill(self, skill_id: str, limit: int = 5) -> List[
                 so filtering uses exact match.
         """
         with self._reader() as conn:
+            # Use DISTINCT to avoid duplicate analyses when multiple judgments match
             analysis_rows = conn.execute(
-                "SELECT ea.* FROM execution_analyses ea "
+                "SELECT DISTINCT ea.* FROM execution_analyses ea "
                 "JOIN skill_judgments sj ON ea.id = sj.analysis_id "
                 "WHERE sj.skill_id = ? "
                 "ORDER BY ea.timestamp DESC LIMIT ?",
@@ -361,21 +363,25 @@ def bulk_upsert_analyses(self, analyses: List[ExecutionAnalysis]) -> None:
                             (a.task_id,),
                         ).fetchone()
                         if existing is None:
-                            self._insert_analysis(a)
+                            self.insert_analysis(a)
                     self._conn.commit()
                 except Exception:
                     self._conn.rollback()
                     raise
         else:
             # Shared connection mode - assume caller has transaction and lock
-            # No mutex acquisition needed, just do the work
+            # Validate ALL analyses first before inserting any to ensure atomicity
+            for a in analyses:
+                a.validate()
+            
+            # Then insert all (caller's transaction guarantees atomicity)
             for a in analyses:
                 existing = self._conn.execute(
                     "SELECT id FROM execution_analyses WHERE task_id=?",
                     (a.task_id,),
                 ).fetchone()
                 if existing is None:
-                    self._insert_analysis(a)
+                    self.insert_analysis(a)
 
     @_db_retry()
     def clear_all_analyses(self) -> None:
@@ -435,7 +441,7 @@ def get_task_skill_summary(self, task_id: str) -> Dict[str, Any]:
                 "timestamp": row["timestamp"],
                 "task_completed": bool(row["task_completed"]),
                 "execution_note": row["execution_note"],
-                "tool_issues": json.loads(row["tool_issues"]),
+                "tool_issues": AnalysisStore._safe_json_loads(row["tool_issues"], []),
                 "candidate_for_evolution": bool(row["candidate_for_evolution"]),
                 "evolution_suggestions": evo_suggestions,
                 "analyzed_by": row["analyzed_by"],
@@ -464,9 +470,17 @@ def get_analysis_stats(self) -> Dict[str, int]:
                 "evolution_candidates": n_candidates,
             }
 
+    @staticmethod
+    def _safe_json_loads(json_str: str, default: Any) -> Any:
+        """Safely parse JSON string, returning default on error."""
+        try:
+            return json.loads(json_str)
+        except json.JSONDecodeError:
+            return default
+
     # ── Private Helpers ────────────────────────────────────────────────
 
-    def _insert_analysis(self, a: ExecutionAnalysis) -> int:
+    def insert_analysis(self, a: ExecutionAnalysis) -> int:
         """Insert an execution_analyses row + its skill_judgments.
 
         Called within a transaction holding ``self._mu``.
@@ -537,7 +551,7 @@ def _to_analysis(conn: sqlite3.Connection, row: sqlite3.Row) -> ExecutionAnalysi
             timestamp=datetime.fromisoformat(row["timestamp"]),
             task_completed=bool(row["task_completed"]),
             execution_note=row["execution_note"],
-            tool_issues=json.loads(row["tool_issues"]),
+            tool_issues=AnalysisStore._safe_json_loads(row["tool_issues"], []),
             skill_judgments=[
                 SkillJudgment(
                     skill_id=jr["skill_id"],

openspace/skill_engine/store.py

@@ -567,7 +567,7 @@ def _record_analysis_sync(self, analysis: ExecutionAnalysis) -> None:
             self._conn.execute("BEGIN")
             try:
                 # Delegate analysis storage to AnalysisStore (Epic 3.4)
-                self._analyses._insert_analysis(analysis)
+                self._analyses.insert_analysis(analysis)
 
                 # Update skill counters in skill_records (remains in SkillStore)
                 now_iso = datetime.now().isoformat()

tests/test_analysis_store.py

@@ -406,4 +406,124 @@ def test_analysis_store_integration_with_skill_store(temp_db_path):
     summary = store.get_task_skill_summary("integration_test") 
     assert summary["task_id"] == "integration_test"
 
+    store.close()
+
+
+def test_duplicate_judgments_no_duplicate_analysis_rows(temp_db_path):
+    """Test that multiple judgments for same analysis don't create duplicate analysis rows."""
+    store = AnalysisStore(db_path=temp_db_path)
+    
+    # Create analysis with multiple judgments for same skill
+    analysis = ExecutionAnalysis(
+        task_id="duplicate_judgment_test",
+        timestamp=datetime.now(),
+        skill_judgments=[
+            SkillJudgment(skill_id="shared_skill", skill_applied=True, note="First judgment"),
+            SkillJudgment(skill_id="shared_skill", skill_applied=False, note="Second judgment"),
+            SkillJudgment(skill_id="other_skill", skill_applied=True, note="Other judgment"),
+        ]
+    )
+    
+    store.record_execution_analysis_sync(analysis)
+    
+    # Load analyses for the shared skill - should only get one analysis back
+    analyses = store.load_analyses(skill_id="shared_skill", limit=10)
+    assert len(analyses) == 1, "Should not get duplicate analysis rows due to multiple judgments"
+    assert analyses[0].task_id == "duplicate_judgment_test"
+    
+    # Load recent analyses - should also not have duplicates  
+    recent_analyses = store.load_recent_analyses_for_skill("shared_skill", limit=10)
+    assert len(recent_analyses) == 1, "Should not get duplicate analysis rows in recent analyses"
+    
+    store.close()
+
+
+def test_bulk_upsert_atomicity_with_validation_error(temp_db_path):
+    """Test that bulk_upsert with mixed valid/invalid analyses is all-or-nothing."""
+    import sqlite3
+    import threading
+    
+    # Test in shared connection mode where atomicity issue exists
+    conn = sqlite3.connect(str(temp_db_path))
+    conn.row_factory = sqlite3.Row
+    lock = threading.Lock()
+    
+    # Initialize DDL
+    from openspace.skill_engine.analysis_store import _DDL
+    conn.executescript(_DDL)
+    
+    store = AnalysisStore(conn=conn, lock=lock)
+    
+    # Create mix of valid and invalid analyses
+    valid_analysis1 = ExecutionAnalysis(
+        task_id="valid_1",
+        timestamp=datetime.now(),
+        execution_note="Valid analysis 1"
+    )
+    
+    valid_analysis2 = ExecutionAnalysis(
+        task_id="valid_2", 
+        timestamp=datetime.now(),
+        execution_note="Valid analysis 2"
+    )
+    
+    # Create invalid analysis (missing required task_id)
+    invalid_analysis = ExecutionAnalysis(
+        task_id="",  # Invalid empty task_id
+        timestamp=datetime.now(),
+        execution_note="Invalid analysis"
+    )
+    
+    analyses = [valid_analysis1, invalid_analysis, valid_analysis2]
+    
+    # Bulk upsert should fail due to invalid analysis
+    with pytest.raises(Exception):  # Should raise validation error
+        with lock:
+            conn.execute("BEGIN")
+            try:
+                store.bulk_upsert_analyses(analyses)
+                conn.commit()
+            except Exception:
+                conn.rollback()
+                raise
+    
+    # Verify NO analyses were inserted (atomicity)
+    all_analyses = store.load_all_analyses()
+    assert len(all_analyses) == 0, "No analyses should be inserted when bulk operation fails"
+    
+    conn.close()
+
+
+def test_store_calls_public_insert_analysis_method(tmp_path):
+    """Test that store.py calls the public insert_analysis method (not private)."""
+    # This test verifies the layering fix - store.py should call public methods
+    temp_db_path = tmp_path / "test_public_method.db"
+    
+    from openspace.skill_engine.store import SkillStore
+    
+    store = SkillStore(db_path=temp_db_path)
+    
+    # Verify the AnalysisStore has public insert_analysis method
+    assert hasattr(store._analyses, 'insert_analysis'), "insert_analysis should be public"
+    
+    # Verify it's callable (not private)
+    assert callable(store._analyses.insert_analysis), "insert_analysis should be callable"
+    
+    # This should work without accessing private methods
+    analysis = ExecutionAnalysis(
+        task_id="public_method_test", 
+        timestamp=datetime.now(),
+        skill_judgments=[
+            SkillJudgment(skill_id="test_skill", skill_applied=True)
+        ]
+    )
+    
+    import asyncio
+    asyncio.run(store.record_analysis(analysis))
+    
+    # Verify it was recorded successfully
+    loaded = store.load_analyses_for_task("public_method_test")
+    assert loaded is not None
+    assert loaded.task_id == "public_method_test"
+    
     store.close()