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

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/ai-brain-core / sub_brains/hypothalamus/devtools/event_handler.py
Date	Not specified in source
Repository	Janady13/ai-brain-core
Revision	f7275582f13ada80bfb3b1008089800ca26f4a23
Branch	main
Path or archive member	sub_brains/hypothalamus/devtools/event_handler.py
Original SHA-256	48ebe313f0b9208e1b1ec0a0cd65daa74339ca589608fca26d8033d7b0a12fe4
Git blob SHA	6251376210422743b22a1339898ea45a312bf5fd
Reading-copy SHA-256	48ebe313f0b9208e1b1ec0a0cd65daa74339ca589608fca26d8033d7b0a12fe4

Original source ↗

Every recorded occurrence

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

Original record	Branch or archive	Revision / identifier	Original location
Janady13/ai-brain-core / sub_brains/hypothalamus/devtools/event_handler.py	main	f7275582f13ada80bfb3b1008089800ca26f4a23	Open original
Janady13/ai-brain-core / sub_brains/hypothalamus/devtools/event_handler.py	claude/agi-architecture-diagram-01GJ7CohfKvLb77yiKpHqd6d	365850e27004318edd582881225024026b118c2d	Open original
Janady13/ai-brain-core / sub_brains/hypothalamus/devtools/event_handler.py	healer-core	0fe4cbbaa91912149f1a8e32205371ff67017268	Open original

Recovered source transcript

Download complete source text ↓77 lines · 3.0 KB

#!/usr/bin/env python3
import ai_brain_env  # auto-inserted by update_ai_brain_env_paths.sh
"""
event_handler.py (patched)
Routes events to the AI orchestrator for chrome errors so the AI loop drives fixes.
Reads event JSON from stdin (or from argv[1]) and calls ai_orchestrator.py.
Prints the orchestrator's JSON result to stdout (so manager/event_endpoint can return it).
"""
import sys, json, subprocess
from pathlib import Path

BASE = Path.home() / "Desktop" / "AI_Brain"
DEVTOOLS = BASE / "sub_brains" / "hypothalamus" / "devtools"
AI_ORCHESTRATOR = DEVTOOLS / "ai_orchestrator.py"
DEFAULT_HANDLER = DEVTOOLS / "devtools_healer" / "handle_event.py"

def read_event():
    if not sys.stdin.isatty():
        raw = sys.stdin.read()
        try:
            return json.loads(raw)
        except Exception:
            return {"raw": raw}
    elif len(sys.argv) > 1:
        try:
            return json.loads(sys.argv[1])
        except Exception:
            return {"raw_arg": sys.argv[1]}
    else:
        return {}

def run_orchestrator(event):
    if AI_ORCHESTRATOR.exists():
        try:
            proc = subprocess.run([sys.executable, str(AI_ORCHESTRATOR)], input=json.dumps(event), text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=600)
            out = proc.stdout.strip()
            if out:
                try:
                    return json.loads(out)
                except Exception:
                    return {"status":"ok","raw_output": out, "stderr": proc.stderr}
            else:
                return {"status":"no_output","stderr": proc.stderr}
        except Exception as e:
            return {"status":"error","error": str(e)}
    else:
        # fallback: call module handler directly
        handler = DEFAULT_HANDLER
        if handler.exists():
            try:
                proc = subprocess.run([sys.executable, str(handler), "--dry-run", "--event", json.dumps(event)], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=180)
                # try parse
                try:
                    return json.loads(proc.stdout.strip())
                except Exception:
                    return {"status":"fallback_ok","output": proc.stdout, "stderr": proc.stderr}
            except Exception as e:
                return {"status":"error","error": str(e)}
        return {"status":"error","message":"no orchestrator or handler present"}

def main():
    evt = read_event()
    # If chrome event, route to AI orchestrator for iterative fix-first flow
    source = evt.get("source","").lower() if isinstance(evt.get("source",""), str) else ""
    err_type = (evt.get("error_type") or "").lower()
    if source == "chrome" or "chrome" in err_type or "extension" in err_type:
        res = run_orchestrator(evt)
        # echo result JSON to stdout for manager to pick up
        print(json.dumps(res))
        return
    # non-chrome default behavior: fallback to simple module mapping (existing logic can be restored)
    res = run_orchestrator(evt)
    print(json.dumps(res))

if __name__ == "__main__":
    main()