aios/agents/scripts/autonomous_controller.py.v1

#!/usr/bin/env python3
"""
AIOS Autonomous Controller

Detect -> Diagnose -> Fix -> Test -> Patch -> Approve -> Commit -> Notify

Runs alongside the command watcher. Checks system health periodically.
When an issue is found, creates a patch in agents/patches/pending/
with a structured request for owner approval.
"""

import json
import os
import hashlib
import time
import sys
import logging
import subprocess
import requests
from pathlib import Path
from datetime import datetime, timezone, timedelta

WORKSPACE = Path(os.environ.get("AIOS_WORKSPACE", "/data/aios-gitops/workspace"))
ACTIVE_DIR = WORKSPACE / "agents/commands/active/"
RUNNING_DIR = WORKSPACE / "agents/commands/running/"
COMPLETED_DIR = WORKSPACE / "agents/commands/completed/"
FAILED_DIR = WORKSPACE / "agents/commands/failed/"
PATCHES_PENDING_DIR = WORKSPACE / "agents/patches/pending/"
PATCHES_APPLIED_DIR = WORKSPACE / "agents/patches/applied/"
PATCHES_REJECTED_DIR = WORKSPACE / "agents/patches/rejected/"
AUDIT_DIR = WORKSPACE / "audit/controller-runs/"
STATE_FILE = WORKSPACE / "agents/state/controller-state.json"
HEALTH_FILE = WORKSPACE / "agents/state/health.json"
POLL_INTERVAL = int(os.environ.get("CONTROLLER_POLL_INTERVAL_SECONDS", "300"))
GITEA_URL = os.environ.get("GITEA_URL", "http://127.0.0.1:3000")
API_URL = os.environ.get("AIOS_API_URL", "http://127.0.0.1:18890")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [controller] %(levelname)s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
log = logging.getLogger("autonomous-controller")

CONTROLLER_VERSION = "1.0.0"


def load_state():
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text())
        except (json.JSONDecodeError, OSError):
            pass
    return {"controller_version": CONTROLLER_VERSION, "checks": {}, "patches": {}}


def save_state(state):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))


def write_audit_record(check_name, status, details):
    record = {
        "controller_run_id": f"controller-{int(time.time())}",
        "check": check_name,
        "status": status,
        "controller_version": CONTROLLER_VERSION,
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "details": details,
    }
    AUDIT_DIR.mkdir(parents=True, exist_ok=True)
    af = AUDIT_DIR / f"{record['controller_run_id']}.json"
    af.write_text(json.dumps(record, indent=2, ensure_ascii=False))
    return record


def create_patch(check_name, diagnosis, proposed_fix, files_to_change):
    """Create a structured patch request for owner approval."""
    patch_id = f"patch-{int(time.time())}"
    patch = {
        "patch_id": patch_id,
        "check": check_name,
        "diagnosis": diagnosis,
        "proposed_fix": proposed_fix,
        "files_to_change": files_to_change,
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "status": "pending",
        "approved": False,
        "applied": False,
    }
    PATCHES_PENDING_DIR.mkdir(parents=True, exist_ok=True)
    pf = PATCHES_PENDING_DIR / f"{patch_id}.json"
    pf.write_text(json.dumps(patch, indent=2, ensure_ascii=False))
    log.info(f"Created patch: {patch_id}")
    return patch


def get_gitea_version():
    try:
        r = requests.get(f"{GITEA_URL}/api/v1/version", timeout=5)
        if r.status_code == 200:
            return {"healthy": True, "data": r.json()}
        return {"healthy": False, "error": f"HTTP {r.status_code}"}
    except Exception as e:
        return {"healthy": False, "error": str(e)}


def get_gitea_repo_info(owner="aios", repo="aios-v23"):
    try:
        branches = []
        page = 1
        while True:
            r = requests.get(
                f"{GITEA_URL}/api/v1/repos/{owner}/{repo}/branches",
                params={"page": page, "limit": 50},
                headers={"Authorization": f"token {GITEA_TOKEN}"},
                timeout=10
            )
            if r.status_code != 200:
                break
            data = r.json()
            if not data:
                break
            branches.extend(data)
            page += 1
        tags = []
        page = 1
        while True:
            r = requests.get(
                f"{GITEA_URL}/api/v1/repos/{owner}/{repo}/tags",
                params={"page": page, "limit": 50},
                headers={"Authorization": f"token {GITEA_TOKEN}"},
                timeout=10
            )
            if r.status_code != 200:
                break
            data = r.json()
            if not data:
                break
            tags.extend(data)
            page += 1
        return {
            "healthy": True,
            "branches": len(branches),
            "tags": len(tags),
            "branch_list": [b["name"] for b in branches],
            "tag_list": [t["name"] for t in tags],
        }
    except Exception as e:
        return {"healthy": False, "error": str(e)}


def get_api_baseline():
    try:
        r = requests.get(f"{API_URL}/admin/ai-evolution/api/data", timeout=5)
        if r.status_code == 200:
            data = r.json()
            return {"healthy": True, "baseline": data.get("baseline")}
        return {"healthy": False, "error": f"HTTP {r.status_code}"}
    except Exception as e:
        return {"healthy": False, "error": str(e)}


def check_barerepo_branch_sync():
    bare_repo = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git")
    try:
        result = subprocess.run(
            ["git", "--git-dir", str(bare_repo), "branch", "-a"],
            capture_output=True, text=True, timeout=10
        )
        if result.returncode != 0:
            return {"healthy": False, "error": result.stderr[:200]}
        bare_branches = [
            l.strip().replace("remotes/origin/", "").replace("* ", "")
            for l in result.stdout.strip().split("\n") if l.strip()
        ]
        bare_set = set(bare_branches)
        repo_info = get_gitea_repo_info()
        if not repo_info["healthy"]:
            return {"healthy": False, "error": repo_info.get("error")}
        gitea_set = set(repo_info["branch_list"])
        missing_in_gitea = bare_set - gitea_set
        extra_in_gitea = gitea_set - bare_set
        return {
            "healthy": len(missing_in_gitea) == 0,
            "bare_count": len(bare_set),
            "gitea_count": len(gitea_set),
            "missing_in_gitea": list(missing_in_gitea),
            "extra_in_gitea": list(extra_in_gitea),
        }
    except Exception as e:
        return {"healthy": False, "error": str(e)}


def check_stuck_commands():
    stuck = []
    threshold_minutes = 30
    if RUNNING_DIR.exists():
        now = time.time()
        for cmd_file in RUNNING_DIR.iterdir():
            if cmd_file.name.endswith(".md"):
                mtime = cmd_file.stat().st_mtime
                age_minutes = (now - mtime) / 60
                if age_minutes > threshold_minutes:
                    stuck.append({
                        "name": cmd_file.name,
                        "age_minutes": round(age_minutes, 1),
                        "path": str(cmd_file),
                    })
    return {"healthy": len(stuck) == 0, "stuck_count": len(stuck), "stuck_commands": stuck}



def run_all_checks():
    """Run all health checks and return results."""
    state = load_state()
    results = {}
    patches_created = []
    
    # 1. Gitea version
    log.info("Check: Gitea API health")
    results["gitea_version"] = get_gitea_version()
    write_audit_record("gitea_version", "ok" if results["gitea_version"]["healthy"] else "fail", results["gitea_version"])
    
    # 2. API baseline
    log.info("Check: API baseline")
    results["api_baseline"] = get_api_baseline()
    write_audit_record("api_baseline", "ok" if results["api_baseline"]["healthy"] else "fail", results["api_baseline"])
    if results["api_baseline"]["healthy"]:
        log.info(f"  Baseline: {results['api_baseline'].get('baseline')}")
    
    # 3. Bare repo branch sync
    log.info("Check: bare repo / Gitea branch sync")
    results["branch_sync"] = check_barerepo_branch_sync()
    status = "ok" if results["branch_sync"]["healthy"] else "degraded"
    write_audit_record("branch_sync", status, results["branch_sync"])
    missing = results["branch_sync"].get("missing_in_gitea", [])
    if missing:
        log.warning(f"  {len(missing)} branches missing in Gitea DB")
        for b in missing:
            log.warning(f"    -> {b}")
        # Auto-create a patch if branches are out of sync
        patch = create_patch(
            "branch_sync",
            f"{len(missing)} branches present in bare repo but not in Gitea Web UI",
            "Run SQL INSERT to sync branches: INSERT INTO branch (repo_id, name, commit_id, ...) "
            "for each missing branch. Verify with bare repo SHA.",
            ["runtime/gitea-bin/data/... (SQL via gitea CLI or direct DB)"]
        )
        patches_created.append(patch)
    
    # 4. Stuck commands
    log.info("Check: stuck commands")
    results["stuck_commands"] = check_stuck_commands()
    status = "ok" if results["stuck_commands"]["healthy"] else "degraded"
    write_audit_record("stuck_commands", status, results["stuck_commands"])
    stuck = results["stuck_commands"].get("stuck_commands", [])
    if stuck:
        log.warning(f"  {len(stuck)} commands stuck in running/")
        for c in stuck:
            log.warning(f"    -> {c['name']} (age: {c['age_minutes']}m)")
    
    # Update state
    state["checks"] = {
        "last_run": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "results": {k: {"healthy": v.get("healthy"), "error": v.get("error")} for k, v in results.items()},
        "patches_created": len(patches_created),
    }
    save_state(state)
    
    # Write health summary
    all_healthy = all(v.get("healthy", False) for v in results.values())
    summary = {
        "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "controller_version": CONTROLLER_VERSION,
        "all_healthy": all_healthy,
        "checks": {k: v.get("healthy") for k, v in results.items()},
        "patches_pending": len(patches_created),
    }
    HEALTH_FILE.parent.mkdir(parents=True, exist_ok=True)
    HEALTH_FILE.write_text(json.dumps(summary, indent=2, ensure_ascii=False))
    
    return results, patches_created


def run_forever():
    log.info(f"Autonomous Controller v{CONTROLLER_VERSION} started")
    log.info(f"Poll interval: {POLL_INTERVAL}s")
    log.info(f"Workspace: {WORKSPACE}")
    log.info(f"Gitea: {GITEA_URL}, API: {API_URL}")
    
    while True:
        try:
            log.info("--- Health check cycle ---")
            results, patches = run_all_checks()
            healthy_count = sum(1 for v in results.values() if v.get("healthy"))
            total = len(results)
            log.info(f"Results: {healthy_count}/{total} healthy")
            if patches:
                log.warning(f"Created {len(patches)} patch(es) for owner review")
            else:
                log.info("No issues detected requiring patches")
        except Exception as e:
            log.exception(f"Check cycle failed: {e}")
        
        time.sleep(POLL_INTERVAL)


def run_once():
    log.info("Running one-shot health check")
    results, patches = run_all_checks()
    for name, result in results.items():
        status = "OK" if result.get("healthy") else "FAIL"
        log.info(f"  [{status}] {name}")
        if not result.get("healthy") and result.get("error"):
            log.info(f"         error: {result['error'][:200]}")
    if patches:
        log.info(f"Patches created: {len(patches)}")
    else:
        log.info("No patches created")


if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--once":
        run_once()
    else:
        run_forever()