""" 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]