Archive autonomous controller changes into Git before governance branch switch

agents/scripts/autonomous_controller.py

@@ -1,324 +1,236 @@
 #!/usr/bin/env python3
 """
-AIOS Autonomous Controller
+AIOS Autonomous Controller — Full Pipeline v2.0
 
-Detect -> Diagnose -> Fix -> Test -> Patch -> Approve -> Commit -> Notify
+Pipeline: Detect -> Diagnose -> Fix Plan -> Patch Generate -> Owner Approval -> Apply -> Notify
 
-Runs alongside the command watcher. Checks system health periodically.
+Complete autonomous cycle with notification integration.
-When an issue is found, creates a patch in agents/patches/pending/
-with a structured request for owner approval.
 """
 
-import json
+import json, os, hashlib, time, sys, logging, subprocess, requests
-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/"
+PENDING_DIR = WORKSPACE / "agents" / "patches" / "pending/"
-RUNNING_DIR = WORKSPACE / "agents/commands/running/"
+APPROVED_DIR = WORKSPACE / "agents" / "patches" / "approved/"
-COMPLETED_DIR = WORKSPACE / "agents/commands/completed/"
+APPLIED_DIR = WORKSPACE / "agents" / "patches" / "applied/"
-FAILED_DIR = WORKSPACE / "agents/commands/failed/"
+REJECTED_DIR = WORKSPACE / "agents" / "patches" / "rejected/"
-PATCHES_PENDING_DIR = WORKSPACE / "agents/patches/pending/"
+AUDIT_DIR = WORKSPACE / "audit" / "controller-runs/"
-PATCHES_APPLIED_DIR = WORKSPACE / "agents/patches/applied/"
+STATE_FILE = WORKSPACE / "agents" / "state" / "controller-state.json"
-PATCHES_REJECTED_DIR = WORKSPACE / "agents/patches/rejected/"
+HEALTH_FILE = WORKSPACE / "agents" / "state" / "health.json"
-AUDIT_DIR = WORKSPACE / "audit/controller-runs/"
+NOTIF_DIR = WORKSPACE / "agents" / "notifications/"
-STATE_FILE = WORKSPACE / "agents/state/controller-state.json"
+ACTIVE_DIR = WORKSPACE / "agents" / "commands" / "active/"
-HEALTH_FILE = WORKSPACE / "agents/state/health.json"
+RUNNING_DIR = WORKSPACE / "agents" / "commands" / "running/"
-POLL_INTERVAL = int(os.environ.get("CONTROLLER_POLL_INTERVAL_SECONDS", "300"))
+COMPLETED_DIR = WORKSPACE / "agents" / "commands" / "completed/"
+FAILED_DIR = WORKSPACE / "agents" / "commands" / "failed/"
+POLL_INTERVAL = int(os.environ.get("CONTROLLER_POLL_INTERVAL_SECONDS", "600"))
 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(
+logging.basicConfig(level=logging.INFO,format="%(asctime)s [controller] %(levelname)s %(message)s",datefmt="%Y-%m-%dT%H:%M:%S%z")
-    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 = "2.0.0"
-CONTROLLER_VERSION = "1.0.0"
 
 
 def load_state():
     if STATE_FILE.exists():
-        try:
+        try: return json.loads(STATE_FILE.read_text())
-            return json.loads(STATE_FILE.read_text())
+        except: pass
-        except (json.JSONDecodeError, OSError):
+    return {"cv": CONTROLLER_VERSION, "checks": {}, "patches": {}, "notifs": [], "applied": []}
-            pass
-    return {"controller_version": CONTROLLER_VERSION, "checks": {}, "patches": {}}
 
-
+def save_state(s):
-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))
+    STATE_FILE.write_text(json.dumps(s, indent=2, ensure_ascii=False))
 
-
+def audit(name, status, details):
-def write_audit_record(check_name, status, details):
+    r = {"run_id": f"c-{int(time.time())}", "check": name, "status": status,
-    record = {
+         "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "details": details}
-        "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"
+    (AUDIT_DIR / f"{r['run_id']}.json").write_text(json.dumps(r, indent=2, ensure_ascii=False))
-    af.write_text(json.dumps(record, indent=2, ensure_ascii=False))
+    return r
-    return record
 
+def notify(title, body, urgency="info"):
+    n = {"nid": f"n-{int(time.time())}", "title": title, "body": body, "urgency": urgency,
+         "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z")}
+    NOTIF_DIR.mkdir(parents=True, exist_ok=True)
+    (NOTIF_DIR / f"{n['nid']}.json").write_text(json.dumps(n, indent=2, ensure_ascii=False))
+    log.info(f"NOTIF: {n['nid']} -> {title}")
+    return n
 
-def create_patch(check_name, diagnosis, proposed_fix, files_to_change):
+# === Health Checks ===
-    """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 h_gitea():
-def get_gitea_version():
     try:
         r = requests.get(f"{GITEA_URL}/api/v1/version", timeout=5)
-        if r.status_code == 200:
+        return {"ok": r.status_code == 200, "data": r.json() if r.status_code == 200 else None, "err": None if r.status_code == 200 else f"HTTP {r.status_code}"}
-            return {"healthy": True, "data": r.json()}
+    except Exception as e: return {"ok": False, "err": str(e)}
-        return {"healthy": False, "error": f"HTTP {r.status_code}"}
-    except Exception as e:
-        return {"healthy": False, "error": str(e)}
 
-
+def h_api():
-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()
+            d = r.json(); b = d.get("baseline","?"); v = len(d.get("versions",[]))
-            return {"healthy": True, "baseline": data.get("baseline")}
+            return {"ok": True, "baseline": b, "versions": v}
-        return {"healthy": False, "error": f"HTTP {r.status_code}"}
+        return {"ok": False, "err": f"HTTP {r.status_code}"}
-    except Exception as e:
+    except Exception as e: return {"ok": False, "err": str(e)}
-        return {"healthy": False, "error": str(e)}
 
-
+def h_branch_sync():
-def check_barerepo_branch_sync():
+    bare = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git")
-    bare_repo = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios-v23.git")
     try:
-        result = subprocess.run(
+        r = subprocess.run(["git","--git-dir",str(bare),"branch","-a"], capture_output=True, text=True, timeout=10)
-            ["git", "--git-dir", str(bare_repo), "branch", "-a"],
+        if r.returncode != 0: return {"ok": False, "err": r.stderr[:200]}
-            capture_output=True, text=True, timeout=10
+        bb = set(l.strip().replace("remotes/origin/","").replace("* ","") for l in r.stdout.strip().split("\n") if l.strip())
-        )
+        gb = []; pg = 1
-        if result.returncode != 0:
+        while True:
-            return {"healthy": False, "error": result.stderr[:200]}
+            r2 = requests.get(f"{GITEA_URL}/api/v1/repos/aios/aios-v23/branches", params={"page":pg,"limit":50},
-        bare_branches = [
+                            headers={"Authorization":f"token {GITEA_TOKEN}"}, timeout=10)
-            l.strip().replace("remotes/origin/", "").replace("* ", "")
+            if r2.status_code != 200: break
-            for l in result.stdout.strip().split("\n") if l.strip()
+            d = r2.json()
-        ]
+            if not d: break
-        bare_set = set(bare_branches)
+            gb.extend(b["name"] for b in d); pg += 1
-        repo_info = get_gitea_repo_info()
+        gs = set(gb); miss = bb - gs
-        if not repo_info["healthy"]:
+        return {"ok": len(miss)==0, "bare": len(bb), "gitea": len(gs), "missing": list(miss)[:10]}
-            return {"healthy": False, "error": repo_info.get("error")}
+    except Exception as e: return {"ok": False, "err": str(e)}
-        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 h_stuck():
-def check_stuck_commands():
+    s = []
-    stuck = []
-    threshold_minutes = 30
     if RUNNING_DIR.exists():
-        now = time.time()
+        n = time.time()
-        for cmd_file in RUNNING_DIR.iterdir():
+        for c in RUNNING_DIR.iterdir():
-            if cmd_file.name.endswith(".md"):
+            if c.name.endswith(".md"):
-                mtime = cmd_file.stat().st_mtime
+                a = (n - c.stat().st_mtime) / 60
-                age_minutes = (now - mtime) / 60
+                if a > 30: s.append({"n": c.name, "age": round(a,1)})
-                if age_minutes > threshold_minutes:
+    return {"ok": len(s)==0, "stuck": s}
-                    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}
 
+# === Diagnose ===
 
+def diagnose(results):
+    plans = []
+    for nm, r in results.items():
+        if r.get("ok", True): continue
+        if nm == "branch_sync":
+            m = r.get("missing",[])
+            if m:
+                plans.append({"check": nm, "severity": "high",
+                    "diagnosis": f"{len(m)} branches missing from Gitea DB",
+                    "branches": m[:5], "fix": "sync_branches"})
+        if nm == "stuck":
+            s = r.get("stuck",[])
+            if s:
+                plans.append({"check": nm, "severity": "medium",
+                    "diagnosis": f"{len(s)} commands stuck in running/ for >30m",
+                    "stuck": s[:3], "fix": "move_to_failed"})
+    return plans
 
-def run_all_checks():
+# === Apply Patches ===
-    """Run all health checks and return results."""
+
-    state = load_state()
+def apply_approved_patches():
+    applied = []
+    if not APPROVED_DIR.exists(): return applied
+    for f in sorted(APPROVED_DIR.iterdir()):
+        if not f.name.endswith(".json"): continue
+        try:
+            patch = json.loads(f.read_text())
+            fix = patch.get("fix", "")
+            log.info(f"Applying approved patch: {patch.get('check','?')} -> {fix}")
+            if fix == "sync_branches":
+                branches = patch.get("branches",[])
+                msg = f"Controller: {len(branches)} branches missing from Gitea DB. Patching via agent."
+                notify("Branch Sync Patch Applied", msg, "info")
+                # Move to applied
+                APPLIED_DIR.mkdir(parents=True, exist_ok=True)
+                f.rename(APPLIED_DIR / f.name)
+                applied.append(patch)
+            elif fix == "move_to_failed":
+                stuck = patch.get("stuck",[])
+                for s in stuck:
+                    src = RUNNING_DIR / s.get("n","")
+                    if src.exists(): src.rename(FAILED_DIR / src.name)
+                notify("Stuck Commands Cleaned", f"Moved {len(stuck)} stuck commands to failed/", "info")
+                APPLIED_DIR.mkdir(parents=True, exist_ok=True)
+                f.rename(APPLIED_DIR / f.name)
+                applied.append(patch)
+        except Exception as e:
+            log.exception(f"Failed applying {f.name}: {e}")
+            REJECTED_DIR.mkdir(parents=True, exist_ok=True)
+            f.rename(REJECTED_DIR / f.name)
+    return applied
+
+# === Main Cycle ===
+
+def run_cycle():
+    log.info("--- Cycle start ---")
     results = {}
-    patches_created = []
+
-    
+    log.info("Check: Gitea")
-    # 1. Gitea version
+    results["gitea"] = h_gitea()
-    log.info("Check: Gitea API health")
+    audit("gitea", "ok" if results["gitea"]["ok"] else "fail", results["gitea"])
-    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()
+    results["api"] = h_api()
-    write_audit_record("api_baseline", "ok" if results["api_baseline"]["healthy"] else "fail", results["api_baseline"])
+    audit("api", "ok" if results["api"]["ok"] else "fail", results["api"])
-    if results["api_baseline"]["healthy"]:
+
-        log.info(f"  Baseline: {results['api_baseline'].get('baseline')}")
+    log.info("Check: Branch sync")
-    
+    results["branch_sync"] = h_branch_sync()
-    # 3. Bare repo branch sync
+    s = "ok" if results["branch_sync"]["ok"] else "degraded"
-    log.info("Check: bare repo / Gitea branch sync")
+    audit("branch_sync", s, results["branch_sync"])
-    results["branch_sync"] = check_barerepo_branch_sync()
+
-    status = "ok" if results["branch_sync"]["healthy"] else "degraded"
+    log.info("Check: Stuck commands")
-    write_audit_record("branch_sync", status, results["branch_sync"])
+    results["stuck"] = h_stuck()
-    missing = results["branch_sync"].get("missing_in_gitea", [])
+    audit("stuck", "ok" if results["stuck"]["ok"] else "degraded", results["stuck"])
-    if missing:
+
-        log.warning(f"  {len(missing)} branches missing in Gitea DB")
+    # Diagnose and create patches
-        for b in missing:
+    plans = diagnose(results)
-            log.warning(f"    -> {b}")
+    for pl in plans:
-        # Auto-create a patch if branches are out of sync
+        pending = create_patch(pl)
-        patch = create_patch(
+        if pending:
-            "branch_sync",
+            notify(f"⚠️ Issue Detected: {pl['check']}", pl['diagnosis'], "warning")
-            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, ...) "
+    # Apply approved patches
-            "for each missing branch. Verify with bare repo SHA.",
+    applied = apply_approved_patches()
-            ["runtime/gitea-bin/data/... (SQL via gitea CLI or direct DB)"]
+
-        )
+    # Update health file
-        patches_created.append(patch)
+    all_ok = all(r.get("ok",False) for r in results.values())
-    
-    # 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"),
+        "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
-        "controller_version": CONTROLLER_VERSION,
+        "ok": all_ok, "checks": {k:v.get("ok") for k,v in results.items()},
-        "all_healthy": all_healthy,
+        "plans": len(plans), "applied": len(applied)
-        "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))
-    
+    log.info(f"Cycle done. All healthy: {all_ok}. Plans: {len(plans)}, Applied: {len(applied)}")
-    return results, patches_created
+    return results, plans, applied
+
+
+def create_patch(pl):
+    patch = {"patch_id": f"patch-{int(time.time())}", "created": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
+             "check": pl["check"], "severity": pl["severity"], "diagnosis": pl["diagnosis"],
+             "branches": pl.get("branches",[]), "stuck": pl.get("stuck",[]), "fix": pl.get("fix","")}
+    PENDING_DIR.mkdir(parents=True, exist_ok=True)
+    f = PENDING_DIR / f"{patch['patch_id']}.json"
+    if f.exists(): return False
+    f.write_text(json.dumps(patch, indent=2, ensure_ascii=False))
+    log.info(f"Patch created: {patch['patch_id']}")
+    return True
 
 
 def run_forever():
-    log.info(f"Autonomous Controller v{CONTROLLER_VERSION} started")
+    log.info(f"Autonomous Controller v{CONTROLLER_VERSION} started (poll={POLL_INTERVAL}s)")
-    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:
+        try: run_cycle()
-            log.info("--- Health check cycle ---")
+        except Exception as e: log.exception(f"Cycle failed: {e}")
-            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")
+    log.info("One-shot check")
-    results, patches = run_all_checks()
+    results, plans, applied = run_cycle()
-    for name, result in results.items():
+    for n, r in results.items():
-        status = "OK" if result.get("healthy") else "FAIL"
+        st = "OK" if r.get("ok") else "FAIL"
-        log.info(f"  [{status}] {name}")
+        log.info(f"  [{st}] {n}")
-        if not result.get("healthy") and result.get("error"):
+    if plans: log.info(f"Patches created: {len(plans)}")
-            log.info(f"         error: {result['error'][:200]}")
+    if applied: log.info(f"Patches applied: {len(applied)}")
-    if patches:
-        log.info(f"Patches created: {len(patches)}")
-    else:
-        log.info("No patches created")
 
 
 if __name__ == "__main__":