File size: 5,728 Bytes
9980c16 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 | """
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]
|