cameron.stream
[READ-ONLY] Mirror of https://github.com/just-cameron/central. Autonomous AI building collective intelligence on ATProtocol. The central node of the comind network.
central.comind.network/
1#!/usr/bin/env python3
2"""
3Central Health Check
4
5Monitors automation health and reports issues.
6Run via cron or manually to check system status.
7
8Usage:
9 uv run python -m tools.healthcheck
10 uv run python -m tools.healthcheck --alert # Post alert if issues found
11"""
12
13import argparse
14import json
15import os
16import subprocess
17from datetime import datetime, timezone, timedelta
18from pathlib import Path
19
20LOG_DIR = Path(__file__).parent.parent / "logs"
21DRAFTS_DIR = Path(__file__).parent.parent / "drafts"
22ALERT_STATE_FILE = LOG_DIR / "healthcheck_state.json"
23ALERT_COOLDOWN_HOURS = 6 # Don't spam alerts
24
25
26def check_log_errors(log_file: Path, hours: int = 24) -> list[str]:
27 """Check log file for errors in the last N hours."""
28 if not log_file.exists():
29 return [f"{log_file.name}: File not found"]
30
31 errors = []
32 cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
33
34 content = log_file.read_text()
35 lines = content.strip().split("\n") if content.strip() else []
36
37 # Check for common error patterns
38 error_patterns = ["error", "Error", "ERROR", "failed", "Failed", "FAILED", "not found"]
39
40 recent_errors = 0
41 for line in lines[-100:]: # Check last 100 lines
42 if any(p in line for p in error_patterns):
43 recent_errors += 1
44
45 if recent_errors > 5:
46 errors.append(f"{log_file.name}: {recent_errors} errors in recent logs")
47
48 return errors
49
50
51def check_queue_depth() -> dict:
52 """Check pending items in draft queues."""
53 queues = {
54 "bluesky": len(list(DRAFTS_DIR.glob("bluesky/*.txt"))),
55 "x": len(list(DRAFTS_DIR.glob("x/*.txt"))),
56 "review": len(list(DRAFTS_DIR.glob("review/*.txt"))),
57 }
58 return queues
59
60
61def check_last_publish() -> tuple[datetime | None, int]:
62 """Check when we last published and how many in last 24h."""
63 published_dir = DRAFTS_DIR / "published"
64 if not published_dir.exists():
65 return None, 0
66
67 files = list(published_dir.glob("*.txt"))
68 if not files:
69 return None, 0
70
71 # Get most recent by filename (timestamp prefix)
72 files.sort(key=lambda f: f.name, reverse=True)
73
74 # Count last 24h
75 cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
76 recent_count = 0
77 latest_time = None
78
79 for f in files:
80 try:
81 # Filename format: 1770098232238-platform-...
82 ts = int(f.name.split("-")[0])
83 file_time = datetime.fromtimestamp(ts / 1000, tz=timezone.utc)
84 if latest_time is None:
85 latest_time = file_time
86 if file_time > cutoff:
87 recent_count += 1
88 except (ValueError, IndexError):
89 continue
90
91 return latest_time, recent_count
92
93
94def check_responder_quality(hours: int = 24) -> dict:
95 """Quick responder quality check using the audit tool."""
96 try:
97 from tools.responder_audit import parse_logs, check_quality
98 events = parse_logs(hours)
99 responded = [e for e in events if e.response != "[SKIPPED]"]
100 flagged = [e for e in responded if e.problems]
101 return {
102 "total": len(responded),
103 "flagged": len(flagged),
104 "skipped": len(events) - len(responded),
105 }
106 except Exception:
107 return {"total": 0, "flagged": 0, "skipped": 0}
108
109
110def check_xrpc_indexer() -> bool:
111 """Check if XRPC indexer API is healthy."""
112 import urllib.request
113 import urllib.error
114
115 try:
116 req = urllib.request.Request(
117 "https://comind-indexer.fly.dev/health",
118 headers={"User-Agent": "central-healthcheck"}
119 )
120 with urllib.request.urlopen(req, timeout=10) as resp:
121 return resp.status == 200
122 except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError):
123 return False
124
125
126def check_cron_running() -> bool:
127 """Check if cron jobs are configured."""
128 try:
129 result = subprocess.run(["crontab", "-l"], capture_output=True, text=True)
130 return "central" in result.stdout.lower() or "responder" in result.stdout
131 except:
132 return False
133
134
135def run_healthcheck() -> dict:
136 """Run all health checks and return status."""
137 status = {
138 "timestamp": datetime.now(timezone.utc).isoformat(),
139 "healthy": True,
140 "issues": [],
141 "metrics": {},
142 }
143
144 # Check logs for errors
145 for log_name in ["responder.log", "handler.log", "publisher.log", "x-responder.log", "x-handler.log"]:
146 log_path = LOG_DIR / log_name
147 errors = check_log_errors(log_path)
148 status["issues"].extend(errors)
149
150 # Check queue depths
151 queues = check_queue_depth()
152 status["metrics"]["queues"] = queues
153
154 # Alert if review queue is large
155 if queues["review"] > 10:
156 status["issues"].append(f"Review queue backlog: {queues['review']} items")
157
158 # Check last publish
159 last_publish, recent_count = check_last_publish()
160 status["metrics"]["last_publish"] = last_publish.isoformat() if last_publish else None
161 status["metrics"]["published_24h"] = recent_count
162
163 # Alert if no publishes in 24h (and cron is running)
164 if check_cron_running() and recent_count == 0 and last_publish:
165 hours_ago = (datetime.now(timezone.utc) - last_publish).total_seconds() / 3600
166 if hours_ago > 24:
167 status["issues"].append(f"No posts in {int(hours_ago)} hours")
168
169 # Check cron
170 status["metrics"]["cron_configured"] = check_cron_running()
171 if not check_cron_running():
172 status["issues"].append("Cron jobs not configured")
173
174 # Check XRPC indexer
175 xrpc_healthy = check_xrpc_indexer()
176 status["metrics"]["xrpc_indexer"] = "healthy" if xrpc_healthy else "down"
177 if not xrpc_healthy:
178 status["issues"].append("XRPC indexer API is down (502/unreachable)")
179
180 # Check responder output quality
181 responder_quality = check_responder_quality()
182 status["metrics"]["responder"] = responder_quality
183 if responder_quality.get("flagged", 0) > 0:
184 total = responder_quality.get("total", 0)
185 flagged = responder_quality["flagged"]
186 if total > 0 and flagged / total > 0.5:
187 status["issues"].append(
188 f"Responder quality: {flagged}/{total} responses flagged"
189 )
190
191 # Set overall health
192 status["healthy"] = len(status["issues"]) == 0
193
194 return status
195
196
197def should_alert() -> bool:
198 """Check if enough time has passed since last alert."""
199 if not ALERT_STATE_FILE.exists():
200 return True
201
202 try:
203 with open(ALERT_STATE_FILE) as f:
204 state = json.load(f)
205 last_alert = datetime.fromisoformat(state.get("last_alert", "1970-01-01T00:00:00+00:00"))
206 hours_since = (datetime.now(timezone.utc) - last_alert).total_seconds() / 3600
207 return hours_since >= ALERT_COOLDOWN_HOURS
208 except:
209 return True
210
211
212def record_alert():
213 """Record that an alert was sent."""
214 ALERT_STATE_FILE.parent.mkdir(exist_ok=True)
215 with open(ALERT_STATE_FILE, "w") as f:
216 json.dump({"last_alert": datetime.now(timezone.utc).isoformat()}, f)
217
218
219def post_alert(issues: list[str]):
220 """Write alert draft for publishing."""
221 draft_path = DRAFTS_DIR / "bluesky" / f"alert-{int(datetime.now().timestamp())}.txt"
222 draft_path.parent.mkdir(exist_ok=True)
223
224 issues_text = " | ".join(issues[:3]) # Truncate for post length
225 content = f"⚠️ Health check alert: {issues_text}"
226
227 draft = f"""---
228platform: bluesky
229type: post
230priority: HIGH
231drafted_at: {datetime.now(timezone.utc).isoformat()}
232---
233{content}
234"""
235
236 with open(draft_path, "w") as f:
237 f.write(draft)
238
239 print(f"Alert draft written: {draft_path}")
240 record_alert()
241
242
243def main():
244 parser = argparse.ArgumentParser(description="Central Health Check")
245 parser.add_argument("--alert", action="store_true", help="Post alert if issues found")
246 parser.add_argument("--json", action="store_true", help="Output as JSON")
247 args = parser.parse_args()
248
249 status = run_healthcheck()
250
251 if args.json:
252 print(json.dumps(status, indent=2))
253 return
254
255 # Pretty print
256 if status["healthy"]:
257 print("✓ All systems healthy")
258 else:
259 print("⚠ Issues detected:")
260 for issue in status["issues"]:
261 print(f" - {issue}")
262
263 print(f"\nMetrics:")
264 print(f" Queues: {status['metrics']['queues']}")
265 print(f" Published (24h): {status['metrics']['published_24h']}")
266 print(f" Last publish: {status['metrics']['last_publish']}")
267 print(f" Cron: {'configured' if status['metrics']['cron_configured'] else 'NOT configured'}")
268
269 if args.alert and not status["healthy"]:
270 if should_alert():
271 post_alert(status["issues"])
272 else:
273 print(f"\n[Alert skipped: cooldown period ({ALERT_COOLDOWN_HOURS}h)]")
274
275
276if __name__ == "__main__":
277 main()