Software Orchestrator
120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
"""Self-test runner for the orchestrator plugin."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from typing import List, Tuple
|
|
|
|
PLUGIN_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
|
|
EXPECTED_AGENTS = [
|
|
"a0_software_orchestrator",
|
|
"codebase_explorer",
|
|
"requirements_analyst",
|
|
"solution_architect",
|
|
"implementation_engineer",
|
|
"test_debug_engineer",
|
|
"runtime_devops_engineer",
|
|
"security_data_engineer",
|
|
"release_auditor",
|
|
"quality_reviewer",
|
|
"deploy_agent",
|
|
]
|
|
|
|
EXPECTED_TOOLS = [
|
|
"artifact_guard.py",
|
|
"block_compactor.py",
|
|
"block_resume.py",
|
|
"capability_check.py",
|
|
"context_compactor.py",
|
|
"handover_to_specialist.py",
|
|
"next_step.py",
|
|
"orchestrator_state.py",
|
|
"plan_mode_guard.py",
|
|
"project_registry.py",
|
|
"quality_gate.py",
|
|
"read_briefing.py",
|
|
"repo_manifest.py",
|
|
"resume_checker.py",
|
|
"scorecard_update.py",
|
|
"tool_registry.py",
|
|
"orchestrator_self_check.py",
|
|
]
|
|
|
|
|
|
def check_structure() -> List[Tuple[str, bool, str]]:
|
|
results: List[Tuple[str, bool, str]] = []
|
|
results.append(("plugin.yaml exists", os.path.exists(os.path.join(PLUGIN_DIR, "plugin.yaml")), ""))
|
|
results.append((".toggle-1 exists", os.path.exists(os.path.join(PLUGIN_DIR, ".toggle-1")), ""))
|
|
results.append((".toggle-0 absent", not os.path.exists(os.path.join(PLUGIN_DIR, ".toggle-0")), ""))
|
|
|
|
expected_top = {
|
|
"plugin.yaml", "execute.py", "hooks.py", "default_config.yaml", "config.json",
|
|
"README.md", "LICENSE", "__init__.py", "agents", "api", "tools",
|
|
"helpers", "prompts", "extensions", "webui", "help", "scripts", "utils", "docs",
|
|
}
|
|
for item in os.listdir(PLUGIN_DIR):
|
|
if item.startswith("."):
|
|
continue
|
|
if item not in expected_top:
|
|
results.append((f"unexpected top-level: {item}", False, ""))
|
|
|
|
# Packaging hygiene: release zips must not contain runtime/compiled artifacts.
|
|
bad_artifacts = []
|
|
for dirpath, dirnames, filenames in os.walk(PLUGIN_DIR):
|
|
rel_dir = os.path.relpath(dirpath, PLUGIN_DIR)
|
|
if "__pycache__" in dirpath.split(os.sep):
|
|
bad_artifacts.append(rel_dir)
|
|
for fname in filenames:
|
|
if fname.endswith(".pyc") or fname.endswith(".pyo") or fname.startswith("patterns_backup_"):
|
|
# patterns.db is a legitimate plugin database, not a runtime artifact
|
|
if not (fname == "patterns.db" and rel_dir == "helpers/library"):
|
|
bad_artifacts.append(os.path.join(rel_dir, fname))
|
|
results.append(("no runtime/compiled artifacts", len(bad_artifacts) == 0, ", ".join(bad_artifacts[:10])))
|
|
|
|
agents_dir = os.path.join(PLUGIN_DIR, "agents")
|
|
for agent in EXPECTED_AGENTS:
|
|
agent_dir = os.path.join(agents_dir, agent)
|
|
results.append((f"agent {agent}: agent.yaml", os.path.exists(os.path.join(agent_dir, "agent.yaml")), ""))
|
|
results.append((f"agent {agent}: specifics.md", os.path.exists(os.path.join(agent_dir, "prompts", "agent.system.main.specifics.md")), ""))
|
|
results.append((f"agent {agent}: quality_contract.md", os.path.exists(os.path.join(agent_dir, "quality_contract.md")), ""))
|
|
|
|
tools_dir = os.path.join(PLUGIN_DIR, "tools")
|
|
for tool in EXPECTED_TOOLS:
|
|
results.append((f"tool {tool}", os.path.exists(os.path.join(tools_dir, tool)), ""))
|
|
|
|
ext_dir = os.path.join(PLUGIN_DIR, "extensions", "python")
|
|
for ext in [
|
|
"system_prompt/inject_orchestrator_rules.py",
|
|
"system_prompt/_30_inject_profile_specifics.py",
|
|
"monologue_start/load_project_state.py",
|
|
"message_loop_prompts_after/inject_context_budget.py",
|
|
"tool_execute_before/guard_risky_actions.py",
|
|
"monologue_end/remind_state_update.py",
|
|
]:
|
|
results.append((f"extension {ext}", os.path.exists(os.path.join(ext_dir, ext)), ""))
|
|
results.append(("fake tool-list injection extension absent", not os.path.exists(os.path.join(ext_dir, "system_prompt", "_20_inject_tool_list.py")), ""))
|
|
|
|
return results
|
|
|
|
|
|
def run_self_test() -> int:
|
|
print("=== A0 Software Orchestrator Self-Test ===\n")
|
|
results = check_structure()
|
|
passed = 0
|
|
failed = 0
|
|
for name, ok, msg in results:
|
|
if ok:
|
|
passed += 1
|
|
else:
|
|
failed += 1
|
|
print(f"[{'PASS' if ok else 'FAIL'}] {name}{f' ({msg})' if msg else ''}")
|
|
print("\n---")
|
|
print(f"Total: {passed} passed, {failed} failed")
|
|
print("Result: PASS" if failed == 0 else "Result: FAIL")
|
|
return 0 if failed == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(run_self_test())
|