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