purpose_agent/skills/schema.py

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