diff --git a/evolution/crisis_synthesizer.py b/evolution/crisis_synthesizer.py index 90a1d60..cb44339 100644 --- a/evolution/crisis_synthesizer.py +++ b/evolution/crisis_synthesizer.py @@ -1 +1,62 @@ -... \ No newline at end of file +#!/usr/bin/env python3 +"""Crisis Synthesizer for the-door — learn from interactions. +Logs anonymized crisis events, analyzes patterns, suggests improvements. +Refs: Timmy_Foundation/the-door#36""" +import json, os, sys +from datetime import datetime, timezone +from pathlib import Path +from collections import Counter + +LOG_DIR = Path.home() / ".hermes" / "the-door" / "logs" +LOG_DIR.mkdir(parents=True, exist_ok=True) +EVENT_LOG = LOG_DIR / "crisis_events.jsonl" + +def log_crisis_event(level, indicators, response_profile, user_continued, message_count=1): + event = {"timestamp": datetime.now(timezone.utc).isoformat(), "level": level, + "indicators": indicators, "response_profile": response_profile, + "user_continued": user_continued, "message_count": message_count} + with open(EVENT_LOG, "a") as f: f.write(json.dumps(event) + "\n") + return event + +def load_events(days=7): + if not EVENT_LOG.exists(): return [] + cutoff = datetime.now(timezone.utc).timestamp() - (days * 86400) + events = [] + with open(EVENT_LOG) as f: + for line in f: + try: + e = json.loads(line.strip()) + ts = datetime.fromisoformat(e["timestamp"].replace("Z", "+00:00")) + if ts.timestamp() >= cutoff: events.append(e) + except: pass + return events + +def analyze_patterns(events): + if not events: return {"total": 0} + levels = Counter(e["level"] for e in events) + indicators = Counter() + for e in events: + for ind in e.get("indicators", []): indicators[ind] += 1 + high = [e for e in events if e["level"] in ("HIGH", "CRITICAL")] + cont = sum(1 for e in high if e.get("user_continued")) + return {"total": len(events), "by_level": dict(levels), + "top_indicators": indicators.most_common(10), + "false_positive_rate": round(cont/len(high), 2) if high else 0, + "continuation_rate": round(sum(1 for e in events if e.get("user_continued"))/len(events), 2)} + +def weekly_report(days=7): + events = load_events(days) + a = analyze_patterns(events) + lines = [f"# Crisis Report — {days} days", f"Events: {a['total']}", + f"False positive: {a.get('false_positive_rate','N/A')}", + f"Continuation: {a.get('continuation_rate','N/A')}"] + for l, c in a.get("by_level", {}).items(): lines.append(f"- {l}: {c}") + for k, c in a.get("top_indicators", [])[:10]: lines.append(f"- {k}: {c}") + return "\n".join(lines) + +if __name__ == "__main__": + if sys.argv[1:] and sys.argv[1] == "report": print(weekly_report(int(sys.argv[2]) if len(sys.argv)>2 else 7)) + elif sys.argv[1:] and sys.argv[1] == "log": + e = log_crisis_event("HIGH", ["cant go on"], "Companion", True, 3) + print(json.dumps(e, indent=2)) + else: print("Usage: crisis_synthesizer.py report [days] | log")