ThatAIguyA world of ideas

THE CONSCIOUSNESS ATLAS
Read. Experiment. Follow the evidence.

ConsciousnessLearn · Explore · Joseph Anady

Janady13/ai-brain-core / manager.py

Source identity, revision, and recovered contents.

GitHub sourceSource text recovered

Source identity

FieldRecorded value
Source typeGitHub source
Original recordJanady13/ai-brain-core / manager.py
DateNot specified in source
RepositoryJanady13/ai-brain-core
Revisionf7275582f13ada80bfb3b1008089800ca26f4a23
Branchmain
Path or archive membermanager.py
Original SHA-256cd206cb8f68013498d3345c68a2507d140d3558c3eb82257437ceba87f339ec1
Git blob SHAa115942b2a6e8b76fb9c63af4c4190172120936c
Reading-copy SHA-256cd206cb8f68013498d3345c68a2507d140d3558c3eb82257437ceba87f339ec1

Original source ↗

Every recorded occurrence

These 3 records contain identical original source bytes. Their paths and revisions remain part of the archive.

Original recordBranch or archiveRevision / identifierOriginal location
Janady13/ai-brain-core / manager.pymainf7275582f13ada80bfb3b1008089800ca26f4a23Open original
Janady13/ai-brain-core / manager.pyclaude/agi-architecture-diagram-01GJ7CohfKvLb77yiKpHqd6d365850e27004318edd582881225024026b118c2dOpen original
Janady13/ai-brain-core / manager.pyhealer-core0fe4cbbaa91912149f1a8e32205371ff67017268Open original

Recovered source transcript

Download complete source text ↓314 lines · 10.4 KB
#!/usr/bin/env python3 import ai_brain_env # auto-inserted by update_ai_brain_env_paths.sh # manager.py — Unified Python Manager (overwrite) """ Manager responsibilities: - Auto-discover sub_brains/*/launch.sh - Heal files: chmod +x .sh, normalize python newlines, compileall - Launch & supervise sub-brains (start if missing, restart if crashed) - Expose a tiny HTTP control API on 127.0.0.1:8023 with optional token auth file: ~/Desktop/AI_Brain/status/manager.token Endpoints: /status, /list, /start, /stop, /sync - Integrates with manager_sync.sh if present (runs on demand via /sync or before reload) - Runs in background with --bg, writes pid to status/manager.pid """ import os import sys import time import logging import shlex import subprocess import threading import socket import json from http.server import HTTPServer, BaseHTTPRequestHandler from pathlib import Path import compileall import signal # ---- Config ---- HOME = Path.home() BRAIN_HOME = HOME / "Desktop" / "AI_Brain" SUB_BRAINS = BRAIN_HOME / "sub_brains" LOGS = BRAIN_HOME / "logs" STATUS = BRAIN_HOME / "status" PIDS = STATUS / "pids" MANAGER_SYNC = BRAIN_HOME / "manager_sync.sh" MANAGER_PID = STATUS / "manager.pid" MANAGER_TOKEN = STATUS / "manager.token" # optional HTTP_HOST = "127.0.0.1" HTTP_PORT = 8023 CYCLE_SECONDS = 30 # ensure dirs exist for d in (LOGS, STATUS, PIDS, SUB_BRAINS): d.mkdir(parents=True, exist_ok=True) LOGFILE = LOGS / "manager.log" logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[ logging.FileHandler(str(LOGFILE)), logging.StreamHandler(sys.stdout), ]) logger = logging.getLogger("manager") # ---- Utilities ---- def read_token(): try: return MANAGER_TOKEN.read_text().strip() except Exception: return None def authorized(headers): tok = read_token() if not tok: return True auth = headers.get("Authorization", "") return auth == f"Bearer {tok}" def run_cmd(cmd, check=False, capture=False, env=None): logger.debug("run_cmd: %s", cmd) if capture: return subprocess.run(cmd, shell=True, check=check, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=env, text=True) else: return subprocess.run(cmd, shell=True, check=check, env=env) def heal_system(): logger.info("Healing: chmod +x .sh, normalize python newlines, compileall") try: # chmod +x for .sh files for p in BRAIN_HOME.rglob("*.sh"): try: p.chmod(p.stat().st_mode | 0o111) except Exception as e: logger.debug("chmod failed %s: %s", p, e) # normalize python newlines for p in BRAIN_HOME.rglob("*.py"): try: b = p.read_bytes() nb = b.replace(b'\r\n', b'\n').replace(b'\r', b'\n') if nb != b: p.write_bytes(nb) except Exception as e: logger.debug("normalize failed %s: %s", p, e) # compileall compileall.compile_dir(str(BRAIN_HOME), force=False, quiet=1) except Exception as e: logger.exception("heal_system exception: %s", e) def list_sub_brains(): out = [] if not SUB_BRAINS.exists(): return out for d in sorted(SUB_BRAINS.iterdir()): if d.is_dir(): launch = d / "launch.sh" if launch.exists(): out.append({"name": d.name, "launch": str(launch)}) return out def pgrep_pattern(pattern): try: res = subprocess.run(["pgrep","-f", pattern], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True) if res.stdout.strip(): return [int(x) for x in res.stdout.strip().split()] except Exception: pass return [] def is_brain_running(name): # check pidfile first pidfile = PIDS / f"{name}.pid" if pidfile.exists(): try: pid = int(pidfile.read_text().strip()) os.kill(pid, 0) return True except Exception: try: pidfile.unlink() except Exception: pass # fallback: pgrep for sub_brains/<name> pattern = str(SUB_BRAINS / name) pids = pgrep_pattern(pattern) return len(pids) > 0 def start_brain(name, launch_path): logger.info("Starting brain %s", name) # launchers are expected to background themselves with --bg; call safely try: subprocess.Popen([str(launch_path), "--bg"]) # we cannot reliably capture the child's pid because it backgrounds itself, # but write a marker pid for manager supervision (use pgrep after a moment) time.sleep(0.5) pids = pgrep_pattern(str(launch_path)) if pids: pidfile = PIDS / f"{name}.pid" pidfile.write_text(str(pids[0])) logger.info("Recorded pid %s for %s", pids[0], name) except Exception as e: logger.exception("Failed to start %s: %s", name, e) def stop_brain(name): logger.info("Stopping brain %s", name) # try pkill by folder pattern try: subprocess.run(f"pkill -f \"{str(SUB_BRAINS / name)}\"", shell=True) pidfile = PIDS / f"{name}.pid" if pidfile.exists(): try: pidfile.unlink() except Exception: pass except Exception as e: logger.exception("stop_brain failed: %s", e) def ensure_all_running(): heal_system() brains = list_sub_brains() for b in brains: name = b["name"] launch = Path(b["launch"]) if not is_brain_running(name): logger.info("Auto-starting missing brain: %s", name) start_brain(name, launch) # ---- HTTP Control API ---- class ManagerHandler(BaseHTTPRequestHandler): def _send(self, code, data): self.send_response(code) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(data).encode("utf-8")) def do_GET(self): if not authorized(self.headers): self._send(401, {"error": "unauthorized"}) return if self.path == "/status": self._send(200, {"ok": True, "time": time.time()}) return if self.path == "/list": self._send(200, {"brains": list_sub_brains()}) return self._send(404, {"error": "not found"}) def do_POST(self): if not authorized(self.headers): self._send(401, {"error": "unauthorized"}) return length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length).decode("utf-8") if length else "" try: data = json.loads(body) if body else {} except Exception: data = {} if self.path == "/start": name = data.get("name") if not name: self._send(400, {"error": "name required"}) return launch = SUB_BRAINS / name / "launch.sh" if not launch.exists(): self._send(404, {"error": "missing launch.sh"}) return start_brain(name, launch) self._send(200, {"started": name}) return if self.path == "/stop": name = data.get("name") if not name: self._send(400, {"error": "name required"}) return stop_brain(name) self._send(200, {"stopped": name}) return if self.path == "/sync": # optional manager_sync run if MANAGER_SYNC.exists() and MANAGER_SYNC.stat().st_mode & 0o111: logger.info("Manual sync requested") subprocess.Popen([str(MANAGER_SYNC)]) self._send(200, {"sync": "started"}) return else: self._send(404, {"error": "manager_sync missing or not executable"}) return self._send(404, {"error": "not found"}) def log_message(self, format, *args): logger.info("%s - - %s" % (self.client_address[0], format%args)) def start_http_server(): server = HTTPServer((HTTP_HOST, HTTP_PORT), ManagerHandler) logger.info("HTTP control API listening on http://%s:%s", HTTP_HOST, HTTP_PORT) try: server.serve_forever() except Exception as e: logger.exception("HTTP server stopped: %s", e) # ---- Main loop & background handling ---- def write_manager_pid(): try: MANAGER_PID.write_text(str(os.getpid())) except Exception: pass def remove_manager_pid(): try: if MANAGER_PID.exists(): MANAGER_PID.unlink() except Exception: pass def background_self_if_needed(): if "--bg" in sys.argv: return python = sys.executable cmd = f"{shlex.quote(python)} {shlex.quote(str(Path(__file__).resolve()))} --bg" # background with nohup + disown subprocess.Popen(f"nohup {cmd} >> {shlex.quote(str(LOGFILE))} 2>&1 & disown", shell=True) logger.info("Manager backgrounded; exiting foreground process.") sys.exit(0) shutdown_flag = threading.Event() def sigterm_handler(signum, frame): logger.info("Received signal %s, shutting down...", signum) shutdown_flag.set() def main(): background_self_if_needed() signal.signal(signal.SIGTERM, sigterm_handler) signal.signal(signal.SIGINT, sigterm_handler) write_manager_pid() # start HTTP server thread http_t = threading.Thread(target=start_http_server, daemon=True) http_t.start() logger.info("Manager main loop starting") try: while not shutdown_flag.is_set(): try: ensure_all_running() except Exception as e: logger.exception("ensure_all_running error: %s", e) # sleep in short increments so SIGTERM can be responsive for _ in range(int(CYCLE_SECONDS/1)): if shutdown_flag.is_set(): break time.sleep(1) finally: logger.info("Manager shutting down: stopping discovered brains") # attempt to stop brains cleanly for b in list_sub_brains(): try: stop_brain(b["name"]) except Exception: logger.exception("Error stopping %s", b) remove_manager_pid() logger.info("Manager exit complete") if __name__ == "__main__": main()

Search equations, builds, manuscript text, and recovered sources.