"""Structured models for Beaver skill lifecycle.""" from __future__ import annotations from dataclasses import dataclass, field from enum import Enum from typing import Any class SkillReviewState(str, Enum): DRAFT = "draft" IN_REVIEW = "in_review" APPROVED = "approved" REJECTED = "rejected" PUBLISHED = "published" DISABLED = "disabled" ARCHIVED = "archived" class SkillStatus(str, Enum): ACTIVE = "active" DISABLED = "disabled" ARCHIVED = "archived" @dataclass(slots=True) class SkillSpec: name: str display_name: str description: str created_at: str updated_at: str current_version: str | None status: str = SkillStatus.ACTIVE.value tags: list[str] = field(default_factory=list) owners: list[str] = field(default_factory=list) source_kind: str = "workspace" lineage: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return { "name": self.name, "display_name": self.display_name, "description": self.description, "created_at": self.created_at, "updated_at": self.updated_at, "current_version": self.current_version, "status": self.status, "tags": list(self.tags), "owners": list(self.owners), "source_kind": self.source_kind, "lineage": list(self.lineage), } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillSpec": return cls( name=str(payload["name"]), display_name=str(payload.get("display_name") or payload["name"]), description=str(payload.get("description") or payload.get("display_name") or payload["name"]), created_at=str(payload.get("created_at") or ""), updated_at=str(payload.get("updated_at") or payload.get("created_at") or ""), current_version=_coerce_optional_str(payload.get("current_version")), status=str(payload.get("status") or SkillStatus.ACTIVE.value), tags=_coerce_string_list(payload.get("tags")), owners=_coerce_string_list(payload.get("owners")), source_kind=str(payload.get("source_kind") or "workspace"), lineage=_coerce_string_list(payload.get("lineage")), ) @dataclass(slots=True) class SkillVersion: skill_name: str version: str content_hash: str summary_hash: str created_at: str created_by: str change_reason: str parent_version: str | None = None review_state: str = SkillReviewState.PUBLISHED.value frontmatter: dict[str, Any] = field(default_factory=dict) summary: str = "" tool_hints: list[str] = field(default_factory=list) provenance: dict[str, Any] = field(default_factory=dict) tree_hash: str = "" def to_dict(self) -> dict[str, Any]: return { "skill_name": self.skill_name, "version": self.version, "content_hash": self.content_hash, "summary_hash": self.summary_hash, "created_at": self.created_at, "created_by": self.created_by, "change_reason": self.change_reason, "parent_version": self.parent_version, "review_state": self.review_state, "frontmatter": dict(self.frontmatter), "summary": self.summary, "tool_hints": list(self.tool_hints), "provenance": dict(self.provenance), "tree_hash": self.tree_hash, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillVersion": return cls( skill_name=str(payload["skill_name"]), version=str(payload["version"]), content_hash=str(payload.get("content_hash") or ""), summary_hash=str(payload.get("summary_hash") or ""), created_at=str(payload.get("created_at") or ""), created_by=str(payload.get("created_by") or "unknown"), change_reason=str(payload.get("change_reason") or ""), parent_version=_coerce_optional_str(payload.get("parent_version")), review_state=str(payload.get("review_state") or SkillReviewState.PUBLISHED.value), frontmatter=dict(payload.get("frontmatter") or {}), summary=str(payload.get("summary") or ""), tool_hints=_coerce_string_list(payload.get("tool_hints")), provenance=dict(payload.get("provenance") or {}), tree_hash=str(payload.get("tree_hash") or ""), ) @dataclass(slots=True) class SkillUpstreamSnapshot: skill_name: str source_kind: str source_id: str source_version: str source_path: str skill_content_hash: str skill_tree_hash: str created_at: str frontmatter: dict[str, Any] = field(default_factory=dict) staged_root: Any | None = field(default=None, repr=False, compare=False) def to_dict(self) -> dict[str, Any]: return { "skill_name": self.skill_name, "source_kind": self.source_kind, "source_id": self.source_id, "source_version": self.source_version, "source_path": self.source_path, "skill_content_hash": self.skill_content_hash, "skill_tree_hash": self.skill_tree_hash, "created_at": self.created_at, "frontmatter": dict(self.frontmatter), } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillUpstreamSnapshot": return cls( skill_name=str(payload["skill_name"]), source_kind=str(payload.get("source_kind") or ""), source_id=str(payload.get("source_id") or ""), source_version=str(payload.get("source_version") or ""), source_path=str(payload.get("source_path") or ""), skill_content_hash=str(payload.get("skill_content_hash") or ""), skill_tree_hash=str(payload.get("skill_tree_hash") or ""), created_at=str(payload.get("created_at") or ""), frontmatter=dict(payload.get("frontmatter") or {}), ) @dataclass(slots=True) class SkillDraft: draft_id: str skill_name: str base_version: str | None proposed_content: str proposed_frontmatter: dict[str, Any] created_at: str created_by: str trigger_run_id: str | None = None trigger_session_id: str | None = None reason: str = "" status: str = SkillReviewState.DRAFT.value evidence_refs: list[dict[str, Any]] = field(default_factory=list) proposal_kind: str = "revise_skill" def to_dict(self) -> dict[str, Any]: return { "draft_id": self.draft_id, "skill_name": self.skill_name, "base_version": self.base_version, "proposed_content": self.proposed_content, "proposed_frontmatter": dict(self.proposed_frontmatter), "created_at": self.created_at, "created_by": self.created_by, "trigger_run_id": self.trigger_run_id, "trigger_session_id": self.trigger_session_id, "reason": self.reason, "status": self.status, "evidence_refs": list(self.evidence_refs), "proposal_kind": self.proposal_kind, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillDraft": return cls( draft_id=str(payload["draft_id"]), skill_name=str(payload["skill_name"]), base_version=_coerce_optional_str(payload.get("base_version")), proposed_content=str(payload.get("proposed_content") or ""), proposed_frontmatter=dict(payload.get("proposed_frontmatter") or {}), created_at=str(payload.get("created_at") or ""), created_by=str(payload.get("created_by") or "unknown"), trigger_run_id=_coerce_optional_str(payload.get("trigger_run_id")), trigger_session_id=_coerce_optional_str(payload.get("trigger_session_id")), reason=str(payload.get("reason") or ""), status=str(payload.get("status") or SkillReviewState.DRAFT.value), evidence_refs=list(payload.get("evidence_refs") or []), proposal_kind=str(payload.get("proposal_kind") or "revise_skill"), ) @dataclass(slots=True) class SkillReviewRecord: review_id: str draft_id: str skill_name: str requested_at: str requested_by: str status: str reviewer: str | None = None reviewed_at: str | None = None notes: str = "" def to_dict(self) -> dict[str, Any]: return { "review_id": self.review_id, "draft_id": self.draft_id, "skill_name": self.skill_name, "requested_at": self.requested_at, "requested_by": self.requested_by, "status": self.status, "reviewer": self.reviewer, "reviewed_at": self.reviewed_at, "notes": self.notes, } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillReviewRecord": return cls( review_id=str(payload["review_id"]), draft_id=str(payload["draft_id"]), skill_name=str(payload["skill_name"]), requested_at=str(payload.get("requested_at") or ""), requested_by=str(payload.get("requested_by") or "unknown"), status=str(payload.get("status") or SkillReviewState.IN_REVIEW.value), reviewer=_coerce_optional_str(payload.get("reviewer")), reviewed_at=_coerce_optional_str(payload.get("reviewed_at")), notes=str(payload.get("notes") or ""), ) @dataclass(slots=True) class SkillActivationReceipt: run_id: str session_id: str skill_name: str skill_version: str content_hash: str activated_at: str activation_reason: str tool_hints: list[str] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: return { "run_id": self.run_id, "session_id": self.session_id, "skill_name": self.skill_name, "skill_version": self.skill_version, "content_hash": self.content_hash, "activated_at": self.activated_at, "activation_reason": self.activation_reason, "tool_hints": list(self.tool_hints), } @classmethod def from_dict(cls, payload: dict[str, Any]) -> "SkillActivationReceipt": return cls( run_id=str(payload["run_id"]), session_id=str(payload["session_id"]), skill_name=str(payload["skill_name"]), skill_version=str(payload["skill_version"]), content_hash=str(payload.get("content_hash") or ""), activated_at=str(payload.get("activated_at") or ""), activation_reason=str(payload.get("activation_reason") or ""), tool_hints=_coerce_string_list(payload.get("tool_hints")), ) def _coerce_optional_str(value: Any) -> str | None: if value in (None, ""): return None return str(value) def _coerce_string_list(value: Any) -> list[str]: if not isinstance(value, list): return [] result: list[str] = [] for item in value: text = str(item).strip() if text: result.append(text) return result