"""A0 Software Orchestrator – plugin lifecycle hooks. Responsibilities: - create/upgrade the local PatternDB using the plugin migration layer - create conservative backups before updates/deactivation - keep plugin activation deterministic for the orchestrator profiles - avoid mutating the Agent Zero runtime beyond scoped plugin toggle files """ from __future__ import annotations import os import shutil from pathlib import Path from datetime import datetime PLUGIN_DIR = Path(__file__).parent LIBRARY_DIR = PLUGIN_DIR / "helpers" / "library" DB_PATH = LIBRARY_DIR / "patterns.db" BACKUP_DIR = LIBRARY_DIR / "backups" A0_ROOT = Path("/a0") PLUGIN_NAME = "a0_software_orchestrator" ORCH_PROFILE_NAMES = [ "a0_software_orchestrator", "quality_reviewer", "requirements_analyst", "solution_architect", "implementation_engineer", "test_debug_engineer", "runtime_devops_engineer", "security_data_engineer", "release_auditor", "codebase_explorer", "deploy_agent", "ui_design_specialist", ] def install(): """Initialize/upgrade the plugin database when the plugin is enabled.""" os.makedirs(LIBRARY_DIR, exist_ok=True) os.makedirs(BACKUP_DIR, exist_ok=True) _ensure_activation_toggles() try: if DB_PATH.exists(): _backup_db() from usr.plugins.a0_software_orchestrator.helpers.library.db import PatternDB db = PatternDB(DB_PATH) db.conn.execute("SELECT 1").fetchone() print(f"[a0_software_orchestrator] PatternDB ready: {DB_PATH}") _check_agents() except Exception as e: print(f"[a0_software_orchestrator] DB initialization failed: {e}") raise def cache(): """Warm up optional capabilities without installing packages or changing runtime.""" try: from usr.plugins.a0_software_orchestrator.helpers.library.db import PatternDB db = PatternDB(DB_PATH) count = db.conn.execute("SELECT COUNT(*) FROM patterns").fetchone()[0] print(f"[a0_software_orchestrator] PatternDB warm: {count} patterns") except Exception as e: print(f"[a0_software_orchestrator] PatternDB warm-up skipped: {e}") def deactivate(): """Create a final backup when the plugin is disabled.""" if DB_PATH.exists(): _backup_db() print("[a0_software_orchestrator] Backup created on deactivate") def pre_update(): """Create a backup before plugin updates.""" if DB_PATH.exists(): _backup_db() def _ensure_activation_toggles(): """Ensure only the plugin root ON marker exists. Agent profiles are shipped inside this plugin under agents// and are discovered by Agent Zero from enabled plugin paths. Do not create or mutate /a0/usr/agents scoped override folders from this hook; scoped activation belongs to Agent Zero's plugin config UI/API. """ try: (PLUGIN_DIR / ".toggle-0").unlink(missing_ok=True) (PLUGIN_DIR / ".toggle-1").touch(exist_ok=True) except Exception as e: print(f"[a0_software_orchestrator] Root toggle update skipped: {e}") def _backup_db(): """Create a timestamped DB backup, keeping the latest 10.""" if not DB_PATH.exists(): return os.makedirs(BACKUP_DIR, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_path = BACKUP_DIR / f"patterns_backup_{timestamp}.db" try: shutil.copy2(str(DB_PATH), str(backup_path)) backups = sorted(BACKUP_DIR.glob("patterns_backup_*.db")) for old in backups[:-10]: old.unlink() print(f"[a0_software_orchestrator] Backup created: {backup_path}") except Exception as e: print(f"[a0_software_orchestrator] Backup failed: {e}") def _check_agents(): """Check required Agent Zero specialist profiles shipped with the plugin.""" agents_dir = PLUGIN_DIR / "agents" missing = [agent for agent in ORCH_PROFILE_NAMES if not (agents_dir / agent / "agent.yaml").exists()] if missing: print(f"[a0_software_orchestrator] Missing agent profiles: {', '.join(missing)}")