Janady13/ai-brain-core / orchestrator/orchestrator.py

1#!/usr/bin/env python3 2""" 3Production-ready asynchronous orchestrator. 4- Uses a SQLite DB (WAL) for jobs & metrics 5- Parses env files (static) for candidates (no 'source') 6- Communicates with local adapter via UDS or localhost token 7- Dry-run default, live mode requires enabling via enable_live.sh 8""" 9import os, json, asyncio, time, sqlite3, logging, hashlib 10from pathlib import Path 11import httpx 12from prometheus_client import start_http_server, Counter, Gauge 13from filelock import FileLock 14 15BASE = Path(__file__).resolve().parent.parent 16CFG = json.load(open(BASE / 'orchestrator' / 'config.json')) 17DB_PATH = Path(CFG['db_path']).resolve() 18LOG_FILE = Path(CFG['log_file']).resolve() 19ADAPTER_SOCKET = Path(CFG['adapter_socket']).resolve() 20ADAPTER_TOKEN_FILE = Path(CFG['adapter_token_file']).resolve() 21 22# logging 23logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s: %(message)s', 24 handlers=[logging.FileHandler(LOG_FILE), logging.StreamHandler()]) 25 26# metrics 27JOBS_PROCESSED = Counter('hypothalumus_jobs_processed_total', 'Total jobs processed') 28JOB_FAILURES = Counter('hypothalumus_job_failures_total', 'Job failures') 29ACTIVE_JOBS = Gauge('hypothalumus_active_jobs', 'Currently active jobs') 30 31# DB helper 32def init_db(): 33 conn = sqlite3.connect(DB_PATH, timeout=30, isolation_level=None) 34 conn.execute('PRAGMA journal_mode=WAL;') 35 c = conn.cursor() 36 c.execute(''' 37 CREATE TABLE IF NOT EXISTS jobs ( 38 id TEXT PRIMARY KEY, 39 var TEXT, 40 url TEXT, 41 created_at INTEGER, 42 status TEXT, 43 attempts INTEGER, 44 last_response TEXT, 45 created_src TEXT 46 ) 47 ''') 48 c.execute(''' 49 CREATE TABLE IF NOT EXISTS metrics ( 50 url TEXT PRIMARY KEY, 51 success INTEGER, 52 failure INTEGER, 53 ewma_latency REAL, 54 last_failure INTEGER, 55 open_until INTEGER 56 ) 57 ''') 58 conn.commit() 59 return conn 60 61# safe env parsing (static): read wrappers or desktop envs for *_API_ entries 62def load_candidates(env_dir): 63 candidates = {} 64 for f in Path(env_dir).glob('*.env'): 65 for line in f.read_text(encoding='utf8').splitlines(): 66 line=line.strip() 67 if not line or line.startswith('#'): continue 68 if line.lower().startswith('export '): line = line[7:].strip() 69 if '=' not in line: continue 70 key,val=line.split('=',1) 71 key=key.strip(); val=val.strip().strip('\'"') 72 if '_API_' in key: 73 prefix=key.split('_API_')[0].lower() 74 candidates.setdefault(prefix,[]).append((key,val)) 75 return candidates 76 77# insert job to DB (idempotent) 78def insert_job(conn, jid, var, url, src): 79 ts=int(time.time()) 80 with conn: 81 conn.execute('INSERT OR IGNORE INTO jobs (id,var,url,created_at,status,attempts,created_src) VALUES (?,?,?,?,?,?,?)', 82 (jid,var,url,ts,'pending',0,src)) 83 84# simple circuit breaker helpers 85def is_open(conn, url): 86 r = conn.execute('SELECT open_until FROM metrics WHERE url=?', (url,)).fetchone() 87 if not r: return False 88 return (r[0] or 0) > int(time.time()) 89 90def mark_success(conn, url, latency): 91 row = conn.execute('SELECT success,failure,ewma_latency FROM metrics WHERE url=?',(url,)).fetchone() 92 if row: 93 s,f,ew=row 94 s=s+1 95 ew = (0.2*latency) + (0.8*(ew or latency)) 96 conn.execute('UPDATE metrics SET success=?, ewma_latency=? WHERE url=?',(s,ew,url)) 97 else: 98 conn.execute('INSERT INTO metrics(url,success,failure,ewma_latency,last_failure,open_until) VALUES (?,?,?,?,?,?)', 99 (url,1,0,latency, None, 0)) 100 101def mark_failure(conn, url): 102 row = conn.execute('SELECT success,failure FROM metrics WHERE url=?',(url,)).fetchone() 103 if row: 104 s,f=row 105 f=f+1 106 open_until = int(time.time()) + CFG['circuit_open_seconds'] if f >= CFG['failures_to_open'] else 0 107 conn.execute('UPDATE metrics SET failure=?, last_failure=?, open_until=? WHERE url=?',(f,int(time.time()),open_until,url)) 108 else: 109 conn.execute('INSERT INTO metrics(url,success,failure,ewma_latency,last_failure,open_until) VALUES (?,?,?,?,?,?)', 110 (url,0,1,None,int(time.time()),0)) 111 112# perform the probe / action 113async def call_url(client, url): 114 try: 115 r = await client.get(url, timeout=CFG['http_timeout']) 116 return True, r.status_code, r.elapsed.total_seconds(), r.text[:1000] 117 except Exception as e: 118 return False, None, None, str(e) 119 120# call adapter (for privileged actions) via HTTP-to-UDS or direct HTTP with token 121async def call_adapter(action, payload): 122 # Adapter expects JSON: {"action":"restart_network","params":{},"id":"..."} 123 token = ADAPTER_TOKEN_FILE.read_text().strip() if ADAPTER_TOKEN_FILE.exists() else None 124 headers = {} 125 if token: 126 headers['X-Adapter-Token'] = token 127 # adapter may be on UDS; but httpx needs http://localhost + transport 128 url = f"http://unix:{ADAPTER_SOCKET}:/{action}" 129 async with httpx.AsyncClient() as client: 130 try: 131 r = await client.post(url, json=payload, headers=headers, timeout=CFG['http_timeout']) 132 return True, r.status_code, r.text[:2000] 133 except Exception as e: 134 return False, None, str(e) 135 136async def process_job(conn, job): 137 jid, var, url = job 138 logging.info(f"PROCESS JOB {jid} -> {url}") 139 ACTIVE_JOBS.inc() 140 try: 141 if is_open(conn, url): 142 logging.warning("Circuit open for %s; skipping", url) 143 return 144 async with httpx.AsyncClient() as client: 145 ok, status, latency, body = await call_url(client, url) 146 if ok and (status is None or status < 400): 147 logging.info("Probe ok for %s", url) 148 mark_success(conn, url, latency or 0.1) 149 conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', 150 ('ok', json.dumps({'status':status,'lat':latency}), jid)) 151 JOBS_PROCESSED.inc() 152 else: 153 logging.warning("Probe failed for %s : %s", url, body) 154 mark_failure(conn, url) 155 # If not dry_run, issue adapter or remote action depending on var prefix rule 156 if not CFG.get('dry_run', True): 157 # Example adaptability: if var prefix 'network' then call adapter restart 158 prefix = var.split('_API_')[0].lower() 159 if prefix in ('network','device','chrome','safari'): 160 action = prefix + "_repair" 161 payload = {"url":url,"var":var,"job_id":jid} 162 ok2, code2, resp2 = await call_adapter(action, payload) 163 if ok2 and (code2 is None or int(code2) < 400): 164 conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', 165 ('actioned', json.dumps({'adapter':resp2}), jid)) 166 else: 167 conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', 168 ('failed', json.dumps({'error':resp2}), jid)) 169 JOB_FAILURES.inc() 170 else: 171 conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', 172 ('failed', json.dumps({'probe_error':body}), jid)) 173 JOB_FAILURES.inc() 174 else: 175 # dry-run: just mark dry-run-failed 176 conn.execute('UPDATE jobs SET status=?, attempts=attempts+1, last_response=? WHERE id=?', 177 ('dryrun-failed', json.dumps({'probe':body}), jid)) 178 JOB_FAILURES.inc() 179 finally: 180 ACTIVE_JOBS.dec() 181 182async def worker_loop(): 183 conn = sqlite3.connect(DB_PATH, timeout=30, isolation_level=None) 184 while True: 185 rows = conn.execute("SELECT id,var,url FROM jobs WHERE status IN ('pending','dryrun-failed') ORDER BY created_at LIMIT 20").fetchall() 186 if not rows: 187 await asyncio.sleep(CFG['jobs_poll_seconds']) 188 continue 189 tasks=[] 190 sem = asyncio.Semaphore(CFG['max_parallel']) 191 async with httpx.AsyncClient() as client: 192 loop = asyncio.get_event_loop() 193 for r in rows: 194 jid,var,url = r 195 # run process_job in task 196 t = loop.create_task(process_job(conn, (jid,var,url))) 197 tasks.append(t) 198 if tasks: 199 await asyncio.gather(*tasks) 200 await asyncio.sleep(0.1) 201 202def ensure_db_jobs_from_env(conn): 203 # parse env dir wrappers for *_API_* and insert into jobs table if missing 204 env_dir = BASE / 'env' 205 for f in env_dir.glob('*.env'): 206 for line in f.read_text(encoding='utf8').splitlines(): 207 l=line.strip() 208 if not l or l.startswith('#'): continue 209 if l.lower().startswith('export '): l = l[7:].strip() 210 if '=' not in l: continue 211 key,val = l.split('=',1) 212 if '_API_' in key: 213 key=key.strip() 214 val=val.strip().strip('\'"') 215 jid = hashlib.sha1((key+val).encode()).hexdigest()[:20] 216 insert_job(conn, jid, key, val, f.name) 217 218def main(): 219 DB_PATH.parent.mkdir(parents=True, exist_ok=True) 220 conn = init_db() 221 ensure_db_jobs_from_env(conn) 222 # start prometheus metrics 223 try: 224 start_http_server(CFG.get('prometheus_port',9001)) 225 except Exception as e: 226 logging.warning("Prometheus port may already be in use: %s", e) 227 # main loop 228 asyncio.run(worker_loop()) 229 230if __name__ == '__main__': 231 main() 232