a0_software_orchestrator/execute.py

#!/usr/bin/env python3
"""Manual maintenance/self-test script for A0 Software Orchestrator plugin.

This script is intentionally standalone. It validates package structure and source
contracts only. Runtime registration must be tested inside Agent Zero after install
by calling the registered `orchestrator_self_check` tool.
"""
from __future__ import annotations

import ast
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
REQUIRED = [
    "plugin.yaml",
    "README.md",
    "tools",
    "prompts",
    "helpers",
    "agents",
]
TOOL_FILES = sorted((ROOT / "tools").glob("*.py")) if (ROOT / "tools").exists() else []


def fail(msg: str) -> int:
    print(f"[FAIL] {msg}")
    return 1


def ok(msg: str):
    print(f"[OK] {msg}")


def check_structure() -> int:
    rc = 0
    for item in REQUIRED:
        if not (ROOT / item).exists():
            print(f"[FAIL] missing {item}")
            rc = 1
        else:
            ok(f"exists {item}")
    if (ROOT / ".toggle-0").exists():
        print("[FAIL] .toggle-0 present")
        rc = 1
    if not (ROOT / ".toggle-1").exists():
        print("[FAIL] .toggle-1 missing")
        rc = 1
    else:
        ok(".toggle-1 present")
    return rc


def check_python_compile() -> int:
    rc = 0
    for py in ROOT.rglob("*.py"):
        if "__pycache__" in py.parts:
            continue
        try:
            ast.parse(py.read_text(encoding="utf-8"), filename=str(py))
        except SyntaxError as e:
            print(f"[FAIL] syntax {py.relative_to(ROOT)}: {e}")
            rc = 1
    if rc == 0:
        ok("python syntax")
    return rc


def check_tools() -> int:
    rc = 0
    if not TOOL_FILES:
        return fail("no tools/*.py files")
    for py in TOOL_FILES:
        if py.name == "__init__.py":
            continue
        name = py.stem
        prompt = ROOT / "prompts" / f"agent.system.tool.{name}.md"
        if not prompt.exists():
            print(f"[FAIL] missing tool prompt for {name}")
            rc = 1
        text = py.read_text(encoding="utf-8")
        tree = ast.parse(text)
        has_tool_import = "from helpers.tool import Tool" in text
        has_response_import = "Response" in text
        has_tool_subclass = any(isinstance(n, ast.ClassDef) and any(getattr(b, 'id', '') == 'Tool' or getattr(b, 'attr', '') == 'Tool' for b in n.bases) for n in tree.body)
        if not (has_tool_import and has_response_import and has_tool_subclass):
            print(f"[FAIL] tool contract {py.relative_to(ROOT)}")
            rc = 1
    if rc == 0:
        ok("tool files + prompt docs")
    return rc


def check_api_handlers() -> int:
    """Validate plugin API handlers use the Agent Zero framework ApiHandler directly."""
    rc = 0
    api_dir = ROOT / "api"
    if not api_dir.exists():
        return 0
    for py in api_dir.glob("*.py"):
        text = py.read_text(encoding="utf-8")
        tree = ast.parse(text, filename=str(py))
        if "from helpers.api import ApiHandler" not in text:
            print(f"[FAIL] api contract {py.relative_to(ROOT)}: must import from helpers.api")
            rc = 1
        has_handler = any(
            isinstance(n, ast.ClassDef) and any(
                (getattr(b, "id", "") == "ApiHandler") or (getattr(b, "attr", "") == "ApiHandler")
                for b in n.bases
            )
            for n in tree.body
        )
        if not has_handler:
            print(f"[FAIL] api contract {py.relative_to(ROOT)}: no ApiHandler subclass")
            rc = 1
    if (ROOT / "helpers" / "api.py").exists():
        print("[FAIL] helpers/api.py present; API handlers must use framework helpers.api directly")
        rc = 1
    if rc == 0:
        ok("api handler contracts")
    return rc


def check_extensions() -> int:
    """Validate plugin extensions use the Agent Zero framework Extension directly."""
    rc = 0
    ext_dir = ROOT / "extensions"
    if not ext_dir.exists():
        return 0
    for py in ext_dir.rglob("*.py"):
        text = py.read_text(encoding="utf-8")
        tree = ast.parse(text, filename=str(py))
        has_extension_class = any(
            isinstance(n, ast.ClassDef) and any(
                (getattr(b, "id", "") == "Extension") or (getattr(b, "attr", "") == "Extension")
                for b in n.bases
            )
            for n in tree.body
        )
        if has_extension_class and "from helpers.extension import Extension" not in text:
            print(f"[FAIL] extension contract {py.relative_to(ROOT)}: must import from helpers.extension")
            rc = 1
    if rc == 0:
        ok("extension contracts")
    return rc


def check_forbidden() -> int:
    rc = 0
    forbidden = ["sys.path" + ".insert", "orch" + "_helpers", "plugin" + "_tools"]
    for p in ROOT.rglob("*"):
        if not p.is_file() or p.suffix.lower() not in {".py", ".md", ".yaml", ".yml", ".html", ".js", ".json"}:
            continue
        text = p.read_text(encoding="utf-8", errors="ignore")
        for token in forbidden:
            if token in text:
                print(f"[FAIL] forbidden token {token!r} in {p.relative_to(ROOT)}")
                rc = 1
    if rc == 0:
        ok("no forbidden legacy tokens")
    return rc




def check_packaging_hygiene() -> int:
    rc = 0
    bad = []
    for p in ROOT.rglob("*"):
        rel = p.relative_to(ROOT)
        if "__pycache__" in p.parts or p.suffix in {".pyc", ".pyo"} or p.suffix == ".db" or p.name.startswith("patterns_backup_"):
            bad.append(str(rel))
    if bad:
        print("[FAIL] runtime/compiled artifacts present:")
        for item in bad[:50]:
            print(f"  - {item}")
        rc = 1
    else:
        ok("no runtime/compiled artifacts")
    return rc

def check_security_lint() -> int:
    """Run the plugin security prompt lint as part of the package self-test."""
    lint = ROOT / "scripts" / "security_prompt_lint.py"
    if not lint.exists():
        print("[FAIL] missing scripts/security_prompt_lint.py")
        return 1
    import subprocess
    proc = subprocess.run([sys.executable, str(lint)], cwd=str(ROOT), text=True, capture_output=True)
    if proc.returncode != 0:
        out = (proc.stdout + proc.stderr).strip()
        print("[FAIL] security prompt lint")
        if out:
            print(out)
        return 1
    ok("security prompt lint")
    return 0


def main() -> int:
    rc = 0
    rc |= check_structure()
    rc |= check_python_compile()
    rc |= check_tools()
    rc |= check_api_handlers()
    rc |= check_extensions()
    rc |= check_forbidden()
    rc |= check_packaging_hygiene()
    rc |= check_security_lint()
    if rc == 0:
        print("PASS")
    else:
        print("FAIL")
    return rc


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