#!/usr/bin/env python3 """Validate the repo-native Leo/Teleo skill pack and its local routes.""" from __future__ import annotations import argparse import hashlib import json import os import re import shlex import subprocess import sys from pathlib import Path DEFAULT_MANIFEST = Path("docs/reports/leo-working-state-20260709/skill-pack-manifest.json") REPO_PREFIXES = ( ".agents/", ".github/", "deploy/", "docs/", "fixtures/", "hermes-agent/", "lib/", "ops/", "outputs/", "schemas/", "scripts/", "systemd/", "tests/", ) PATH_SUFFIXES = (".json", ".jsonl", ".md", ".py", ".sh", ".sql", ".yaml", ".yml") CODE_SPAN = re.compile(r"`([^`\n]+)`") MARKDOWN_LINK = re.compile(r"\[[^\]]+\]\(([^)]+)\)") FENCED_SHELL = re.compile(r"```(?:bash|sh|shell)\n(.*?)```", re.DOTALL) def _frontmatter(text: str) -> dict[str, str]: if not text.startswith("---\n"): return {} try: block = text.split("---\n", 2)[1] except IndexError: return {} values: dict[str, str] = {} for line in block.splitlines(): if ":" not in line: continue key, value = line.split(":", 1) values[key.strip()] = value.strip().strip('"') return values def _tracked_paths(root: Path) -> set[Path]: result = subprocess.run( ["git", "-C", str(root), "ls-files", "-z"], check=False, capture_output=True, ) if result.returncode != 0: raise RuntimeError(result.stderr.decode("utf-8", errors="replace")) return {Path(item.decode("utf-8")) for item in result.stdout.split(b"\0") if item} def _is_tracked(root: Path, path: Path, tracked: set[Path]) -> bool: relative = path.resolve().relative_to(root.resolve()) if path.is_file(): return relative in tracked prefix = f"{relative.as_posix().rstrip('/')}/" return any(item.as_posix().startswith(prefix) for item in tracked) def _clean_token(token: str) -> str: token = token.strip().strip("'\"") token = token.rstrip(",;.)]") token = re.sub(r":\d+$", "", token) return token def _candidate_tokens(span: str) -> list[str]: try: tokens = shlex.split(span) except ValueError: tokens = span.split() return [_clean_token(token) for token in tokens] def _resolve_candidate(root: Path, source: Path, token: str) -> Path | None: if not token or token in {".md", ".json", ".py", ".sh", ".sql"}: return None if any(marker in token for marker in ("*", "<", ">", "...", "${", "$LEDGER")): return None if token.startswith(("http://", "https://", "/", "~/")): return None if token.startswith(REPO_PREFIXES) or token in {"README.md", "CODEOWNERS"}: candidate = root / token elif token.endswith(PATH_SUFFIXES) and "/" in token: candidate = source.parent / token else: return None try: candidate.resolve().relative_to(root.resolve()) except ValueError: return None return candidate def _inline_paths(root: Path, source: Path) -> set[Path]: text = source.read_text(encoding="utf-8") candidates: set[Path] = set() spans = CODE_SPAN.findall(text) + MARKDOWN_LINK.findall(text) for block in FENCED_SHELL.findall(text): spans.extend(line for line in block.splitlines() if line.strip()) for span in spans: for token in _candidate_tokens(span): candidate = _resolve_candidate(root, source, token) if candidate is not None: candidates.add(candidate) return candidates def validate(root: Path, manifest_path: Path) -> dict[str, object]: problems: list[dict[str, str]] = [] checked: set[Path] = set() manifest_file = root / manifest_path if not manifest_file.is_file(): return { "artifact": "leo_teleo_skill_pack_path_validation", "status": "fail", "problems": [{"kind": "missing_manifest", "path": str(manifest_path)}], } manifest = json.loads(manifest_file.read_text(encoding="utf-8")) tracked = _tracked_paths(root) if not _is_tracked(root, manifest_file, tracked): problems.append({"kind": "untracked_manifest", "path": str(manifest_path)}) skills = manifest.get("skills", []) skill_names = {entry.get("name") for entry in skills} for entry in skills: relative = Path(entry["path"]) path = root / relative checked.add(path) if not path.is_file(): problems.append({"kind": "missing_skill", "path": str(relative)}) continue if not _is_tracked(root, path, tracked): problems.append({"kind": "untracked_skill", "path": str(relative)}) metadata = _frontmatter(path.read_text(encoding="utf-8")) if metadata.get("name") != entry.get("name"): problems.append({"kind": "skill_name_mismatch", "path": str(relative)}) if not metadata.get("description"): problems.append({"kind": "missing_skill_description", "path": str(relative)}) if set(metadata) != {"name", "description"}: problems.append({"kind": "invalid_skill_frontmatter_keys", "path": str(relative)}) if not re.fullmatch(r"[a-z0-9-]{1,64}", metadata.get("name", "")): problems.append({"kind": "invalid_skill_name", "path": str(relative)}) if relative.parent.name != metadata.get("name"): problems.append({"kind": "skill_folder_name_mismatch", "path": str(relative)}) required_coverage = manifest.get("required_coverage", {}) for area, owners in required_coverage.items(): for owner in owners: if owner not in skill_names: problems.append({"kind": "unknown_coverage_owner", "path": f"{area}:{owner}"}) reference_paths = list(manifest.get("reference_files", [])) command_paths = list(manifest.get("command_files", [])) scan_paths = list(manifest.get("scan_files", [])) for relative_text in reference_paths + command_paths + scan_paths: relative = Path(relative_text) path = root / relative checked.add(path) if not path.is_file(): problems.append({"kind": "missing_manifest_path", "path": str(relative)}) elif not _is_tracked(root, path, tracked): problems.append({"kind": "untracked_manifest_path", "path": str(relative)}) for relative_text in manifest.get("executable_files", []): relative = Path(relative_text) path = root / relative checked.add(path) if not path.is_file(): problems.append({"kind": "missing_executable", "path": str(relative)}) elif not os.access(path, os.X_OK): problems.append({"kind": "not_executable", "path": str(relative)}) scan_sources = [root / entry["path"] for entry in skills] scan_sources += [root / relative for relative in manifest.get("scan_files", [])] for source in scan_sources: if not source.is_file(): continue for candidate in _inline_paths(root, source): checked.add(candidate) if not candidate.exists(): problems.append( { "kind": "missing_inline_path", "path": str(candidate.resolve().relative_to(root.resolve())), "source": str(source.relative_to(root)), } ) elif not _is_tracked(root, candidate, tracked): problems.append( { "kind": "untracked_inline_path", "path": str(candidate.resolve().relative_to(root.resolve())), "source": str(source.relative_to(root)), } ) relative_paths = sorted(str(path.resolve().relative_to(root.resolve())) for path in checked) digest = hashlib.sha256("\n".join(relative_paths).encode("utf-8")).hexdigest() return { "artifact": "leo_teleo_skill_pack_path_validation", "required_tier": "T2_runtime", "status": "pass" if not problems else "fail", "manifest": str(manifest_path), "skill_count": len(skills), "coverage_area_count": len(required_coverage), "checked_path_count": len(relative_paths), "checked_paths_sha256": digest, "contains_secrets": False, "production_mutation_authorized": False, "problems": problems, } 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("--output", type=Path) args = parser.parse_args() root = args.root.resolve() result = validate(root, args.manifest) payload = json.dumps(result, indent=2, sort_keys=True) + "\n" if args.output: output = args.output if args.output.is_absolute() else root / args.output output.parent.mkdir(parents=True, exist_ok=True) output.write_text(payload, encoding="utf-8") sys.stdout.write(payload) return 0 if result["status"] == "pass" else 1 if __name__ == "__main__": raise SystemExit(main())