| """AgentGlancer Dashboard Server — serves connectome snapshots + static files""" |
| import http.server |
| import json, os, glob, time |
| from pathlib import Path |
|
|
| PORT = 9011 |
| SNAPSHOT_DIR = Path("/tmp/agent-connectome-snapshots") |
| STATIC_DIR = Path("/home/y7/hayula/web/glancer") |
|
|
| class GlancerServer(http.server.SimpleHTTPRequestHandler): |
| def __init__(self, *args, **kwargs): |
| super().__init__(*args, directory=str(STATIC_DIR), **kwargs) |
| |
| def do_GET(self): |
| if self.path == '/api/connectome': |
| self.send_response(200) |
| self.send_header('Content-Type', 'application/json') |
| self.send_header('Cache-Control', 'no-cache') |
| self.send_header('Access-Control-Allow-Origin', '*') |
| self.end_headers() |
| |
| |
| data = {"agent_list": [], "hubs": [], "bottlenecks": [], "critical_paths": []} |
| |
| if SNAPSHOT_DIR.exists(): |
| snapshots = sorted(SNAPSHOT_DIR.glob("connectome-*.json"), reverse=True) |
| if snapshots: |
| try: |
| with open(snapshots[0]) as f: |
| data = json.load(f) |
| except: |
| pass |
| |
| |
| if not data.get("agent_list"): |
| data = generate_demo_data() |
| |
| self.wfile.write(json.dumps(data).encode()) |
| else: |
| super().do_GET() |
| |
| def log_message(self, format, *args): |
| pass |
|
|
| def generate_demo_data(): |
| """Generate demo data when FFAM snapshots not available.""" |
| return { |
| "agent_list": ["rushd", "wafa", "awf", "dragon", "hermes", "musa", "zeus", "haytham", |
| "saif", "averroes", "orphanim", "uta", "bait", "0xZeus", "exploiter"], |
| "hubs": [ |
| {"agent": "dragon", "degree": 13, "type": "router"}, |
| {"agent": "rushd", "degree": 12, "type": "router"}, |
| {"agent": "haytham", "degree": 11, "type": "aggregator"}, |
| {"agent": "awf", "degree": 10, "type": "worker"}, |
| {"agent": "hermes", "degree": 9, "type": "router"}, |
| ], |
| "bottlenecks": [ |
| {"agent": "rushd", "betweenness": 0.45, "severity": "moderate", "recommendation": "Consider load balancing"}, |
| {"agent": "awf", "betweenness": 0.32, "severity": "moderate", "recommendation": "Monitor"} |
| ], |
| "critical_paths": [ |
| {"path": ["rushd", "dragon", "wafa"], "frequency": 6}, |
| {"path": ["haytham", "musa"], "frequency": 8}, |
| {"path": ["hermes", "awf", "dragon"], "frequency": 5}, |
| {"path": ["rushd", "awf", "wafa"], "frequency": 4}, |
| ], |
| "stats": {"agents": 15, "skills": 5, "events": 200} |
| } |
|
|
| if __name__ == "__main__": |
| import socketserver |
| print(f"[AgentGlancer] Serving on http://localhost:{PORT}") |
| print(f"[AgentGlancer] Reading snapshots from: {SNAPSHOT_DIR}") |
| with socketserver.ThreadingTCPServer(("0.0.0.0", PORT), GlancerServer) as httpd: |
| try: |
| httpd.serve_forever() |
| except KeyboardInterrupt: |
| httpd.shutdown() |
|
|
