dev-workflows/adapters/omp/build.py

"""Build the native OMP package from the canonical dev-workflows checkout."""
from __future__ import annotations

import hashlib
import json
import os
import re
import shutil
import tempfile
from pathlib import Path


OWNER_MARKER = ".dev-workflows-omp-package"
def _owner_value(src: Path) -> str:
    return f"dev-workflows\nsource_root={src}\n"


def _content_digest(package: Path) -> str:
    h = hashlib.sha256()
    for path in sorted(package.rglob("*")):
        if path.is_symlink():
            raise FileExistsError(f"symlink em pacote gerado: {path}")
        if not path.is_file() or path.name == OWNER_MARKER:
            continue
        h.update(path.relative_to(package).as_posix().encode("utf-8"))
        h.update(b"\0")
        h.update(str(path.stat().st_mode & 0o777).encode("ascii"))
        with path.open("rb") as src:
            for chunk in iter(lambda: src.read(1024 * 1024), b""):
                h.update(chunk)
    return h.hexdigest()

AGENT_NAMES = (
    "explorer", "implementer", "planner", "reviewer",
    "security-auditor", "ui-critic", "verifier",
)
MODEL_ID = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+(?::(?:none|minimal|low|medium|high|xhigh|max))?$")
TOOL_NAMES = {
    "Read": "read", "Edit": "edit", "Write": "write", "Bash": "bash",
    "Grep": "grep", "Glob": "glob", "WebFetch": "web_search",
}


def _agent(src: Path, name: str, models: dict[str, str], package_root: Path) -> str:
    text = (src / "agents" / f"{name}.md").read_text(encoding="utf-8")
    match = re.match(r"\A---\s*\n(.*?)\n---\s*\n(.*)\Z", text, re.S)
    if not match:
        raise ValueError(f"frontmatter inválido em agents/{name}.md")
    raw, prompt = match.groups()
    meta: dict[str, str] = {}
    for line in raw.splitlines():
        key, sep, value = line.partition(":")
        if sep:
            meta[key.strip()] = value.strip()
    if meta.get("name") != name or not meta.get("description"):
        raise ValueError(f"metadados incompletos em agents/{name}.md")

    source_tools = [item.strip() for item in meta.get("tools", "").split(",") if item.strip()]
    # An explicit OMP allowlist prevents read-only agents from inheriting write tools.
    tools = list(dict.fromkeys(TOOL_NAMES[item] for item in source_tools if item in TOOL_NAMES))
    if not tools:
        raise ValueError(f"nenhuma ferramenta OMP reconhecida em agents/{name}.md")
    lines = ["---", f"name: {name}", f"description: {meta['description']}", "tools: " + ", ".join(tools)]
    skills = re.search(r"^skills:\s*\[(.*?)\]\s*$", raw, re.M)
    if skills:
        values = [value.strip().strip("'\"") for value in skills.group(1).split(",") if value.strip()]
        if values:
            lines.append("autoloadSkills:")
            lines.extend(f"  - {value}" for value in values)
    model = models.get(name)
    if model is not None:
        if not isinstance(model, str) or not MODEL_ID.fullmatch(model):
            raise ValueError(f"ID de modelo inválido para {name}: {model!r}")
        lines.append(f"model: {model}")
    body = prompt.replace("${CLAUDE_PLUGIN_ROOT}", str(package_root))
    return "\n".join(lines + ["---", "", body])


def build_package(src_root, dest, models=None):
    """Atomically build an owned OMP plugin package at *dest*.

    ``models`` maps canonical agent names (roles) to verified concrete
    ``provider/model[:thinking]`` selectors. Missing roles intentionally omit
    ``model`` so OMP inherits the parent model.
    """
    src = Path(src_root).resolve()
    target = Path(dest).expanduser().absolute()
    roles = {} if models is None else dict(models)
    unknown = set(roles) - set(AGENT_NAMES)
    if unknown:
        raise ValueError("papéis desconhecidos: " + ", ".join(sorted(unknown)))
    if not (src / "agents").is_dir() or not (src / "skills").is_dir():
        raise ValueError(f"checkout canônico inválido: {src}")
    if target.is_symlink():
        raise FileExistsError(f"destino é symlink; recusando substituir: {target}")
    if target.exists():
        marker = target / OWNER_MARKER
        if not target.is_dir() or not marker.is_file():
            raise FileExistsError(f"destino existente não pertence a dev-workflows: {target}")
        if marker.read_text(encoding="utf-8") != _owner_value(src) + f"sha256={_content_digest(target)}\n":
            raise FileExistsError(f"pacote existente tem alterações manuais ou origem diferente: {target}")

    target.parent.mkdir(parents=True, exist_ok=True)
    stage = Path(tempfile.mkdtemp(prefix=f".{target.name}.stage-", dir=target.parent))
    backup = None
    try:
        skills_src = src / "skills"
        skills_dest = stage / "skills"
        shutil.copytree(skills_src, skills_dest, symlinks=False,
                        ignore=shutil.ignore_patterns(".git", "__pycache__", "*.pyc"))
        skill_count = sum(1 for path in skills_dest.glob("*/SKILL.md") if path.is_file())
        for path in skills_dest.rglob("*"):
            if not path.is_file():
                continue
            try:
                content = path.read_text(encoding="utf-8")
            except UnicodeError:
                continue
            if "${CLAUDE_PLUGIN_ROOT}" in content:
                path.write_text(content.replace("${CLAUDE_PLUGIN_ROOT}", str(target)), encoding="utf-8")
        if skill_count != 21:
            raise ValueError(f"esperadas 21 skills canônicas; encontradas {skill_count}")

        agents_dest = stage / "agents"
        agents_dest.mkdir()
        for name in AGENT_NAMES:
            if not (src / "agents" / f"{name}.md").is_file():
                raise ValueError(f"agente canônico ausente: {name}")
            (agents_dest / f"{name}.md").write_text(_agent(src, name, roles, target), encoding="utf-8")

        # The quality-gate executable is a package asset; Claude hooks are deliberately
        # excluded. The loop/controller remains responsible for enforcing its verdict.
        gate = src / "scripts" / "quality-gate"
        if gate.is_file():
            (stage / "scripts").mkdir()
            shutil.copy2(gate, stage / "scripts" / "quality-gate")
        manifest = {
            "name": "dev-workflows",
            "version": "1.0.0",
            "description": "Skills e agentes nativos para oh-my-pi; gate imposto pelo controlador externo.",
            "omp": {"extensions": []},
        }
        (stage / "package.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
        (stage / OWNER_MARKER).write_text(_owner_value(src) + f"sha256={_content_digest(stage)}\n", encoding="utf-8")

        if target.exists():
            backup = Path(tempfile.mkdtemp(prefix=f".{target.name}.backup-", dir=target.parent))
            backup.rmdir()
            os.replace(target, backup)
        try:
            os.replace(stage, target)
        except BaseException:
            if backup is not None:
                os.replace(backup, target)
                backup = None
            raise
        if backup is not None:
            shutil.rmtree(backup)
            backup = None
        return target
    finally:
        if stage.exists():
            shutil.rmtree(stage)
        if backup is not None and backup.exists():
            if not target.exists():
                os.replace(backup, target)
            else:
                shutil.rmtree(backup)