"""Skill mirroring and sync orchestration for declarative plugins.""" from __future__ import annotations from pathlib import Path from typing import Any from uuid import uuid4 from beaver.foundation.utils.file_lock import WorkspaceWriteLock from beaver.memory.skills.store import SkillLearningStore from beaver.plugins.models import PluginDiscoveryError, PluginManifest, PluginSkillBinding, PluginState from beaver.plugins.state import PluginStateStore from beaver.plugins.transaction import PluginSkillTransaction from beaver.skills.catalog.utils import parse_frontmatter, strip_frontmatter from beaver.skills.learning.safety import SkillDraftSafetyChecker from beaver.skills.publisher.service import SkillPublisher from beaver.skills.specs import SkillDraft, SkillReviewState, SkillSpec, SkillSpecStore, SkillStatus, SkillVersion from beaver.skills.specs.serialization import canonical_hash, normalize_frontmatter, summarize_skill_content class PluginManager: def __init__( self, *, workspace: Path, manifests: dict[str, PluginManifest], discovery_errors: list[PluginDiscoveryError], state_store: PluginStateStore, skill_store: SkillSpecStore, learning_store: SkillLearningStore, publisher: SkillPublisher, safety_checker: SkillDraftSafetyChecker, write_lock: WorkspaceWriteLock, ) -> None: self.workspace = Path(workspace) self.manifests = dict(manifests) self.discovery_errors = list(discovery_errors) self.state_store = state_store self.skill_store = skill_store self.learning_store = learning_store self.publisher = publisher self.safety_checker = safety_checker self.write_lock = write_lock def list_plugins(self) -> list[PluginState]: states = {state.plugin_id: state for state in self.state_store.list_plugins()} for plugin_id, manifest in self.manifests.items(): if plugin_id not in states: states[plugin_id] = PluginState( plugin_id=plugin_id, enabled=False, installed_version=None, manifest_path=manifest.display_path, status="discovered", ) return [states[key] for key in sorted(states)] def enable(self, plugin_id: str) -> PluginState: manifest = self.manifests.get(plugin_id) if manifest is None: raise ValueError(f"Unknown plugin: {plugin_id}") with self.write_lock.acquire(timeout_seconds=10): current_state = self.state_store.get_plugin(plugin_id) if current_state is not None and current_state.enabled and self._state_synced(current_state, manifest): return current_state transaction = PluginSkillTransaction(self.workspace) try: prepared = self._prepare_initial_mirror(manifest, transaction) for item in prepared: self.skill_store.promote_upstream_snapshot(transaction, item["snapshot"]) for item in prepared: self._publish_initial_mirror(item) state = PluginState( plugin_id=plugin_id, enabled=True, updates_paused=False, installed_version=manifest.version, manifest_path=manifest.display_path, status="synced", skills={ item["skill_name"]: PluginSkillBinding( accepted_upstream_tree_hash=item["snapshot"].skill_tree_hash, observed_upstream_tree_hash=item["snapshot"].skill_tree_hash, accepted_beaver_version=item["version"].version, current_beaver_version=item["version"].version, status="synced", ) for item in prepared }, ) self.state_store.upsert_plugin(state) return state finally: transaction.cleanup() def _prepare_initial_mirror( self, manifest: PluginManifest, transaction: PluginSkillTransaction, ) -> list[dict[str, Any]]: prepared: list[dict[str, Any]] = [] for declaration in manifest.skills: spec = self.skill_store.get_skill_spec(declaration.name) if spec is not None and spec.source_kind != "plugin": raise ValueError(f"Skill ownership conflict: {declaration.name}") snapshot = self.skill_store.stage_upstream_snapshot( transaction, skill_name=declaration.name, source_kind="plugin", source_id=manifest.plugin_id, source_version=manifest.version, source_path=declaration.relative_path, source_root=declaration.root, ) content = (declaration.root / "SKILL.md").read_text(encoding="utf-8") frontmatter, body = parse_frontmatter(content) draft = SkillDraft( draft_id=uuid4().hex, skill_name=declaration.name, base_version=None, proposed_content=body, proposed_frontmatter=normalize_frontmatter(frontmatter), created_at=_utc_now(), created_by="plugin-manager", reason=f"Initial mirror from plugin {manifest.plugin_id} {manifest.version}", proposal_kind="plugin_initial_mirror", ) safety = self.safety_checker.check(draft) if not safety.passed or safety.risk_level == "critical": raise ValueError(f"Plugin skill safety check failed: {declaration.name}") next_version = self._next_version(declaration.name) version = self._build_version( manifest=manifest, skill_name=declaration.name, version=next_version, content=content, frontmatter=normalize_frontmatter(frontmatter), parent_version=None, provenance={ "source_kind": "plugin", "plugin_id": manifest.plugin_id, "plugin_version": manifest.version, "plugin_skill_path": declaration.relative_path, "upstream_skill_content_hash": snapshot.skill_content_hash, "upstream_skill_tree_hash": snapshot.skill_tree_hash, "merge_mode": "initial_mirror", }, ) prepared.append( { "skill_name": declaration.name, "declaration": declaration, "snapshot": snapshot, "content": content, "frontmatter": normalize_frontmatter(frontmatter), "version": version, } ) return prepared def _publish_initial_mirror(self, item: dict[str, Any]) -> None: skill_name = str(item["skill_name"]) version: SkillVersion = item["version"] declaration = item["declaration"] content = str(item["content"]) self.skill_store.write_skill_version(version, content) self._copy_supporting_files(declaration.root, self.skill_store.root / skill_name / "versions" / version.version) version_dir = self.skill_store.root / skill_name / "versions" / version.version from beaver.plugins.hashing import hash_plugin_skill_tree version.tree_hash = hash_plugin_skill_tree(version_dir).skill_tree_hash self.skill_store._write_json(version_dir / "version.json", version.to_dict()) now = _utc_now() spec = self.skill_store.get_skill_spec(skill_name) if spec is None: spec = SkillSpec( name=skill_name, display_name=skill_name, description=str(version.frontmatter.get("description") or skill_name), created_at=now, updated_at=now, current_version=version.version, status=SkillStatus.ACTIVE.value, tags=[], owners=[], source_kind="plugin", lineage=[f"plugin:{version.provenance.get('plugin_id')}"], ) else: spec.current_version = version.version spec.updated_at = now spec.status = SkillStatus.ACTIVE.value spec.source_kind = "plugin" self.skill_store.write_skill_spec(spec) self.skill_store.set_current_version(skill_name, version.version) self.publisher._refresh_indexes(skill_name, spec.status) def _next_version(self, skill_name: str) -> str: versions = [item for item in self.skill_store.list_versions(skill_name) if item.startswith("v")] if not versions: return "v0001" numbers = [int(item[1:]) for item in versions if item[1:].isdigit()] return f"v{(max(numbers) if numbers else 0) + 1:04d}" def _build_version( self, *, manifest: PluginManifest, skill_name: str, version: str, content: str, frontmatter: dict[str, Any], parent_version: str | None, provenance: dict[str, Any], ) -> SkillVersion: body = strip_frontmatter(content).strip() return SkillVersion( skill_name=skill_name, version=version, content_hash=canonical_hash(content), summary_hash=canonical_hash(body), created_at=_utc_now(), created_by=f"plugin:{manifest.plugin_id}", change_reason=f"Initial mirror from plugin {manifest.plugin_id} {manifest.version}", parent_version=parent_version, review_state=SkillReviewState.PUBLISHED.value, frontmatter=normalize_frontmatter(frontmatter), summary=summarize_skill_content(body), tool_hints=self.skill_store._extract_tool_hints(frontmatter), provenance=dict(provenance), ) @staticmethod def _copy_supporting_files(source_root: Path, target_root: Path) -> None: for source in sorted(source_root.rglob("*"), key=lambda item: item.relative_to(source_root).as_posix()): relative = source.relative_to(source_root) if relative.as_posix() == "SKILL.md": continue if source.is_dir(): continue if source.is_symlink(): raise ValueError(f"Skill tree contains a symlink: {relative.as_posix()}") target = target_root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(source.read_bytes()) @staticmethod def _state_synced(state: PluginState, manifest: PluginManifest) -> bool: return ( state.status == "synced" and state.installed_version == manifest.version and all( binding.status == "synced" and binding.current_beaver_version for binding in state.skills.values() ) and len(state.skills) == len(manifest.skills) ) def _utc_now() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).isoformat()