"""Import no-credential Hermes Agent skills into Beaver.""" from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timezone import json import re import shutil from pathlib import Path from typing import Any from beaver.skills.catalog.utils import parse_frontmatter, strip_frontmatter from beaver.skills.specs import SkillSpec, SkillSpecStore, SkillVersion from beaver.skills.specs.serialization import canonical_hash, normalize_frontmatter, summarize_skill_content HERMES_REPO_URL = "https://github.com/NousResearch/hermes-agent" _CREDENTIAL_PATTERNS = [ re.compile(pattern, re.IGNORECASE) for pattern in [ r"\bapi[_ -]?key\b", r"\boauth\b", r"\bbearer\s+token\b", r"\baccess[_ -]?token\b", r"\bclient[_ -]?secret\b", r"\bsecret\b", r"\bcredential", r"\bspotify\b", r"\bdiscord\b", r"\bfeishu\b", r"\bhome\s*assistant\b", r"\bfal\b", r"\bopenrouter\b", r"\bwandb\b", ] ] @dataclass(slots=True) class HermesMigrationService: store: SkillSpecStore manifest_path: Path | None = None included_tools: list[dict[str, Any]] = field(default_factory=list) skipped_tools: list[dict[str, Any]] = field(default_factory=list) def migrate( self, repo_path: str | Path, *, include_optional: bool = True, dry_run: bool = False, ) -> dict[str, Any]: repo = Path(repo_path) if not repo.exists(): raise ValueError(f"Hermes repository not found: {repo}") skill_files = self._discover_skill_files(repo, include_optional=include_optional) included: list[dict[str, Any]] = [] skipped: list[dict[str, Any]] = [] for skill_file in skill_files: result = self._migrate_skill(repo, skill_file, dry_run=dry_run) if result["status"] in {"included", "unchanged"}: included.append(result) else: skipped.append(result) manifest = { "source": "hermes-agent", "repo_url": HERMES_REPO_URL, "repo_path": str(repo), "generated_at": datetime.now(timezone.utc).isoformat(), "dry_run": dry_run, "included": included, "skipped": skipped, "tools": self._tool_manifest(), } path = self.manifest_path or (self.store.workspace / "hermes_migration_manifest.json") path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return manifest def _discover_skill_files(self, repo: Path, *, include_optional: bool) -> list[Path]: roots = [repo / "skills"] if include_optional: roots.append(repo / "optional-skills") files: list[Path] = [] for root in roots: if root.exists(): files.extend(sorted(root.glob("**/SKILL.md"))) return files def _migrate_skill(self, repo: Path, skill_file: Path, *, dry_run: bool) -> dict[str, Any]: relative = skill_file.relative_to(repo) content = skill_file.read_text(encoding="utf-8") frontmatter, body = parse_frontmatter(content) skill_name = _safe_skill_name(str(frontmatter.get("name") or skill_file.parent.name)) if not skill_name: return _skip(relative, "unsafe_skill_name") credential_reason = _credential_reason(content) if credential_reason: return _skip(relative, credential_reason, skill_name=skill_name) normalized = normalize_frontmatter( { **frontmatter, "name": skill_name, "description": frontmatter.get("description") or skill_name, } ) rendered = _render_skill_content(normalized, body) content_hash = canonical_hash(rendered) existing = self.store.read_published_skill(skill_name) existing_spec = self.store.get_skill_spec(skill_name) if existing is not None and existing.version.content_hash == content_hash: return { "status": "unchanged", "skill_name": skill_name, "version": existing.version.version, "path": str(relative), "reason": "same_content_hash", } next_version = self._next_version(skill_name) if dry_run: return { "status": "included", "skill_name": skill_name, "version": next_version, "path": str(relative), "dry_run": True, } now = datetime.now(timezone.utc).isoformat() skill_version = SkillVersion( skill_name=skill_name, version=next_version, content_hash=content_hash, summary_hash=canonical_hash(strip_frontmatter(rendered).strip()), created_at=now, created_by="hermes_migration", change_reason=f"Import Hermes skill {relative}", parent_version=existing.version.version if existing is not None else None, review_state="published", frontmatter=normalized, summary=summarize_skill_content(body), tool_hints=self.store._extract_tool_hints(normalized), provenance={ "source": "hermes-agent", "repo_url": HERMES_REPO_URL, "repo_path": str(repo), "relative_path": str(relative), }, ) self.store.write_skill_version(skill_version, rendered) self._copy_supporting_files(skill_file.parent, skill_name, next_version) spec = existing_spec or SkillSpec( name=skill_name, display_name=skill_name, description=str(normalized.get("description") or skill_name), created_at=now, updated_at=now, current_version=next_version, status="active", tags=[], owners=["hermes-agent"], source_kind="hermes-agent", lineage=[], ) spec.current_version = next_version spec.updated_at = now spec.status = "active" spec.source_kind = "hermes-agent" if "hermes-agent" not in spec.owners: spec.owners.append("hermes-agent") self.store.write_skill_spec(spec) self.store.set_current_version(skill_name, next_version) published = self.store.read_index("published") if skill_name not in published: published.append(skill_name) self.store.update_index("published", published) return { "status": "included", "skill_name": skill_name, "version": next_version, "path": str(relative), } def _copy_supporting_files(self, source_dir: Path, skill_name: str, version: str) -> None: target_root = self.store.root / skill_name / "versions" / version for source in sorted(source_dir.rglob("*")): if not source.is_file() or source.name == "SKILL.md" or source.is_symlink(): continue relative = source.relative_to(source_dir) if any(part in {"", ".", ".."} for part in relative.parts): continue target = target_root / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(source, target) def _next_version(self, skill_name: str) -> str: versions = [item for item in self.store.list_versions(skill_name) if item.startswith("v")] numbers = [int(item[1:]) for item in versions if item[1:].isdigit()] return f"v{(max(numbers) if numbers else 0) + 1:04d}" def _tool_manifest(self) -> dict[str, list[dict[str, Any]]]: included = self.included_tools or [ {"name": "todo", "reason": "implemented_builtin_no_api"}, {"name": "clarify", "reason": "implemented_builtin_no_api"}, {"name": "delegate", "reason": "implemented_builtin_no_api"}, {"name": "spawn", "reason": "implemented_builtin_no_api"}, {"name": "skills_list", "reason": "implemented_builtin_no_api"}, {"name": "skill_manage", "reason": "implemented_builtin_no_api"}, {"name": "terminal", "reason": "implemented_builtin_no_api"}, {"name": "process", "reason": "implemented_builtin_no_api"}, {"name": "patch", "reason": "implemented_builtin_no_api"}, {"name": "write_file", "reason": "implemented_builtin_no_api"}, {"name": "web_fetch", "reason": "implemented_builtin_no_api"}, {"name": "web_search", "reason": "implemented_builtin_no_api"}, {"name": "execute_code", "reason": "implemented_builtin_no_api"}, ] skipped = self.skipped_tools or [ {"name": "spotify", "reason": "requires_oauth"}, {"name": "discord", "reason": "requires_external_token"}, {"name": "feishu", "reason": "requires_external_token"}, {"name": "home_assistant", "reason": "requires_external_service_credentials"}, {"name": "fal_image_generation", "reason": "requires_api_key"}, {"name": "remote_web_providers", "reason": "requires_api_key_or_oauth"}, ] return {"included": included, "skipped": skipped} def _credential_reason(content: str) -> str | None: for pattern in _CREDENTIAL_PATTERNS: if pattern.search(content): return "requires_external_credentials" return None def _safe_skill_name(value: str) -> str: cleaned = value.strip().replace(" ", "-") if not cleaned or cleaned in {".", ".."} or "/" in cleaned or "\\" in cleaned: return "" if not re.fullmatch(r"[A-Za-z0-9_.-]+", cleaned): return "" return cleaned def _skip(relative: Path, reason: str, *, skill_name: str | None = None) -> dict[str, Any]: result = {"status": "skipped", "path": str(relative), "reason": reason} if skill_name: result["skill_name"] = skill_name return result def _render_skill_content(frontmatter: dict[str, Any], body: str) -> str: lines = ["---"] for key, value in normalize_frontmatter(frontmatter).items(): if isinstance(value, list): lines.append(f"{key}:") for item in value: lines.append(f" - {item}") else: lines.append(f"{key}: {value}") lines.extend(["---", "", body.strip()]) return "\n".join(lines).rstrip() + "\n"