Janady13/ai-brain-core / sub_brains/hypothalamus/devtools/ai_orchestrator.py

1#!/usr/bin/env python3 2import ai_brain_env # auto-inserted by update_ai_brain_env_paths.sh 3""" 4AI Orchestrator for Hypothalamus Devtools 5 6Usage: 7 - Called by event_handler.py with the event JSON on stdin (or as arg). 8 - Environment variables: 9 OPENAI_API_KEY (optional) - if set, orchestrator will call OpenAI ChatCompletion API. 10 LLM_ENDPOINT (optional) - ws/rest endpoint for a local LLM; takes precedence over OpenAI. 11 ALLOW_AUTO_APPLY=1 (optional) - allow automatic --apply operations (use with caution). 12 HUMAN_APPROVAL_FILE (optional) - path to a file to place a human-approval request; orchestrator will wait for this file content "approve" to proceed with risky actions. 13Behavior: 14 - Request plan from LLM (or fallback heuristic). 15 - Execute plan actions one-by-one (dry-run by default). 16 - Report results back to LLM; iterate until resolved or max rounds. 17 - Outputs final JSON result on stdout and writes an audit file to logs/. 18""" 19import os, sys, json, subprocess, time, uuid, shutil 20from pathlib import Path 21from typing import List, Dict, Any 22 23BASE = Path.home() / "Desktop" / "AI_Brain" 24DEVTOOLS = BASE / "sub_brains" / "hypothalamus" / "devtools" 25LOGS = BASE / "logs" 26AUDIT_DIR = LOGS / "ai_orchestrator" 27AUDIT_DIR.mkdir(parents=True, exist_ok=True) 28 29OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") 30LLM_ENDPOINT = os.getenv("LLM_ENDPOINT") # optional local LLM REST 31ALLOW_AUTO_APPLY = os.getenv("ALLOW_AUTO_APPLY") == "1" 32HUMAN_APPROVAL_FILE = os.getenv("HUMAN_APPROVAL_FILE", str(BASE / "status" / "human_approval.request")) 33 34MAX_ITERATIONS = int(os.getenv("AI_MAX_ITER", "5")) 35ACTION_TIMEOUT = int(os.getenv("AI_ACTION_TIMEOUT", "180")) # seconds per action 36 37# Allowed actions schema (the orchestrator will only run these) 38# action.type in {"fetch", "module_dry", "module_apply", "restart", "notify_human", "rollback", "noop"} 39# module name must be a folder under DEVTOOLS 40def log(msg): 41 ts = time.strftime("%Y-%m-%d %H:%M:%S") 42 line = f"{ts} {msg}" 43 print(line, file=sys.stderr) 44 try: 45 with open(LOGS / "ai_orchestrator.log", "a") as f: 46 f.write(line + "\n") 47 except Exception: 48 pass 49 50def audit_write(event_id, payload): 51 fn = AUDIT_DIR / f"{event_id}_{int(time.time())}.json" 52 try: 53 fn.write_text(json.dumps(payload, indent=2)) 54 except Exception: 55 pass 56 57# --- LLM adapter --------------------------------------------------------- 58def run_local_llm(prompt: str) -> str: 59 # Very small fallback: rule-based mapper. Replace with your local LLM call if available. 60 # This fallback should only be used for prototyping. 61 if "extension_crash" in prompt: 62 plan = { 63 "goal": "attempt to repair browser extension crash", 64 "actions": [ 65 {"type":"fetch", "module":"devtools_healer"}, 66 {"type":"module_dry", "module":"devtools_healer", "args":{}}, 67 {"type":"module_apply", "module":"devtools_healer", "args":{}}, 68 {"type":"restart", "service":"chrome"}, 69 {"type":"noop"} 70 ] 71 } 72 return json.dumps(plan) 73 # generic fallback 74 plan = {"goal":"investigate","actions":[{"type":"fetch","module":"devtools_healer"},{"type":"module_dry","module":"devtools_healer"},{"type":"noop"}]} 75 return json.dumps(plan) 76 77def run_openai_chat(system_prompt: str, user_prompt: str) -> str: 78 # Use Chat Completions if openai package available and API key set 79 try: 80 import openai 81 openai.api_key = [credential removed] 82 messages = [ 83 {"role":"system", "content": system_prompt}, 84 {"role":"user", "content": user_prompt} 85 ] 86 resp = openai.ChatCompletion.create(model="gpt-4o-mini", messages=messages, temperature=0.0, max_tokens=512) 87 return resp.choices[0].message.content 88 except Exception as e: 89 log(f"OpenAI call failed: {e}") 90 return run_local_llm(user_prompt) 91 92def ask_llm_for_plan(event: Dict[str,Any], history: List[Dict[str,Any]]=[]) -> Dict[str,Any]: 93 """ 94 Ask LLM to propose a plan for the given event. 95 Expected response: JSON with keys: goal (str) and actions (list). 96 actions: [{type, module, args...}, ...] 97 """ 98 system = ( 99 "You are an AI maintainer agent for a device called Hypothalamus. " 100 "Produce a conservative, safety-first JSON plan to investigate and fix the given event. " 101 "Allowed action types: fetch, module_dry, module_apply, restart, notify_human, rollback, noop. " 102 "Return pure JSON: {\"goal\":..., \"actions\":[...]}. " 103 "Each action: include type, module (if relevant), and optional args. " 104 "Prefer dry-run first. If an apply is recommended, include explanation in args.explain and only request apply if likely safe." 105 ) 106 user = json.dumps({"event": event, "history": history}, indent=2) 107 if OPENAI_API_KEY: 108 out = run_openai_chat(system, user) 109 elif LLM_ENDPOINT: 110 # If you have a local REST LLM endpoint, post to it (this is a placeholder) 111 try: 112 import requests 113 r = requests.post(LLM_ENDPOINT, json={"system":system,"user":user}, timeout=30) 114 out = r.text 115 except Exception as e: 116 log(f"LLM endpoint call failed: {e}") 117 out = run_local_llm(user) 118 else: 119 out = run_local_llm(user) 120 # Parse JSON from output (robust parsing) 121 try: 122 # Try to find first JSON object in output 123 first = out.strip() 124 # Some LLMs return code fences; strip them 125 if first.startswith("```"): 126 first = first.split("```",2)[-1].strip() 127 plan = json.loads(first) 128 return plan 129 except Exception as e: 130 log(f"LLM output parse failed: {e}; raw output: {out}") 131 return {"goal":"llm_parse_error","actions":[{"type":"noop"}]} 132 133# --- action executors --------------------------------------------------- 134def safe_subprocess(cmd: List[str], timeout=ACTION_TIMEOUT) -> Dict[str,Any]: 135 try: 136 p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=timeout) 137 return {"rc": p.returncode, "output": p.stdout[:4000]} 138 except subprocess.TimeoutExpired: 139 return {"rc":124, "output":"timeout"} 140 except Exception as e: 141 return {"rc":1, "output": f"exception: {e}"} 142 143def action_fetch(module: str) -> Dict[str,Any]: 144 fetcher = DEVTOOLS / "fetcher.py" 145 if fetcher.exists(): 146 p = subprocess.Popen([sys.executable, str(fetcher), "--module", module], stdout=subprocess.PIPE, stderr=subprocess.PIPE) 147 return {"rc":0, "info": f"fetcher started pid={p.pid}"} 148 # fallback: if a git-backed module has repo, try git pull in module dir 149 moddir = DEVTOOLS / module 150 if (moddir / ".git").exists(): 151 res = safe_subprocess(["git","-C", str(moddir), "pull", "--ff"], timeout=60) 152 return {"rc": res["rc"], "output": res.get("output")} 153 return {"rc":1, "error":"no fetcher and no git repo"} 154 155def action_module_dry(module: str, args: Dict[str,Any]) -> Dict[str,Any]: 156 # Prefer module's handle_event.py with --dry-run 157 handler = DEVTOOLS / module / "handle_event.py" 158 alt = DEVTOOLS / module / f"{module}.py" 159 if handler.exists(): 160 cmd = [sys.executable, str(handler), "--dry-run", "--event", json.dumps(args.get("event", {}))] 161 res = safe_subprocess(cmd, timeout=ACTION_TIMEOUT) 162 return {"rc": res["rc"], "output": res.get("output")} 163 elif alt.exists(): 164 cmd = [sys.executable, str(alt), "--dry-run", "--event", json.dumps(args.get("event", {}))] 165 res = safe_subprocess(cmd, timeout=ACTION_TIMEOUT) 166 return {"rc": res["rc"], "output": res.get("output")} 167 return {"rc":2, "error":"no handler found"} 168 169def action_module_apply(module: str, args: Dict[str,Any]) -> Dict[str,Any]: 170 # Safety: require ALLOW_AUTO_APPLY or human approval 171 if not ALLOW_AUTO_APPLY: 172 # request human approval file 173 # write a request and return status asking for human approval 174 req = {"module":module, "action":"apply", "args": args, "timestamp": time.time()} 175 Path(HUMAN_APPROVAL_FILE).write_text(json.dumps(req, indent=2)) 176 return {"rc":3, "error":"requires human approval; request written to HUMAN_APPROVAL_FILE", "file": HUMAN_APPROVAL_FILE} 177 handler = DEVTOOLS / module / "handle_event.py" 178 alt = DEVTOOLS / module / f"{module}.py" 179 if handler.exists(): 180 cmd = [sys.executable, str(handler), "--apply", "--event", json.dumps(args.get("event", {}))] 181 res = safe_subprocess(cmd, timeout=ACTION_TIMEOUT) 182 return {"rc": res["rc"], "output": res.get("output")} 183 elif alt.exists(): 184 cmd = [sys.executable, str(alt), "--apply", "--event", json.dumps(args.get("event", {}))] 185 res = safe_subprocess(cmd, timeout=ACTION_TIMEOUT) 186 return {"rc": res["rc"], "output": res.get("output")} 187 return {"rc":2, "error":"no handler found"} 188 189def action_restart(service: str) -> Dict[str,Any]: 190 # Limited set: "chrome", "manager", "watchdog", or custom 191 if service == "chrome": 192 # best-effort: pkill chrome, then try to start (user experience may vary) 193 rc1 = safe_subprocess(["pkill", "-f", "Chrome"]) # may fail harmlessly 194 # starting Chrome headless is not desirable; we rely on user to restart. Return result. 195 return {"rc":0, "info":"chrome killed (if running). user may need to restart UI"} 196 elif service == "manager": 197 # use manager scripts 198 lm = BASE / "launch_manager.sh" 199 if lm.exists(): 200 safe_subprocess([str(lm)]) 201 return {"rc":0, "info":"manager launch invoked"} 202 return {"rc":1, "error":"manager launcher missing"} 203 elif service == "watchdog": 204 p = BASE / "sub_brains" / "watchdog" / "launch.sh" 205 if p.exists(): 206 subprocess.Popen([str(p), "--bg"]) 207 return {"rc":0, "info":"watchdog launcher started"} 208 return {"rc":1, "error":"watchdog launcher missing"} 209 else: 210 return {"rc":2, "error":"unknown service"} 211 212def action_notify_human(payload: Dict[str,Any]) -> Dict[str,Any]: 213 # write a human-readable file or send via manager notify 214 nf = BASE / "status" / f"human_notification_{int(time.time())}.json" 215 nf.write_text(json.dumps(payload, indent=2)) 216 return {"rc":0, "file": str(nf)} 217 218# action dispatcher 219def run_action(action: Dict[str,Any]) -> Dict[str,Any]: 220 typ = action.get("type") 221 module = action.get("module") 222 args = action.get("args", {}) 223 # attach event into args so module handlers can read it 224 if "event" not in args and "event" in action: 225 args["event"] = action["event"] 226 if typ == "fetch": 227 return action_fetch(module) 228 if typ == "module_dry": 229 return action_module_dry(module, args) 230 if typ == "module_apply": 231 return action_module_apply(module, args) 232 if typ == "restart": 233 return action_restart(action.get("service")) 234 if typ == "notify_human": 235 return action_notify_human(action.get("payload", {})) 236 if typ == "rollback": 237 # optional: call a module's rollback handler 238 mdir = DEVTOOLS / module 239 rb = mdir / "rollback.sh" 240 if rb.exists(): 241 return safe_subprocess([str(rb)], timeout=120) 242 return {"rc":1, "error":"no rollback handler"} 243 if typ == "noop": 244 return {"rc":0, "info":"noop"} 245 return {"rc":127, "error":"unknown action type"} 246 247# --- main loop ---------------------------------------------------------- 248def orchestrate(event: Dict[str,Any]) -> Dict[str,Any]: 249 event_id = event.get("id") or str(uuid.uuid4()) 250 audit = {"event": event, "rounds": []} 251 # iterative loop 252 history = [] 253 for rnd in range(MAX_ITERATIONS): 254 log(f"[AI] requesting plan (round {rnd+1})") 255 plan = ask_llm_for_plan(event, history) 256 audit["rounds"].append({"plan":plan}) 257 actions = plan.get("actions", []) if isinstance(plan, dict) else [] 258 if not actions: 259 log("No actions proposed by LLM; exiting") 260 break 261 round_results = [] 262 resolved = False 263 for act in actions: 264 # attach event to actions if not present 265 if "event" not in act: 266 act["event"] = event 267 log(f"[AI] executing action: {act}") 268 res = run_action(act) 269 round_results.append({"action":act, "result": res}) 270 audit["rounds"][-1].setdefault("results", []).append({"action":act, "result":res}) 271 # if any action indicates immediate success, let LLM know 272 if act.get("type") in ("module_apply","restart") and res.get("rc",1) == 0: 273 # probe to see if issue resolved by running a verification (module_dry or custom) 274 # ask LLM if more steps needed 275 pass 276 # simple heuristic: if module_apply returned rc==0, check for success marker 277 if act.get("type") == "module_apply" and res.get("rc",1) == 0: 278 # optimistic: consider resolved and break to re-evaluate 279 resolved = True 280 break 281 history.append({"plan":plan, "results":round_results}) 282 audit_write(event_id, audit) 283 if resolved: 284 log("[AI] resolved in this round") 285 return {"status":"ok","event_id":event_id,"audit":audit} 286 # else continue loop: feed results back to LLM via history 287 log("[AI] max iterations reached or no resolution") 288 return {"status":"unresolved","event_id":event_id,"audit":audit} 289 290def main(): 291 # accept JSON event from stdin or arg 292 if not sys.stdin.isatty(): 293 raw = sys.stdin.read() 294 try: 295 event = json.loads(raw) 296 except Exception: 297 event = {"raw": raw} 298 elif len(sys.argv) > 1: 299 try: 300 event = json.loads(sys.argv[1]) 301 except Exception: 302 event = {"raw_arg": sys.argv[1]} 303 else: 304 print(json.dumps({"status":"error","message":"no input"})) 305 sys.exit(1) 306 log(f"AI Orchestrator invoked for event: {event.get('error_type','<unknown>')}") 307 result = orchestrate(event) 308 print(json.dumps(result)) 309 audit_write(event.get("id", str(uuid.uuid4())), result) 310 311if __name__ == "__main__": 312 main() 313