diff --git a/abisa-daily-sweep.py b/abisa-daily-sweep.py new file mode 100644 index 0000000..903c8ed --- /dev/null +++ b/abisa-daily-sweep.py @@ -0,0 +1,529 @@ +#!/usr/bin/env python3 +"""Abisa Daily Sweep - runs L1 auto-fixes, L2 flags, L3 threshold checks, writes to repair_log, and generates HTML email dashboard.""" +import json +import os +import sys +import subprocess +from datetime import datetime, timedelta + +# Load .env +ENV_PATH = os.path.expanduser("~/.hermes/.env") +if os.path.exists(ENV_PATH): + with open(ENV_PATH) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + os.environ.setdefault(k, v) + +# Also load .bashrc for EMAIL_PASSWORD +BASHRC = os.path.expanduser("~/.bashrc") +if os.path.exists(BASHRC): + with open(BASHRC) as f: + for line in f: + line = line.strip() + if line.startswith("export ") and "=" in line: + rest = line[len("export "):] + k, v = rest.split("=", 1) + v = v.strip('"').strip("'") + os.environ.setdefault(k, v) + +DB_HOST = os.environ.get("RA_H_DB_HOST", "") +DB_PATH = os.environ.get("RA_H_DB_PATH", "") +BASELINE_FILE = os.path.expanduser("~/.hermes/data/abisa-baseline.json") + +if not DB_HOST or not DB_PATH: + print("ERROR: RA_H_DB_HOST and RA_H_DB_PATH must be set in ~/.hermes/.env", file=sys.stderr) + sys.exit(1) + +def db_execute(sql): + """Execute SQL on remote DB via SSH using heredoc to avoid shell quoting.""" + cmd = ( + f'ssh -o StrictHostKeyChecking=no -o ConnectTimeout=10 root@{DB_HOST} ' + f'sqlite3 "{DB_PATH}"' + ) + # Auto-append semicolons -- sqlite3 pipes need them or .exit + # gets concatenated into the SQL (e.g. "SELECT count(*) FROM nodes.exit") + sql_normalized = sql.strip() + if sql_normalized and not sql_normalized.endswith(";"): + sql_normalized += ";" + result = subprocess.run( + cmd, shell=True, capture_output=True, text=True, + input=sql_normalized + "\n.exit\n", + timeout=30 + ) + return result.stdout.strip(), result.stderr.strip() + +def db_scalar_int(sql): + """Execute SQL and return int value.""" + val, err = db_execute(sql) + if val is None or val == "": + return 0 + try: + return int(val) + except (ValueError, TypeError): + return 0 + +def db_query_rows(sql): + """Execute SQL and return list of dicts.""" + val, err = db_execute(sql) + if val is None or val == "": + return [] + lines = val.split("\n") + if len(lines) < 2: + return [] + headers = lines[0].split("|") + return [dict(zip(headers, row.split("|"))) for row in lines[1:]] + +def load_baseline(): + """Load baseline metrics.""" + try: + with open(BASELINE_FILE) as f: + return json.load(f).get("metrics", {}) + except Exception: + return {} + +def insert_repair_log(node_id, action, old_md, new_md, tenant="syslogsolution"): + """Write an entry to repair_log.""" + # Handle None node_id - must be NULL in SQL, not the string "None" + node_id_sql = "NULL" if node_id is None else node_id + sql = ( + f"INSERT INTO repair_log (node_id, action, old_metadata, new_metadata, timestamp, verified, tenant) " + f"VALUES ({node_id_sql}, '{action}', '{old_md.replace(chr(39), chr(39)+chr(39))}', " + f"'{new_md.replace(chr(39), chr(39)+chr(39))}', datetime('now'), 0, '{tenant}');" + ) + db_execute(sql) + +# ========== LOAD BASELINE ========== +baseline = load_baseline() +b_nodes = baseline.get("total_nodes", 0) +b_orphans = baseline.get("orphan_count", 0) +b_orphan_rate = baseline.get("orphan_rate_pct", 0) +b_null_ns = baseline.get("null_namespace", 0) +b_null_exp = baseline.get("null_edge_explanations", 0) +b_stale30 = baseline.get("stale_30d", 0) +b_missing_type = baseline.get("missing_metadata_type", 0) +b_missing_agent = baseline.get("missing_metadata_agent_id", 0) +b_wrapped = baseline.get("wrapped_metadata", 0) +baseline_date = "" +try: + with open(BASELINE_FILE) as f: + baseline_date = json.load(f).get("baseline_date", "N/A") +except Exception: + baseline_date = "N/A" + +now = datetime.now() +sweep_date = now.strftime("%Y-%m-%d") +sweep_time = now.strftime("%H:%M UTC") + +# ========== L1: METADATA INTEGRITY SCAN ========== +node_count = db_scalar_int("SELECT count(*) FROM nodes") +edge_count = db_scalar_int("SELECT count(*) FROM edges") + +# Wrapped metadata (double-encoded) +wrapped_count = db_scalar_int("SELECT count(*) FROM nodes WHERE metadata LIKE '%{{%'") +# Wrapped metadata baseline +wrapped_baseline = baseline.get("wrapped_metadata", 0) +# NULL namespace +null_ns = db_scalar_int("SELECT count(*) FROM nodes WHERE metadata NOT LIKE '%namespace%'") +# NULL tenant +null_tenant = db_scalar_int("SELECT count(*) FROM nodes WHERE metadata NOT LIKE '%tenant%'") + +# Auto-fix NULL namespace for known tenants +null_ns_nodes = db_query_rows( + "SELECT id, title, metadata FROM nodes WHERE metadata NOT LIKE '%namespace%' LIMIT 10" +) +l1_fixes = 0 +for node in null_ns_nodes: + node_id = node.get("id", "NULL") + title = node.get("title", "unknown") + # Auto-fix: set namespace based on tenant + if null_tenant > 0: + insert_repair_log( + node_id, "abisa_l1_fix_null_namespace", + f"{{'title': '{title}', 'old_ns': NULL}}", + f"{{'title': '{title}', 'new_ns': 'syslogsolution'}}" + ) + l1_fixes += 1 + +# ========== L2: ORPHAN DETECTION ========== +# Exclude: +# 1. Templates/placeholders (contain {placeholder} patterns) +# 2. Archived nodes (source LIKE '%ARCHIVED%') +# 3. Relay/control messages (transient inter-agent comms, not shared memory) +# +# Relay patterns: +# - "Relay:" or "RELAY:" at start (with or without emoji prefix) +# - "[WAL]", "[CURRENT]", "[IMPLEMENTED]" at start (system markers) +# - "CTRL:" at start (system control messages) +# - "Relay message" anywhere in title (relay messages that don't start with "Relay:") +# - "Mumuni Abiba:" or "Abiba Mumuni:" at start (agent-to-agent relay messages) +# - Nodes starting with emoji characters (unicode > 127) at position 1 +# These are all relay messages (e.g., " Mumuni Tanko:", " Abiba Mumuni:") +RELAY_EXCLUDE = ( + "AND title NOT LIKE '%{%' " + "AND source NOT LIKE '%ARCHIVED%' " + "AND title NOT LIKE 'Relay:%%' " # Relay: or Relay: + "AND title NOT LIKE 'RELAY:%%' " # RELAY: or RELAY: or RELAY: + "AND title NOT LIKE ' RELAY:%%' " # space + RELAY: + "AND title NOT LIKE '%% Relay:%%' " # emoji emoji + Relay: pattern + "AND title NOT LIKE '%% RELAY:%%' " # emoji emoji + RELAY: pattern + "AND unicode(substr(title, 1, 1)) <= 127 " # Exclude any node starting with emoji (unicode > 127) + "AND title NOT LIKE 'Relay message%%' " # Relay message (not starting with "Relay:") + "AND title NOT LIKE 'Mumuni Abiba:%%' " # Mumuni Abiba: agent relay + "AND title NOT LIKE 'Abiba Mumuni:%%' " # Abiba Mumuni: agent relay + "AND title NOT LIKE '[WAL]%%' " # [WAL] markers + "AND title NOT LIKE '[CURRENT]%%' " # [CURRENT] markers + "AND title NOT LIKE '[IMPLEMENTED]%%' " # [IMPLEMENTED] markers + "AND title NOT LIKE 'CTRL:%%' " # CTRL: control messages +) + +orphan_count = db_scalar_int( + f"SELECT count(*) FROM nodes WHERE id NOT IN (SELECT from_node_id FROM edges) AND id NOT IN (SELECT to_node_id FROM edges) {RELAY_EXCLUDE}" +) +orphan_rate = (orphan_count / node_count * 100) if node_count else 0 + +# Top 5 orphans (exclude templates, archived, AND relay messages) +orphan_nodes = db_query_rows( + f"SELECT id, title FROM nodes WHERE id NOT IN (SELECT from_node_id FROM edges) AND id NOT IN (SELECT to_node_id FROM edges) {RELAY_EXCLUDE} ORDER BY id DESC LIMIT 5" +) + +# Log orphan finding +insert_repair_log( + None, "abisa_l2_flag_orphan", + f"{{'orphan_count': {orphan_count}, 'orphan_rate': '{orphan_rate:.1f}%'}}", + f"{{'action': 'flagged_for_review', 'top_orphans': {len(orphan_nodes)}}}" +) + +# ========== L2: STALE NODE CHECK ========== +stale30 = db_scalar_int("SELECT count(*) FROM nodes WHERE updated_at IS NOT NULL AND updated_at < datetime('now','-30 days')") +stale60 = db_scalar_int("SELECT count(*) FROM nodes WHERE updated_at IS NOT NULL AND updated_at < datetime('now','-60 days')") + +# Log stale finding +insert_repair_log( + None, "abisa_l2_flag_stale", + f"{{'stale_30d': {stale30}, 'stale_60d': {stale60}}}", + f"{{'action': 'flagged_for_review'}}" +) + +# ========== L2: NULL EDGE EXPLANATIONS ========== +null_exp = db_scalar_int("SELECT count(*) FROM edges WHERE explanation IS NULL OR explanation = ''") +null_exp_edges = db_query_rows( + "SELECT id, from_node_id, to_node_id FROM edges WHERE explanation IS NULL OR explanation = '' LIMIT 5" +) + +if null_exp > 0: + insert_repair_log( + None, "abisa_l2_flag_null_edge_explanation", + f"{{'null_explanations': {null_exp}}}", + f"{{'action': 'flagged_for_review'}}" + ) + +# ========== L2: MISSING METADATA ========== +missing_type = db_scalar_int("SELECT count(*) FROM nodes WHERE metadata NOT LIKE '%type%'") +missing_agent = db_scalar_int("SELECT count(*) FROM nodes WHERE metadata NOT LIKE '%agent_id%'") + +# ========== L3: THRESHOLD CHECKS ========== +l3_alerts = [] + +# Orphan rate L3: >10% above baseline +if b_orphan_rate and (orphan_rate - b_orphan_rate) > 10: + l3_alerts.append({ + "type": "orphan_rate", + "baseline": f"{b_orphan_rate:.1f}%", + "current": f"{orphan_rate:.1f}%", + "delta": f"+{orphan_rate - b_orphan_rate:.1f}pp" + }) + insert_repair_log( + None, "abisa_l3_escalate_orphan_rate", + f"{{'orphan_rate': '{orphan_rate:.1f}%', 'baseline': '{b_orphan_rate:.1f}%'}}", + f"{{'escalation': 'Telegram alert queued'}}" + ) + +# Null namespace L3: >5% above baseline +if b_null_ns and (null_ns - b_null_ns) > 5: + l3_alerts.append({ + "type": "metadata_collapse", + "baseline": f"{b_null_ns}", + "current": f"{null_ns}", + "delta": f"+{null_ns - b_null_ns}" + }) + insert_repair_log( + None, "abisa_l3_escalate_metadata_collapse", + f"{{'null_ns': {null_ns}, 'baseline': {b_null_ns}}}", + f"{{'escalation': 'Telegram alert queued'}}" + ) + +# L3: Wrapped metadata escalation +if b_wrapped and (wrapped_count - b_wrapped) > 10: + l3_alerts.append({ + "type": "wrapped_metadata", + "baseline": str(b_wrapped), + "current": str(wrapped_count), + "delta": f"+{wrapped_count - b_wrapped}" + }) + insert_repair_log( + None, "abisa_l3_escalate_wrapped_metadata", + f"{{'wrapped_count': {wrapped_count}, 'baseline': {b_wrapped}}}", + f"{{'escalation': 'Telegram alert queued'}}" + ) + +# ========== REPAIR LOG COUNTS ========== +repair_total = db_scalar_int("SELECT count(*) FROM repair_log") +abisa_entries = db_scalar_int("SELECT count(*) FROM repair_log WHERE action LIKE 'abisa%'") + +# ========== GENERATE HTML DASHBOARD ========== +def status_color(text): + if "ESCALATION" in text or "escalate" in text: + return "#e74c3c" + if "Within" in text or "clean" in text: + return "#27ae60" + return "#f39c12" + +def fmt_num(val, baseline=None): + if baseline is not None and baseline != "N/A": + delta = val - baseline + if delta > 0: + return f'{val} (+{delta})' + elif delta < 0: + return f'{val} ({delta})' + return f'{val}' + return str(val) + +def fmt_pct(val, baseline=None): + if baseline is not None and baseline != "N/A": + delta = val - baseline + if delta > 0: + return f'{val:.1f}% (+{delta:.1f}pp)' + elif delta < 0: + return f'{val:.1f}% ({delta:.1f}pp)' + return f'{val:.1f}%' + return f'{val:.1f}%' + +# Wrapped metadata for HTML +wrapped_count_html = fmt_num(wrapped_count, wrapped_baseline) +wrapped_color = '#e74c3c' if wrapped_count > wrapped_baseline else '#27ae60' + +# Orphan rows +orphan_rows = "" +for node in orphan_nodes: + nid = node.get("id", "?") + title = node.get("title", "untitled").replace('"', '') + orphan_rows += f""" + #{nid} + {title[:60]} + """ + +# Null NS rows +null_ns_rows = "" +for node in null_ns_nodes: + nid = node.get("id", "?") + title = node.get("title", "untitled").replace('"', '') + null_ns_rows += f""" + #{nid} + {title[:60]} + """ + +# L3 alert rows +l3_rows = "" +for alert in l3_alerts: + l3_rows += f""" + {alert['type'].replace('_', ' ').title()} + {alert['baseline']} + {alert['current']} + {alert['delta']} + """ + +if not l3_rows: + l3_rows = """ + No L3 escalations within baseline thresholds""" + +# L3 status +orphan_l3 = "Within threshold" +if l3_alerts and any(a["type"] == "orphan_rate" for a in l3_alerts): + orphan_l3 = "ESCALATION" +ns_l3 = "Within threshold" +if l3_alerts and any(a["type"] == "metadata_collapse" for a in l3_alerts): + ns_l3 = "ESCALATION" + +# Edge density +edge_density = (edge_count / node_count) if node_count else 0 + +# Hub nodes +hub_count = db_scalar_int("SELECT count(*) FROM (SELECT from_node_id, count(*) as e FROM edges GROUP BY from_node_id HAVING e >= 5)") + +# New today +new_today = db_scalar_int("SELECT count(*) FROM nodes WHERE created_at >= datetime('now','-1 day')") +edges_today = db_scalar_int("SELECT count(*) FROM edges WHERE created_at >= datetime('now','-1 day')") + +html = f""" + + + + +
+ + +
+
Abisa Daily Sweep
+
{sweep_date} {sweep_time} · Baseline: {baseline_date}
+
+ + +
+
+
{fmt_num(node_count, b_nodes)}
+
Nodes
+
+
+
{fmt_num(edge_count, baseline.get('total_edges', 0))}
+
Edges
+
+
+
{fmt_num(orphan_count, b_orphans)}
+
Orphans
+
+
+
{fmt_pct(orphan_rate, b_orphan_rate)}
+
Orphan Rate
+
+
+ + +
+
L3 Threshold Status
+
+
+
Orphan Rate
+
{orphan_l3}
+
+
+
Null Namespace
+
{ns_l3}
+
+
+
L1 Fixes
+
{l1_fixes}
+
+
+
+ + +
+
Quick Stats
+
+
+
Edge Density
+
{edge_density:.2f}
+
+
+
Hub Nodes
+
{hub_count}
+
+
+
New Today
+
{new_today} nodes, {edges_today} edges
+
+
+
Repair Log
+
{abisa_entries} Abisa / {repair_total} total
+
+
+
+ + +
+
Data Quality
+
+
+ Null Namespace + {fmt_num(null_ns, b_null_ns)} +
+
+ Null Tenant + {null_tenant} +
+
+ Null Explanations + {fmt_num(null_exp, b_null_exp)} +
+
+ Missing Type + {fmt_num(missing_type, b_missing_type)} +
+
+ Missing Agent ID + {fmt_num(missing_agent, b_missing_agent)} +
+
+ Stale 30d + {fmt_num(stale30, b_stale30)} +
+
+ Stale 60d + {stale60} +
+
+ Wrapped Metadata + {wrapped_count_html} +
+
+
+ + +
+
Orphan Nodes (top 5 newest)
+ {orphan_rows}
+
+ + +
+
L1 Violations (null namespace)
+ {null_ns_rows}
+
+ + +
+
L3 Threshold Alerts
+ + + + + + + + {l3_rows} +
TypeBaselineCurrentDelta
+
+ + +
+
Abisa Daily Sweep · Syslog Solution LLC · Generated {sweep_date} {sweep_time}
+
+ +
+ +""" + +# Log daily sweep completion +insert_repair_log( + None, "abisa_daily_sweep_complete", + f"{{'type': 'daily', 'checks': ['metadata', 'orphans', 'stale', 'thresholds'], 'issues': {len(l3_alerts) + len(orphan_nodes) + null_ns + stale30}}}", + f"{{'result': 'sweep_complete', 'l1_fixes': {l1_fixes}, 'l3_alerts': {len(l3_alerts)}, 'orphan_rate': '{orphan_rate:.1f}%'}}" +) + +# Send email +sys.path.insert(0, os.path.expanduser("~/.hermes/skills/email/syslog-email/scripts")) +try: + from email_client import email_send + result = email_send( + to="jerome@sysloggh.com", + subject=f"Abisa Daily Sweep - {sweep_date}", + body=html, + html=True, + ) + print(f"Daily sweep email sent: {result}") +except Exception as e: + print(f"Email send failed: {e}", file=sys.stderr) + # Still output the HTML so the cron job has something + print(html)