feat(P3): Epic 3.1 — SkillVersion, ValidationError, validate() + /8eyes fixes (#12)

Deepfreezechill · web-flow · commit 54b7bf072956 · 2026-04-01T19:41:13.000-07:00
99+16 new tests (1142 total). SkillVersion with semver parsing, validation on all types, type-safe from_dict(), recursive validation, store-level enforcement. Closes #11
diff --git a/openspace/skill_engine/__init__.py b/openspace/skill_engine/__init__.py
@@ -13,7 +13,9 @@
     "SkillOrigin",
     "SkillLineage",
     "SkillRecord",
+    "SkillVersion",
     "SkillVisibility",
+    "ValidationError",
     "SkillStore",
     "ExecutionAnalyzer",
     "SkillEvolver",
@@ -34,7 +36,9 @@
     "SkillOrigin": (".types", "SkillOrigin"),
     "SkillLineage": (".types", "SkillLineage"),
     "SkillRecord": (".types", "SkillRecord"),
+    "SkillVersion": (".types", "SkillVersion"),
     "SkillVisibility": (".types", "SkillVisibility"),
+    "ValidationError": (".types", "ValidationError"),
     "SkillStore": (".store", "SkillStore"),
     "ExecutionAnalyzer": (".analyzer", "ExecutionAnalyzer"),
     "SkillEvolver": (".evolver", "SkillEvolver"),
diff --git a/openspace/skill_engine/store.py b/openspace/skill_engine/store.py
@@ -35,6 +35,7 @@
     SkillOrigin,
     SkillRecord,
     SkillVisibility,
+    ValidationError,
 )
 
 logger = Logger.get_logger(__name__)
@@ -1089,6 +1090,12 @@ def _upsert(self, record: SkillRecord) -> None:
 
         Called within a transaction holding ``self._mu``.
         """
+        try:
+            record.validate()
+        except ValidationError as exc:
+            raise ValidationError(
+                f"Cannot persist invalid SkillRecord '{record.skill_id}': {exc}"
+            ) from exc
         lin = record.lineage
         # content_snapshot is Dict[str, str]; store as JSON text
         snapshot_json = json.dumps(lin.content_snapshot, ensure_ascii=False)
@@ -1204,6 +1211,12 @@ def _insert_analysis(self, a: ExecutionAnalysis) -> int:
         Returns:
             int: The ``execution_analyses.id`` of the newly inserted row.
         """
+        try:
+            a.validate()
+        except ValidationError as exc:
+            raise ValidationError(
+                f"Cannot persist invalid ExecutionAnalysis '{a.task_id}': {exc}"
+            ) from exc
         cur = self._conn.execute(
             """
             INSERT INTO execution_analyses (
diff --git a/openspace/skill_engine/types.py b/openspace/skill_engine/types.py
@@ -2,12 +2,108 @@
 
 from __future__ import annotations
 
+import re
 from dataclasses import dataclass, field
 from datetime import datetime
 from enum import Enum
+from functools import total_ordering
 from typing import Any, ClassVar, Dict, List, Optional
 
 
+class ValidationError(Exception):
+    """Raised when a data type fails validation."""
+
+
+# ═══════════════════════════════════════════════════════════════════════
+#  SkillVersion — semantic versioning for skills
+# ═══════════════════════════════════════════════════════════════════════
+
+_VERSION_RE = re.compile(r"^(\d+)(?:\.(\d+))?(?:\.(\d+))?$")
+
+
+@total_ordering
+@dataclass(frozen=True)
+class SkillVersion:
+    """Semantic version (major.minor.patch) for skill revisions.
+
+    Supports parsing, comparison, bumping, and serialization.
+    """
+
+    major: int = 0
+    minor: int = 0
+    patch: int = 0
+
+    def __post_init__(self) -> None:
+        if self.major < 0 or self.minor < 0 or self.patch < 0:
+            raise ValueError(
+                f"Version components must be non-negative, got {self.major}.{self.minor}.{self.patch}"
+            )
+
+    # --- Parsing ---
+
+    @classmethod
+    def parse(cls, text: str) -> "SkillVersion":
+        """Parse a version string like ``'1.2.3'``, ``'1.2'``, or ``'1'``."""
+        m = _VERSION_RE.match(text.strip())
+        if not m:
+            raise ValueError(f"Invalid version string: {text!r}")
+        major = int(m.group(1))
+        minor = int(m.group(2)) if m.group(2) is not None else 0
+        patch = int(m.group(3)) if m.group(3) is not None else 0
+        return cls(major=major, minor=minor, patch=patch)
+
+    # --- Comparison ---
+
+    def _as_tuple(self) -> tuple[int, int, int]:
+        return (self.major, self.minor, self.patch)
+
+    def __eq__(self, other: object) -> bool:
+        if not isinstance(other, SkillVersion):
+            return NotImplemented
+        return self._as_tuple() == other._as_tuple()
+
+    def __lt__(self, other: object) -> bool:
+        if not isinstance(other, SkillVersion):
+            return NotImplemented
+        return self._as_tuple() < other._as_tuple()
+
+    def __hash__(self) -> int:
+        return hash(self._as_tuple())
+
+    # --- Bumping ---
+
+    def bump_patch(self) -> "SkillVersion":
+        return SkillVersion(self.major, self.minor, self.patch + 1)
+
+    def bump_minor(self) -> "SkillVersion":
+        return SkillVersion(self.major, self.minor + 1, 0)
+
+    def bump_major(self) -> "SkillVersion":
+        return SkillVersion(self.major + 1, 0, 0)
+
+    # --- Serialization ---
+
+    def __str__(self) -> str:
+        return f"{self.major}.{self.minor}.{self.patch}"
+
+    def __repr__(self) -> str:
+        return f"SkillVersion({self.major}, {self.minor}, {self.patch})"
+
+    def to_dict(self) -> Dict[str, int]:
+        return {"major": self.major, "minor": self.minor, "patch": self.patch}
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "SkillVersion":
+        try:
+            return cls(
+                major=int(data.get("major", 0)),
+                minor=int(data.get("minor", 0)),
+                patch=int(data.get("patch", 0)),
+            )
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid SkillVersion data: {exc}") from exc
+
+
 class SkillCategory(str, Enum):
     """Skill primary category."""
 
@@ -126,6 +222,28 @@ class SkillLineage:
     created_at: datetime = field(default_factory=datetime.now)
     created_by: str = ""  # "human" | model name (version-level actor)
 
+    def validate(self) -> None:
+        """Validate lineage rules by origin type."""
+        if self.generation < 0:
+            raise ValidationError(f"generation must be >= 0, got {self.generation}")
+        origin = self.origin
+        parents = self.parent_skill_ids
+        if origin in (SkillOrigin.IMPORTED, SkillOrigin.CAPTURED):
+            if parents:
+                raise ValidationError(
+                    f"{origin.value} lineage must have no parents, got {len(parents)}"
+                )
+        elif origin == SkillOrigin.FIXED:
+            if len(parents) != 1:
+                raise ValidationError(
+                    f"FIXED lineage must have exactly 1 parent, got {len(parents)}"
+                )
+        elif origin == SkillOrigin.DERIVED:
+            if not parents:
+                raise ValidationError(
+                    "DERIVED lineage must have at least 1 parent"
+                )
+
     def to_dict(self) -> Dict[str, Any]:
         return {
             "origin": self.origin.value,
@@ -173,6 +291,11 @@ class SkillJudgment:
     skill_applied: bool = False  # Whether the skill was actually applied
     note: str = ""  # Per-skill observation (deviation, usage, etc.)
 
+    def validate(self) -> None:
+        """Validate judgment fields."""
+        if not self.skill_id:
+            raise ValidationError("skill_id must be non-empty")
+
     def to_dict(self) -> Dict[str, Any]:
         return {
             "skill_id": self.skill_id,
@@ -205,6 +328,26 @@ class EvolutionSuggestion:
     category: Optional[SkillCategory] = None  # Desired category of the result
     direction: str = ""  # Free-text: what to evolve / capture
 
+    def validate(self) -> None:
+        """Validate target rules per evolution type."""
+        et = self.evolution_type
+        targets = self.target_skill_ids
+        if et == EvolutionType.FIX:
+            if len(targets) != 1:
+                raise ValidationError(
+                    f"FIX suggestion must have exactly 1 target, got {len(targets)}"
+                )
+        elif et == EvolutionType.DERIVED:
+            if not targets:
+                raise ValidationError(
+                    "DERIVED suggestion must have at least 1 target"
+                )
+        elif et == EvolutionType.CAPTURED:
+            if targets:
+                raise ValidationError(
+                    f"CAPTURED suggestion must have no targets, got {len(targets)}"
+                )
+
     @property
     def target_skill_id(self) -> str:
         """Primary (or only) target skill_id.  Empty string if none."""
@@ -287,6 +430,15 @@ def suggestions_by_type(self, evo_type: EvolutionType) -> List[EvolutionSuggesti
         """Filter evolution suggestions by type."""
         return [s for s in self.evolution_suggestions if s.evolution_type == evo_type]
 
+    def validate(self) -> None:
+        """Validate analysis fields and nested objects."""
+        if not self.task_id:
+            raise ValidationError("task_id must be non-empty")
+        for j in self.skill_judgments:
+            j.validate()
+        for s in self.evolution_suggestions:
+            s.validate()
+
     def to_dict(self) -> Dict[str, Any]:
         return {
             "task_id": self.task_id,
@@ -302,17 +454,20 @@ def to_dict(self) -> Dict[str, Any]:
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "ExecutionAnalysis":
-        return cls(
-            task_id=data["task_id"],
-            timestamp=datetime.fromisoformat(data["timestamp"]),
-            task_completed=data.get("task_completed", False),
-            execution_note=data.get("execution_note", ""),
-            tool_issues=data.get("tool_issues", []),
-            skill_judgments=[SkillJudgment.from_dict(j) for j in data.get("skill_judgments", [])],
-            evolution_suggestions=[EvolutionSuggestion.from_dict(s) for s in data.get("evolution_suggestions", [])],
-            analyzed_by=data.get("analyzed_by", ""),
-            analyzed_at=(datetime.fromisoformat(data["analyzed_at"]) if data.get("analyzed_at") else datetime.now()),
-        )
+        try:
+            return cls(
+                task_id=str(data["task_id"]),
+                timestamp=datetime.fromisoformat(data["timestamp"]),
+                task_completed=data.get("task_completed", False),
+                execution_note=data.get("execution_note", ""),
+                tool_issues=data.get("tool_issues", []),
+                skill_judgments=[SkillJudgment.from_dict(j) for j in data.get("skill_judgments", [])],
+                evolution_suggestions=[EvolutionSuggestion.from_dict(s) for s in data.get("evolution_suggestions", [])],
+                analyzed_by=data.get("analyzed_by", ""),
+                analyzed_at=(datetime.fromisoformat(data["analyzed_at"]) if data.get("analyzed_at") else datetime.now()),
+            )
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid ExecutionAnalysis data: {exc}") from exc
 
 
 # Full skill profile (identity + lineage + deps + quality)
@@ -385,6 +540,34 @@ def fallback_rate(self) -> float:
     # performed atomically in SQL by SkillStore.record_analysis().
     # Do NOT duplicate that logic here in Python.
 
+    def validate(self) -> None:
+        """Validate record fields and nested lineage."""
+        if not self.skill_id:
+            raise ValidationError("skill_id must be non-empty")
+        if not self.name:
+            raise ValidationError("name must be non-empty")
+        counters = {
+            "total_selections": self.total_selections,
+            "total_applied": self.total_applied,
+            "total_completions": self.total_completions,
+            "total_fallbacks": self.total_fallbacks,
+        }
+        for field_name, value in counters.items():
+            if value < 0:
+                raise ValidationError(f"{field_name} must be >= 0, got {value}")
+        if self.total_applied > self.total_selections:
+            raise ValidationError(
+                f"total_applied ({self.total_applied}) must not exceed "
+                f"total_selections ({self.total_selections})"
+            )
+        if self.total_completions > self.total_applied:
+            raise ValidationError(
+                f"total_completions ({self.total_completions}) must not exceed "
+                f"total_applied ({self.total_applied})"
+            )
+        if self.lineage is not None:
+            self.lineage.validate()
+
     def to_dict(self) -> Dict[str, Any]:
         return {
             "skill_id": self.skill_id,
@@ -410,30 +593,33 @@ def to_dict(self) -> Dict[str, Any]:
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "SkillRecord":
-        record = cls(
-            skill_id=data["skill_id"],
-            name=data["name"],
-            description=data.get("description", ""),
-            path=data.get("path", ""),
-            is_active=data.get("is_active", True),
-            category=SkillCategory(data["category"]) if data.get("category") else SkillCategory.WORKFLOW,
-            tags=data.get("tags", []),
-            visibility=(SkillVisibility(data["visibility"]) if data.get("visibility") else SkillVisibility.PRIVATE),
-            creator_id=data.get("creator_id", ""),
-            lineage=(
-                SkillLineage.from_dict(data["lineage"])
-                if data.get("lineage")
-                else SkillLineage(origin=SkillOrigin.IMPORTED)
-            ),
-            tool_dependencies=data.get("tool_dependencies", []),
-            critical_tools=data.get("critical_tools", []),
-            total_selections=data.get("total_selections", 0),
-            total_applied=data.get("total_applied", 0),
-            total_completions=data.get("total_completions", 0),
-            total_fallbacks=data.get("total_fallbacks", 0),
-            first_seen=(datetime.fromisoformat(data["first_seen"]) if data.get("first_seen") else datetime.now()),
-            last_updated=(datetime.fromisoformat(data["last_updated"]) if data.get("last_updated") else datetime.now()),
-        )
+        try:
+            record = cls(
+                skill_id=str(data["skill_id"]),
+                name=str(data["name"]),
+                description=data.get("description", ""),
+                path=data.get("path", ""),
+                is_active=data.get("is_active", True),
+                category=SkillCategory(data["category"]) if data.get("category") else SkillCategory.WORKFLOW,
+                tags=data.get("tags", []),
+                visibility=(SkillVisibility(data["visibility"]) if data.get("visibility") else SkillVisibility.PRIVATE),
+                creator_id=data.get("creator_id", ""),
+                lineage=(
+                    SkillLineage.from_dict(data["lineage"])
+                    if data.get("lineage")
+                    else SkillLineage(origin=SkillOrigin.IMPORTED)
+                ),
+                tool_dependencies=data.get("tool_dependencies", []),
+                critical_tools=data.get("critical_tools", []),
+                total_selections=int(data.get("total_selections", 0)),
+                total_applied=int(data.get("total_applied", 0)),
+                total_completions=int(data.get("total_completions", 0)),
+                total_fallbacks=int(data.get("total_fallbacks", 0)),
+                first_seen=(datetime.fromisoformat(data["first_seen"]) if data.get("first_seen") else datetime.now()),
+                last_updated=(datetime.fromisoformat(data["last_updated"]) if data.get("last_updated") else datetime.now()),
+            )
+        except (TypeError, ValueError) as exc:
+            raise ValidationError(f"Invalid SkillRecord data: {exc}") from exc
         for a in data.get("recent_analyses", []):
             record.recent_analyses.append(ExecutionAnalysis.from_dict(a))
         return record
diff --git a/tests/test_skill_types.py b/tests/test_skill_types.py
