Some checks are pending
CI / lint-and-test (push) Waiting to run
* make Leo skill onboarding clean-context reproducible * harden Leo clean-context skill acceptance
106 lines
4.2 KiB
Python
Executable file
106 lines
4.2 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Install the manifest-indexed Leo/Teleo skills into an empty skill root."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
DEFAULT_MANIFEST = Path("docs/reports/leo-working-state-20260709/skill-pack-manifest.json")
|
|
SKILL_NAME = re.compile(r"[a-z0-9-]{1,64}")
|
|
|
|
|
|
def _tracked_skill_files(root: Path, skill_dir: Path) -> list[Path]:
|
|
relative_dir = skill_dir.relative_to(root)
|
|
result = subprocess.run(
|
|
["git", "-C", str(root), "ls-files", "-z", "--", str(relative_dir)],
|
|
check=False,
|
|
capture_output=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stderr.decode("utf-8", errors="replace"))
|
|
files = [root / Path(item.decode("utf-8")) for item in result.stdout.split(b"\0") if item]
|
|
if not files:
|
|
raise ValueError(f"skill has no tracked files: {relative_dir}")
|
|
for source in files:
|
|
if source.is_symlink():
|
|
raise ValueError(f"skill contains a symlink: {source.relative_to(root)}")
|
|
if not source.is_file():
|
|
raise ValueError(f"skill route is not a regular file: {source.relative_to(root)}")
|
|
source.resolve().relative_to(skill_dir.resolve())
|
|
return files
|
|
|
|
|
|
def install(root: Path, manifest_path: Path, target: Path) -> dict[str, object]:
|
|
manifest_file = root / manifest_path
|
|
manifest = json.loads(manifest_file.read_text(encoding="utf-8"))
|
|
skills = manifest["skills"]
|
|
if target.exists():
|
|
raise ValueError(f"target already exists: {target}")
|
|
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
stage = Path(tempfile.mkdtemp(prefix=".teleo-skill-pack-", dir=target.parent))
|
|
installed: list[str] = []
|
|
try:
|
|
for entry in skills:
|
|
name = entry["name"]
|
|
if not isinstance(name, str) or SKILL_NAME.fullmatch(name) is None:
|
|
raise ValueError(f"invalid skill name: {name!r}")
|
|
expected_path = Path(".agents") / "skills" / name / "SKILL.md"
|
|
if Path(entry["path"]) != expected_path:
|
|
raise ValueError(f"skill path must be {expected_path}: {entry['path']!r}")
|
|
if name in installed:
|
|
raise ValueError(f"duplicate skill name: {name}")
|
|
|
|
source = root / expected_path.parent
|
|
source.resolve().relative_to((root / ".agents" / "skills").resolve())
|
|
if not source.is_dir():
|
|
raise FileNotFoundError(source)
|
|
destination = stage / name
|
|
destination.resolve().relative_to(stage.resolve())
|
|
for tracked_source in _tracked_skill_files(root, source):
|
|
relative = tracked_source.relative_to(source)
|
|
tracked_destination = destination / relative
|
|
tracked_destination.resolve().relative_to(destination.resolve())
|
|
tracked_destination.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(tracked_source, tracked_destination)
|
|
installed.append(name)
|
|
stage.rename(target)
|
|
except Exception:
|
|
shutil.rmtree(stage, ignore_errors=True)
|
|
raise
|
|
|
|
return {
|
|
"artifact": "leo_teleo_skill_pack_install",
|
|
"status": "pass",
|
|
"installed_skill_count": len(installed),
|
|
"installed_skills": installed,
|
|
"contains_secrets": False,
|
|
"production_mutation_authorized": False,
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--root", type=Path, default=Path.cwd())
|
|
parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
|
|
parser.add_argument("--target", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
result = install(args.root.resolve(), args.manifest, args.target.resolve())
|
|
except Exception as exc:
|
|
result = {"artifact": "leo_teleo_skill_pack_install", "status": "fail", "error": str(exc)}
|
|
sys.stdout.write(json.dumps(result, indent=2, sort_keys=True) + "\n")
|
|
return 1
|
|
sys.stdout.write(json.dumps(result, indent=2, sort_keys=True) + "\n")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|