teleo-infrastructure/.agents/skills/teleo-leo-onboarding/scripts/validate_skill_pack.py

#!/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())