Sprint 7: skills/schema.py — SkillCard + SkillGenome with versioning

purpose_agent/skills/schema.py

@@ -0,0 +1,140 @@
+"""
+schema.py — SkillCard and SkillGenome data structures.
+
+A SkillCard is the atomic unit of learned procedural knowledge.
+A SkillGenome tracks the evolutionary lineage of a skill across versions.
+"""
+from __future__ import annotations
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class SkillCard:
+    """
+    A single versioned skill — a learned, testable procedure.
+    
+    Lifecycle: candidate → tested → active → (evolved | hibernated | archived)
+    """
+    id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
+    name: str = ""
+    trigger: str = ""                        # When to apply this skill
+    applies_when: list[str] = field(default_factory=list)  # Context conditions
+    procedure: list[str] = field(default_factory=list)     # Step-by-step
+    examples: list[dict[str, Any]] = field(default_factory=list)  # Input→output examples
+    tools: list[str] = field(default_factory=list)         # Tools this skill uses
+    evals: list[str] = field(default_factory=list)         # Test case IDs
+    fitness_score: float = 0.5              # 0-1, updated by CI
+    version: int = 1
+    parent_id: str | None = None            # Previous version's ID
+    source_memory_ids: list[str] = field(default_factory=list)
+    created_at: float = field(default_factory=time.time)
+    created_by: str = ""                    # "consolidation", "retroformer", "human", etc.
+    status: str = "candidate"               # candidate, tested, active, hibernated, archived
+    exportable: bool = True                 # Can be shared via A2A
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "id": self.id, "name": self.name, "trigger": self.trigger,
+            "applies_when": self.applies_when, "procedure": self.procedure,
+            "examples": self.examples, "tools": self.tools, "evals": self.evals,
+            "fitness_score": self.fitness_score, "version": self.version,
+            "parent_id": self.parent_id, "source_memory_ids": self.source_memory_ids,
+            "created_at": self.created_at, "created_by": self.created_by,
+            "status": self.status, "exportable": self.exportable,
+        }
+
+    @classmethod
+    def from_dict(cls, d: dict) -> "SkillCard":
+        return cls(**{k: v for k, v in d.items() if k in cls.__dataclass_fields__})
+
+    def to_markdown(self) -> str:
+        """Export as human-readable markdown."""
+        lines = [f"# Skill: {self.name} (v{self.version})", ""]
+        lines.append(f"**Trigger:** {self.trigger}")
+        if self.applies_when:
+            lines.append(f"**Applies when:** {', '.join(self.applies_when)}")
+        lines.append(f"\n## Procedure")
+        for i, step in enumerate(self.procedure, 1):
+            lines.append(f"{i}. {step}")
+        if self.tools:
+            lines.append(f"\n**Tools:** {', '.join(self.tools)}")
+        lines.append(f"\n**Fitness:** {self.fitness_score:.2f} | **Status:** {self.status}")
+        return "\n".join(lines)
+
+    def evolve(self, new_procedure: list[str] | None = None, new_trigger: str | None = None) -> "SkillCard":
+        """Create a new version of this skill (mutation)."""
+        return SkillCard(
+            name=self.name,
+            trigger=new_trigger or self.trigger,
+            applies_when=list(self.applies_when),
+            procedure=new_procedure or list(self.procedure),
+            examples=list(self.examples),
+            tools=list(self.tools),
+            evals=list(self.evals),
+            fitness_score=self.fitness_score * 0.9,  # Slight penalty for unproven mutation
+            version=self.version + 1,
+            parent_id=self.id,
+            source_memory_ids=list(self.source_memory_ids),
+            created_by="evolution",
+            status="candidate",
+        )
+
+
+@dataclass
+class SkillGenome:
+    """
+    Tracks the evolutionary lineage of a skill across versions.
+    
+    Supports: version history, rollback, fitness tracking over time.
+    """
+    skill_name: str
+    versions: list[SkillCard] = field(default_factory=list)
+    active_version_id: str | None = None
+
+    def add_version(self, card: SkillCard) -> None:
+        self.versions.append(card)
+
+    def promote(self, card_id: str) -> bool:
+        """Promote a version to active."""
+        for v in self.versions:
+            if v.id == card_id:
+                # Deactivate current
+                if self.active_version_id:
+                    for vv in self.versions:
+                        if vv.id == self.active_version_id:
+                            vv.status = "hibernated"
+                v.status = "active"
+                self.active_version_id = card_id
+                return True
+        return False
+
+    def rollback(self) -> SkillCard | None:
+        """Rollback to previous version."""
+        if not self.active_version_id:
+            return None
+        current = next((v for v in self.versions if v.id == self.active_version_id), None)
+        if current and current.parent_id:
+            parent = next((v for v in self.versions if v.id == current.parent_id), None)
+            if parent:
+                current.status = "archived"
+                parent.status = "active"
+                self.active_version_id = parent.id
+                return parent
+        return None
+
+    @property
+    def active(self) -> SkillCard | None:
+        if self.active_version_id:
+            return next((v for v in self.versions if v.id == self.active_version_id), None)
+        return None
+
+    @property
+    def version_count(self) -> int:
+        return len(self.versions)
+
+    @property
+    def fitness_history(self) -> list[float]:
+        return [v.fitness_score for v in self.versions]