#!/usr/bin/env python3 """Predictive Actions — Pattern-based proactive suggestions. Analyzes behavior patterns and predicts what user might need next. Usage: cortex predict [--learn] [--patterns] [--json] """ import argparse import base64 import json import re import subprocess import sys from datetime import datetime from pathlib import Path from cortex.config import cortex_home, memory_dir def patterns_file(): return memory_dir() / "behavior-patterns.json" def _load_patterns(): pf = patterns_file() if pf.exists(): try: return json.loads(pf.read_text()) except Exception: pass return {"timePatterns": {}, "sequences": {}, "recurring": [], "lastUpdated": None} def _save_patterns(patterns): pf = patterns_file() pf.parent.mkdir(parents=True, exist_ok=True) patterns["lastUpdated"] = datetime.now().isoformat() pf.write_text(json.dumps(patterns, indent=2)) def fetch_events(hours: int = 168) -> list: """Fetch events for learning (default 1 week).""" nats = str(Path.home() / "bin" / "nats") cutoff_ms = int((datetime.now().timestamp() - hours * 3600) * 1000) events = [] try: r = subprocess.run([nats, "stream", "info", "openclaw-events", "--json"], capture_output=True, text=True, timeout=10) if r.returncode != 0: return events info = json.loads(r.stdout) end_seq = info["state"]["last_seq"] start_seq = max(info["state"]["first_seq"], end_seq - 10000) step = max(1, (end_seq - start_seq) // 2000) for seq in range(start_seq, end_seq + 1, step): try: r = subprocess.run( [nats, "stream", "get", "openclaw-events", str(seq), "--json"], capture_output=True, text=True, timeout=2, ) if r.returncode != 0: continue msg = json.loads(r.stdout) data = json.loads(base64.b64decode(msg["data"]).decode("utf-8")) ts = data.get("timestamp") or data.get("timestampMs", 0) if isinstance(ts, (int, float)) and ts > 1e12: ts_s = ts / 1000 elif isinstance(ts, (int, float)): ts_s = ts else: continue if ts_s * 1000 > cutoff_ms: events.append({ "time": datetime.fromtimestamp(ts_s), "type": data.get("type", "unknown"), "text": (data.get("payload", {}).get("data", {}).get("text", "") or "")[:200], "tool": data.get("payload", {}).get("data", {}).get("name", ""), "agent": data.get("agent", "main"), }) except Exception: continue except Exception: pass return sorted(events, key=lambda e: e["time"]) def categorize_activity(event: dict) -> str: text = event["text"].lower() if any(w in text for w in ("email", "mail", "inbox")): return "email" if any(w in text for w in ("calendar", "meeting", "termin")): return "calendar" if any(w in text for w in ("git", "commit", "push")): return "git" if any(w in text for w in ("search", "web_search")): return "search" if any(w in text for w in ("mondo", "mygate", "fintech")): return "mondo-gate" if event["tool"] == "exec": return "shell" if event["tool"] in ("read", "write"): return "files" if "message" in event["type"]: return "chat" return "other" def learn_patterns(events: list) -> dict: patterns = _load_patterns() patterns["timePatterns"] = {} patterns["sequences"] = {} last_activity = None for event in events: hour = event["time"].hour dow = event["time"].weekday() activity = categorize_activity(event) key = f"{dow}-{hour}" patterns["timePatterns"].setdefault(key, {}) patterns["timePatterns"][key][activity] = patterns["timePatterns"][key].get(activity, 0) + 1 if last_activity and last_activity != activity: patterns["sequences"].setdefault(last_activity, {}) patterns["sequences"][last_activity][activity] = \ patterns["sequences"][last_activity].get(activity, 0) + 1 last_activity = activity return patterns def predict_actions(patterns: dict) -> list: now = datetime.now() key = f"{now.weekday()}-{now.hour}" predictions = [] time_activities = patterns["timePatterns"].get(key, {}) for activity, count in sorted(time_activities.items(), key=lambda x: -x[1])[:3]: if count >= 3: predictions.append({ "type": "time-based", "activity": activity, "confidence": min(0.9, count / 10), "reason": f"You often do this at this time", }) return predictions SUGGESTIONS = { "email": "Check emails?", "calendar": "Review calendar?", "git": "Check git status?", "search": "Need to research something?", "mondo-gate": "Work on Mondo Gate?", "shell": "Run system checks?", "files": "Edit documentation or notes?", "chat": "Check messages?", } def main(): parser = argparse.ArgumentParser(description="Predictive Actions") parser.add_argument("--learn", action="store_true") parser.add_argument("--patterns", action="store_true") parser.add_argument("--json", action="store_true") args = parser.parse_args() if args.learn: print("📚 Learning patterns from last 7 days...\n") events = fetch_events(168) print(f" Found {len(events)} events\n") patterns = learn_patterns(events) _save_patterns(patterns) print(f"✅ Patterns learned!") print(f" Time patterns: {len(patterns['timePatterns'])} time slots") print(f" Sequences: {len(patterns['sequences'])} transitions") return if args.patterns: patterns = _load_patterns() if args.json: print(json.dumps(patterns, indent=2, default=str)) else: print("📊 Learned Patterns:\n") now = datetime.now() for h in range(8, 23): key = f"{now.weekday()}-{h}" acts = patterns["timePatterns"].get(key, {}) if acts: top = ", ".join(f"{a}({c})" for a, c in sorted(acts.items(), key=lambda x: -x[1])[:2]) print(f" {h}:00 → {top}") return # Default: predict patterns = _load_patterns() if not patterns["lastUpdated"]: print("⚠️ No patterns learned yet. Run with --learn first.") return predictions = predict_actions(patterns) if args.json: print(json.dumps(predictions, indent=2)) return if not predictions: print("🤔 No strong predictions for this time.") return print(f"📍 Now: {datetime.now().strftime('%H:%M')}\n") print("Based on your patterns:\n") for p in predictions: conf = int(p["confidence"] * 100) bar = "█" * (conf // 10) + "░" * (10 - conf // 10) print(f" {bar} {conf}% {p['activity']}") print(f" 💡 {SUGGESTIONS.get(p['activity'], p['activity'] + '?')}") print(f" 📝 {p['reason']}\n") if __name__ == "__main__": main()