208 lines
7.5 KiB
Python
208 lines
7.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Manage builtin/customized/active skill directories for each run."""
|
|
|
|
from pathlib import Path
|
|
import shutil
|
|
from typing import Dict, Iterable, List
|
|
|
|
import yaml
|
|
|
|
from backend.config.bootstrap_config import get_bootstrap_config_for_run
|
|
|
|
|
|
class SkillsManager:
|
|
"""Sync named skills into a run-scoped active skills workspace."""
|
|
|
|
def __init__(self, project_root: Path | None = None):
|
|
self.project_root = (
|
|
project_root or Path(__file__).resolve().parents[2]
|
|
)
|
|
self.builtin_root = self.project_root / "backend" / "skills" / "builtin"
|
|
self.customized_root = (
|
|
self.project_root / "backend" / "skills" / "customized"
|
|
)
|
|
self.runs_root = self.project_root / "runs"
|
|
|
|
def get_active_root(self, config_name: str) -> Path:
|
|
return self.runs_root / config_name / "skills" / "active"
|
|
|
|
def get_activation_manifest_path(self, config_name: str) -> Path:
|
|
return self.runs_root / config_name / "skills" / "activation.yaml"
|
|
|
|
def get_agent_asset_dir(self, config_name: str, agent_id: str) -> Path:
|
|
return self.runs_root / config_name / "agents" / agent_id
|
|
|
|
def ensure_activation_manifest(self, config_name: str) -> Path:
|
|
manifest_path = self.get_activation_manifest_path(config_name)
|
|
manifest_path.parent.mkdir(parents=True, exist_ok=True)
|
|
if not manifest_path.exists():
|
|
manifest_path.write_text(
|
|
"global_enabled_skills: []\n"
|
|
"global_disabled_skills: []\n"
|
|
"agent_enabled_skills: {}\n"
|
|
"agent_disabled_skills: {}\n",
|
|
encoding="utf-8",
|
|
)
|
|
return manifest_path
|
|
|
|
def load_activation_manifest(self, config_name: str) -> Dict[str, object]:
|
|
manifest_path = self.ensure_activation_manifest(config_name)
|
|
with open(manifest_path, "r", encoding="utf-8") as file:
|
|
parsed = yaml.safe_load(file) or {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
|
|
def _resolve_source_dir(self, skill_name: str) -> Path:
|
|
customized_dir = self.customized_root / skill_name
|
|
if customized_dir.exists():
|
|
return customized_dir
|
|
|
|
builtin_dir = self.builtin_root / skill_name
|
|
if builtin_dir.exists():
|
|
return builtin_dir
|
|
|
|
raise FileNotFoundError(f"Unknown skill: {skill_name}")
|
|
|
|
def _persist_runtime_edits(
|
|
self,
|
|
config_name: str,
|
|
skill_name: str,
|
|
active_dir: Path,
|
|
) -> None:
|
|
"""
|
|
Persist run-time edits from active skills into customized skills.
|
|
|
|
This keeps active skill experiments from being lost on the next reload
|
|
while still allowing the active directory to be re-synced cleanly.
|
|
"""
|
|
if not active_dir.exists():
|
|
return
|
|
|
|
source_dir = self._resolve_source_dir(skill_name)
|
|
if active_dir.resolve() == source_dir.resolve():
|
|
return
|
|
|
|
if not self._directories_match(active_dir, source_dir):
|
|
customized_dir = self.customized_root / skill_name
|
|
customized_dir.parent.mkdir(parents=True, exist_ok=True)
|
|
if customized_dir.exists():
|
|
shutil.rmtree(customized_dir)
|
|
shutil.copytree(active_dir, customized_dir)
|
|
|
|
@staticmethod
|
|
def _directories_match(left: Path, right: Path) -> bool:
|
|
"""Compare two directory trees by file contents."""
|
|
if not left.exists() or not right.exists():
|
|
return False
|
|
|
|
left_items = sorted(
|
|
path.relative_to(left)
|
|
for path in left.rglob("*")
|
|
)
|
|
right_items = sorted(
|
|
path.relative_to(right)
|
|
for path in right.rglob("*")
|
|
)
|
|
if left_items != right_items:
|
|
return False
|
|
|
|
for relative_path in left_items:
|
|
left_path = left / relative_path
|
|
right_path = right / relative_path
|
|
if left_path.is_dir() != right_path.is_dir():
|
|
return False
|
|
if left_path.is_file():
|
|
if left_path.read_bytes() != right_path.read_bytes():
|
|
return False
|
|
return True
|
|
|
|
def resolve_agent_skill_names(
|
|
self,
|
|
config_name: str,
|
|
agent_id: str,
|
|
default_skills: Iterable[str],
|
|
) -> List[str]:
|
|
"""Resolve final skill names after bootstrap and activation overlays."""
|
|
bootstrap = get_bootstrap_config_for_run(self.project_root, config_name)
|
|
override = bootstrap.agent_override(agent_id)
|
|
skills = list(override.get("skills", list(default_skills)))
|
|
|
|
manifest = self.load_activation_manifest(config_name)
|
|
for skill_name in manifest.get("global_enabled_skills", []):
|
|
if skill_name not in skills:
|
|
skills.append(skill_name)
|
|
|
|
for skill_name in manifest.get("agent_enabled_skills", {}).get(agent_id, []):
|
|
if skill_name not in skills:
|
|
skills.append(skill_name)
|
|
|
|
disabled = set(manifest.get("global_disabled_skills", []))
|
|
disabled.update(
|
|
manifest.get("agent_disabled_skills", {}).get(agent_id, []),
|
|
)
|
|
|
|
return [skill for skill in skills if skill not in disabled]
|
|
|
|
def sync_active_skills(
|
|
self,
|
|
config_name: str,
|
|
skill_names: Iterable[str],
|
|
) -> List[Path]:
|
|
"""Sync selected skills into the run workspace and return their paths."""
|
|
active_root = self.get_active_root(config_name)
|
|
active_root.mkdir(parents=True, exist_ok=True)
|
|
|
|
synced_paths: List[Path] = []
|
|
wanted = set(skill_names)
|
|
|
|
for existing in active_root.iterdir():
|
|
if existing.is_dir() and existing.name not in wanted:
|
|
self._persist_runtime_edits(
|
|
config_name=config_name,
|
|
skill_name=existing.name,
|
|
active_dir=existing,
|
|
)
|
|
shutil.rmtree(existing)
|
|
|
|
for skill_name in skill_names:
|
|
source_dir = self._resolve_source_dir(skill_name)
|
|
target_dir = active_root / skill_name
|
|
if target_dir.exists():
|
|
self._persist_runtime_edits(
|
|
config_name=config_name,
|
|
skill_name=skill_name,
|
|
active_dir=target_dir,
|
|
)
|
|
shutil.rmtree(target_dir)
|
|
shutil.copytree(source_dir, target_dir)
|
|
synced_paths.append(target_dir)
|
|
|
|
return synced_paths
|
|
|
|
def prepare_active_skills(
|
|
self,
|
|
config_name: str,
|
|
agent_defaults: Dict[str, Iterable[str]],
|
|
) -> Dict[str, List[Path]]:
|
|
"""Resolve all agent skills, sync the union once, and map paths per agent."""
|
|
resolved: Dict[str, List[str]] = {}
|
|
union: List[str] = []
|
|
|
|
for agent_id, default_skills in agent_defaults.items():
|
|
resolved_skills = self.resolve_agent_skill_names(
|
|
config_name=config_name,
|
|
agent_id=agent_id,
|
|
default_skills=default_skills,
|
|
)
|
|
resolved[agent_id] = resolved_skills
|
|
for skill_name in resolved_skills:
|
|
if skill_name not in union:
|
|
union.append(skill_name)
|
|
|
|
self.sync_active_skills(config_name=config_name, skill_names=union)
|
|
active_root = self.get_active_root(config_name)
|
|
|
|
return {
|
|
agent_id: [active_root / skill_name for skill_name in skill_names]
|
|
for agent_id, skill_names in resolved.items()
|
|
}
|