From 641a86b09df5596be293fd27453aa26a223e222e Mon Sep 17 00:00:00 2001 From: Google AI Agent Date: Sun, 5 Apr 2026 22:02:40 +0000 Subject: [PATCH] feat: expose /metrics endpoint in health daemon --- uni-wizard/daemons/health_daemon.py | 52 ++++++++++++++++++++--------- 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/uni-wizard/daemons/health_daemon.py b/uni-wizard/daemons/health_daemon.py index c19cd3e..fe8a967 100644 --- a/uni-wizard/daemons/health_daemon.py +++ b/uni-wizard/daemons/health_daemon.py @@ -24,32 +24,52 @@ class HealthCheckHandler(BaseHTTPRequestHandler): # Suppress default logging pass - def do_GET(self): +def do_GET(self): """Handle GET requests""" if self.path == '/health': self.send_health_response() elif self.path == '/status': self.send_full_status() + elif self.path == '/metrics': + self.send_sovereign_metrics() else: self.send_error(404) - - def send_health_response(self): - """Send simple health check""" - harness = get_harness() - result = harness.execute("health_check") - + + def send_sovereign_metrics(self): + """Send sovereign health metrics as JSON""" try: - health_data = json.loads(result) - status_code = 200 if health_data.get("overall") == "healthy" else 503 - except: - status_code = 503 - health_data = {"error": "Health check failed"} - - self.send_response(status_code) + import sqlite3 + db_path = Path.home() / ".timmy" / "metrics" / "model_metrics.db" + if not db_path.exists(): + data = {"error": "No database found"} + else: + conn = sqlite3.connect(str(db_path)) + row = conn.execute(""" + SELECT local_pct, total_sessions, local_sessions, cloud_sessions, est_cloud_cost, est_saved + FROM sovereignty_score ORDER BY timestamp DESC LIMIT 1 + """).fetchone() + + if row: + data = { + "sovereignty_score": row[0], + "total_sessions": row[1], + "local_sessions": row[2], + "cloud_sessions": row[3], + "est_cloud_cost": row[4], + "est_saved": row[5], + "timestamp": datetime.now().isoformat() + } + else: + data = {"error": "No data"} + conn.close() + except Exception as e: + data = {"error": str(e)} + + self.send_response(200) self.send_header('Content-Type', 'application/json') self.end_headers() - self.wfile.write(json.dumps(health_data).encode()) - + self.wfile.write(json.dumps(data).encode()) + def send_full_status(self): """Send full system status""" harness = get_harness()