Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 3)
1ccf652 verified | """ | |
| Skill — the atomic unit of learned behaviour | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| Defines the data model for every skill in the library. | |
| """ | |
| from __future__ import annotations | |
| import enum | |
| import time | |
| import uuid | |
| from dataclasses import dataclass, field | |
| from typing import Any, Dict, List, Optional | |
| class SkillLevel(enum.IntEnum): | |
| """ | |
| Hierarchical level of a skill. | |
| L0 → primitive atomic action (vm_mouse_move, execute_shell, ...) | |
| L1 → compound (2-10 atomics in sequence) | |
| L2 → task (named objective, uses L0+L1) | |
| L3 → workflow (sequence of tasks, has branching) | |
| L4 → role (persona — uses L0..L3, has long-term goals) | |
| """ | |
| ATOMIC = 0 | |
| COMPOUND = 1 | |
| TASK = 2 | |
| WORKFLOW = 3 | |
| ROLE = 4 | |
| class SkillStatus(enum.Enum): | |
| """Lifecycle state of a skill.""" | |
| PROPOSED = "proposed" # code generated but not yet validated | |
| VALIDATED = "validated" # passed all validators, safe to execute | |
| APPROVED = "approved" # blessed by Queen metacognition + pillar alignment | |
| ACTIVE = "active" # currently running / hot | |
| DEPRECATED = "deprecated" # superseded by a better skill | |
| BLOCKED = "blocked" # failed validation, do not run | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # SkillProposal — the raw output of the writer before validation | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class SkillProposal: | |
| """ | |
| A proposed skill that has been written but not yet validated. | |
| Flows: Observer → Writer → (SkillProposal) → Validator → Skill | |
| """ | |
| proposal_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) | |
| name: str = "" | |
| description: str = "" | |
| level: SkillLevel = SkillLevel.ATOMIC | |
| category: str = "general" | |
| code: str = "" | |
| entry_function: str = "" | |
| params_schema: Dict[str, Any] = field(default_factory=dict) | |
| dependencies: List[str] = field(default_factory=list) # other skill names | |
| observation_sources: List[str] = field(default_factory=list) # snapshot ids | |
| created_at: float = field(default_factory=time.time) | |
| created_by: str = "code_architect" | |
| reasoning: str = "" # why the writer proposed this | |
| target: str = "vm" # vm | local | either | |
| expression_context: Dict[str, Any] = field(default_factory=dict) | |
| def to_dict(self) -> Dict[str, Any]: | |
| return { | |
| "proposal_id": self.proposal_id, | |
| "name": self.name, | |
| "description": self.description, | |
| "level": int(self.level), | |
| "category": self.category, | |
| "code": self.code, | |
| "entry_function": self.entry_function, | |
| "params_schema": self.params_schema, | |
| "dependencies": self.dependencies, | |
| "observation_sources": self.observation_sources, | |
| "created_at": self.created_at, | |
| "created_by": self.created_by, | |
| "reasoning": self.reasoning, | |
| "target": self.target, | |
| "expression_context": self.expression_context, | |
| } | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| # Skill — validated and stored in the library | |
| # ───────────────────────────────────────────────────────────────────────────── | |
| class Skill: | |
| """A validated, executable skill.""" | |
| skill_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8]) | |
| name: str = "" | |
| description: str = "" | |
| level: SkillLevel = SkillLevel.ATOMIC | |
| category: str = "general" | |
| # Code | |
| code: str = "" # Python function source | |
| entry_function: str = "" # name of the function to call | |
| params_schema: Dict[str, Any] = field(default_factory=dict) | |
| dependencies: List[str] = field(default_factory=list) | |
| # Provenance | |
| created_at: float = field(default_factory=time.time) | |
| created_by: str = "code_architect" | |
| observation_sources: List[str] = field(default_factory=list) | |
| # Status | |
| status: SkillStatus = SkillStatus.PROPOSED | |
| target: str = "vm" # vm | local | either | |
| # Metacognitive feedback | |
| queen_verdict: Optional[str] = None # from Queen bridge | |
| queen_confidence: float = 0.0 | |
| pillar_alignment_score: float = 0.0 | |
| pillar_lighthouse: bool = False | |
| harmonic_signature: Dict[str, float] = field(default_factory=dict) | |
| # Execution stats | |
| execution_count: int = 0 | |
| success_count: int = 0 | |
| failure_count: int = 0 | |
| last_execution_at: Optional[float] = None | |
| last_error: Optional[str] = None | |
| mean_duration_s: float = 0.0 | |
| # Tags | |
| tags: List[str] = field(default_factory=list) | |
| expression_context: Dict[str, Any] = field(default_factory=dict) | |
| def success_rate(self) -> float: | |
| if self.execution_count == 0: | |
| return 0.0 | |
| return self.success_count / self.execution_count | |
| def confidence(self) -> float: | |
| """ | |
| Composite confidence = success_rate × queen_confidence × alignment. | |
| New skills start at 0 and earn confidence through execution. | |
| """ | |
| if self.execution_count == 0: | |
| return 0.5 * max(self.queen_confidence, 0.0) * max(self.pillar_alignment_score, 0.0) | |
| return self.success_rate * max(self.queen_confidence, 0.5) * max(self.pillar_alignment_score, 0.5) | |
| def to_dict(self) -> Dict[str, Any]: | |
| return { | |
| "skill_id": self.skill_id, | |
| "name": self.name, | |
| "description": self.description, | |
| "level": int(self.level), | |
| "level_name": self.level.name, | |
| "category": self.category, | |
| "code": self.code, | |
| "entry_function": self.entry_function, | |
| "params_schema": self.params_schema, | |
| "dependencies": self.dependencies, | |
| "created_at": self.created_at, | |
| "created_by": self.created_by, | |
| "observation_sources": self.observation_sources, | |
| "status": self.status.value, | |
| "target": self.target, | |
| "queen_verdict": self.queen_verdict, | |
| "queen_confidence": self.queen_confidence, | |
| "pillar_alignment_score": self.pillar_alignment_score, | |
| "pillar_lighthouse": self.pillar_lighthouse, | |
| "harmonic_signature": self.harmonic_signature, | |
| "execution_count": self.execution_count, | |
| "success_count": self.success_count, | |
| "failure_count": self.failure_count, | |
| "success_rate": round(self.success_rate, 4), | |
| "confidence": round(self.confidence, 4), | |
| "last_execution_at": self.last_execution_at, | |
| "last_error": self.last_error, | |
| "mean_duration_s": round(self.mean_duration_s, 4), | |
| "tags": self.tags, | |
| "expression_context": self.expression_context, | |
| } | |
| def from_dict(cls, data: Dict[str, Any]) -> "Skill": | |
| """Rehydrate a Skill from a dict (JSON load).""" | |
| skill = cls() | |
| skill.skill_id = data.get("skill_id", skill.skill_id) | |
| skill.name = data.get("name", "") | |
| skill.description = data.get("description", "") | |
| skill.level = SkillLevel(int(data.get("level", 0))) | |
| skill.category = data.get("category", "general") | |
| skill.code = data.get("code", "") | |
| skill.entry_function = data.get("entry_function", "") | |
| skill.params_schema = data.get("params_schema", {}) or {} | |
| skill.dependencies = data.get("dependencies", []) or [] | |
| skill.created_at = float(data.get("created_at", time.time())) | |
| skill.created_by = data.get("created_by", "code_architect") | |
| skill.observation_sources = data.get("observation_sources", []) or [] | |
| status_val = data.get("status", "proposed") | |
| try: | |
| skill.status = SkillStatus(status_val) | |
| except ValueError: | |
| skill.status = SkillStatus.PROPOSED | |
| skill.target = data.get("target", "vm") | |
| skill.queen_verdict = data.get("queen_verdict") | |
| skill.queen_confidence = float(data.get("queen_confidence", 0.0) or 0.0) | |
| skill.pillar_alignment_score = float(data.get("pillar_alignment_score", 0.0) or 0.0) | |
| skill.pillar_lighthouse = bool(data.get("pillar_lighthouse", False)) | |
| skill.harmonic_signature = data.get("harmonic_signature", {}) or {} | |
| skill.execution_count = int(data.get("execution_count", 0) or 0) | |
| skill.success_count = int(data.get("success_count", 0) or 0) | |
| skill.failure_count = int(data.get("failure_count", 0) or 0) | |
| skill.last_execution_at = data.get("last_execution_at") | |
| skill.last_error = data.get("last_error") | |
| skill.mean_duration_s = float(data.get("mean_duration_s", 0.0) or 0.0) | |
| skill.tags = data.get("tags", []) or [] | |
| skill.expression_context = data.get("expression_context", {}) or {} | |
| return skill | |
| def from_proposal(cls, proposal: SkillProposal) -> "Skill": | |
| """Build a Skill from a SkillProposal.""" | |
| return cls( | |
| name=proposal.name, | |
| description=proposal.description, | |
| level=proposal.level, | |
| category=proposal.category, | |
| code=proposal.code, | |
| entry_function=proposal.entry_function, | |
| params_schema=proposal.params_schema, | |
| dependencies=proposal.dependencies, | |
| created_by=proposal.created_by, | |
| observation_sources=proposal.observation_sources, | |
| target=proposal.target, | |
| status=SkillStatus.PROPOSED, | |
| expression_context=proposal.expression_context, | |
| ) | |