fit-boilerplate-wox/evals/lint_wxo_yaml.py

#!/usr/bin/env python3
"""Linter de buenas prácticas WxO.

Falla con exit code 1 si encuentra violaciones. Usado en CI antes de
deploy. Las reglas están en docs/wxo-best-practices.md.

Reglas enforced:
- A1: máx 10 tools por agente
- A3: orchestrator no tiene tools de remediación
- A6: agente con style=react debe tener KB o tools
- T1: toda función @tool debe estar bajo @observable_tool
- T3: tools.py debe inlinear el _compat shim
- D1: docker-compose no usa `wget --spider`
- I-005: OpenAPI tiene description per-operation
"""
from __future__ import annotations

import re
import sys
import textwrap
from pathlib import Path

try:
    import yaml
except ImportError:
    print("✗ pyyaml required: pip install pyyaml", file=sys.stderr)
    sys.exit(2)


ROOT = Path(__file__).resolve().parent.parent

# Tools típicas de remediación que un orchestrator NO debe tener.
REMEDIATION_TOOLS = {
    "reset_password", "restart_service", "create_user", "grant_access",
    "assign_groups", "rotate_logs", "delete_user", "revoke_access",
    "kill_process", "redeploy", "rollback",
}

errors: list[str] = []
warnings: list[str] = []


def err(msg: str) -> None:
    errors.append(msg)
    print(f"✗ {msg}")


def warn(msg: str) -> None:
    warnings.append(msg)
    print(f"⚠  {msg}")


def ok(msg: str) -> None:
    print(f"✓ {msg}")


# ─── Agents ──────────────────────────────────────────────────────────────────

def lint_agents() -> None:
    agent_dir = ROOT / "wxo" / "agents"
    for f in sorted(agent_dir.glob("*.agent.yaml")):
        if f.name.startswith("_"):
            continue
        try:
            data = yaml.safe_load(f.read_text())
        except yaml.YAMLError as e:
            err(f"{f}: invalid YAML — {e}")
            continue

        name = data.get("name", f.stem)
        tools = data.get("tools") or []
        kb = data.get("knowledge_base") or []
        style = data.get("style", "")
        collaborators = data.get("collaborators") or []
        instructions = data.get("instructions") or ""

        # A1: máx 10 tools
        if len(tools) > 10:
            err(f"{f.name}: agente '{name}' tiene {len(tools)} tools (regla A1: máx 10)")

        # A3: orchestrator sin tools de remediación
        is_orchestrator = bool(collaborators) or "orchestrator" in name or "n1" in name
        if is_orchestrator:
            bad = [t for t in tools if t in REMEDIATION_TOOLS]
            if bad:
                err(f"{f.name}: orchestrator '{name}' declara tools de remediación: {bad} (regla A3)")

        # A6: style=react sin KB ni tools = agente sin propósito
        if style == "react" and not tools and not kb:
            err(f"{f.name}: agente '{name}' tiene style=react pero ni tools ni KB (regla A6)")

        # P1: LLM pinneado a gpt-oss-120b (warning si no)
        llm = data.get("llm", "")
        if "gpt-oss-120b" not in llm and "llama" in llm:
            warn(f"{f.name}: agente '{name}' usa llama. Issue I-002 — preferible gpt-oss-120b.")

        # P2: REGLA #0 presente en instructions
        if "REGLA #0" not in instructions and "NUNCA escribas la tool call" not in instructions:
            warn(f"{f.name}: agente '{name}' no menciona REGLA #0 en instructions (regla P2)")

        # ok message
        if len(errors) == 0:
            ok(f"agent {name} (tools={len(tools)}, kb={len(kb)}, style={style})")


# ─── Tools (Python) ──────────────────────────────────────────────────────────

def lint_python_tools() -> None:
    tools_dir = ROOT / "wxo" / "tools" / "python"
    for f in sorted(tools_dir.glob("*.py")):
        if f.name.startswith("_"):
            continue  # skip helpers
        src = f.read_text()

        # T3: compat shim presente
        if "inline compat shim" not in src and "from ibm_watsonx_orchestrate.agent_builder.tools import tool" not in src:
            err(f"{f.name}: falta compat shim inline (regla T3, issue I-010)")

        # T1: @tool sin @observable_tool
        # Buscar @tool( no precedido por observable_tool
        for line_no, line in enumerate(src.splitlines(), start=1):
            line_stripped = line.strip()
            if re.match(r"^@tool\b", line_stripped):
                # ¿el archivo lo usa solo como import alias _adk_tool? OK.
                # Si no, es violación.
                err(f"{f.name}:{line_no}: usa @tool directo, debe ser @observable_tool (regla T1)")

        ok(f"tools {f.name}")


# ─── OpenAPI ─────────────────────────────────────────────────────────────────

def lint_openapi() -> None:
    openapi_dir = ROOT / "wxo" / "tools" / "openapi"
    for f in sorted(list(openapi_dir.glob("*.yaml")) + list(openapi_dir.glob("*.json"))):
        if f.name.startswith("_"):
            continue
        try:
            data = yaml.safe_load(f.read_text())
        except yaml.YAMLError as e:
            err(f"{f.name}: invalid YAML/JSON — {e}")
            continue

        paths = data.get("paths", {})
        for path, item in paths.items():
            for method, op in item.items():
                if not isinstance(op, dict) or "responses" not in op:
                    continue
                # I-005: description per-op
                if not op.get("description"):
                    err(f"{f.name}: {method.upper()} {path} sin description (issue I-005)")
                # I-006: security per-op
                if op.get("security") is None and data.get("security") is not None:
                    warn(f"{f.name}: {method.upper()} {path} no declara security per-op (issue I-006)")

        ok(f"openapi {f.name}")


# ─── docker-compose ──────────────────────────────────────────────────────────

def lint_compose() -> None:
    for f in [ROOT / "docker-compose.yml", ROOT / "docker-compose.local.yml"]:
        if not f.exists():
            continue
        src = f.read_text()
        # D1: no `wget --spider`
        if "wget --spider" in src or "wget -S --spider" in src:
            err(f"{f.name}: usa `wget --spider`, cambiá por `wget -qO-` o `curl -sf` (issue I-008)")
        # I-009: labels Traefik manuales
        if re.search(r"traefik\.http\.routers\.", src):
            warn(f"{f.name}: declara labels Traefik manuales — Coolify v4 los autogenera (issue I-009)")
        ok(f"compose {f.name}")


# ─── Main ────────────────────────────────────────────────────────────────────

def main() -> int:
    print("─── Linting agents ───")
    lint_agents()
    print("\n─── Linting Python tools ───")
    lint_python_tools()
    print("\n─── Linting OpenAPI ───")
    lint_openapi()
    print("\n─── Linting docker-compose ───")
    lint_compose()

    print("\n" + "─" * 60)
    if errors:
        print(f"✗ {len(errors)} ERROR(es), {len(warnings)} warning(s)")
        return 1
    print(f"✓ Linting OK ({len(warnings)} warning(s))")
    return 0


if __name__ == "__main__":
    sys.exit(main())