#!/usr/bin/env python3 import sys as _sys _sys.path.insert(0, "/root/aria/lib") from aria_logging import get_logger as _get_logger _log = _get_logger("aria-memory-consolidate-night") """ Aria Sleep-Time Memory Consolidation ===================================== Nightly Cron-Job (default 03:00 Europe/Berlin): 1. Liest aria_chat_log letzte 24h aus Supabase 2. Detected Patterns: - Wiederkehrende Fragen (3x+ similar) → Memory-Kandidat - Wiederkehrende Korrekturen ("nein", "anders", "stop") → CORRECTIONS-Kandidat - Skill-Trigger ohne Skill-Match → potential new Skill - User-Frustration-Signale → SELF-IMPROVEMENT-Kandidat 3. Schreibt Vorschlaege in /root/aria-brain/00-Inbox/memory-consolidation-YYYY-MM-DD.md (Inbox-Queue, Kais approves manuell — kein Auto-Write) 4. Optional: Telegram-Notification an Kais mit Anzahl Vorschlaege Inspiriert von: - Letta sleep-time compute - AutoSkill (arXiv 2603.01145) - EvoSkill (arXiv 2603.02766) Env vars: ARIA_BRAIN (default $HOME/aria/brain), TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID """ import os import sys import json import urllib.request import urllib.parse from collections import Counter from datetime import datetime, timedelta from pathlib import Path import re # ─── Config ────────────────────────────────────────── ARIA_BRAIN = Path(os.environ.get("ARIA_BRAIN", os.path.expanduser("~/aria/brain"))) INBOX = ARIA_BRAIN / "00-Inbox" TODAY = datetime.now().strftime("%Y-%m-%d") OUT_FILE = INBOX / f"memory-consolidation-{TODAY}.md" ENV_FILES = ["/root/.env", os.path.expanduser("~/aria/.env"), os.path.expanduser("~/.env")] def env(key): for f in ENV_FILES: if not os.path.exists(f): continue for line in open(f): if line.startswith(f"{key}="): return line.split("=", 1)[1].strip().strip('"\'') return os.environ.get(key, "") # Auth via Supabase REST + service_role (Konvention der AKP/chat-logger-Scripts). # Frueher: Management-API + SUPABASE_ACCESS_TOKEN (existierte nie in env -> Script lief nie). Fix 2026-06-03. SUPA_URL = env("ARIA_SUPABASE_URL") or env("SUPABASE_URL") SUPA_KEY = env("SUPABASE_SERVICE_ROLE_KEY") TG_TOKEN = env("TELEGRAM_BOT_TOKEN") TG_CHAT = env("TELEGRAM_CHAT_ID") # ─── Supabase fetch last 24h chat log (REST/PostgREST) ──────────────── def fetch_recent_chats(hours=24): if not (SUPA_URL and SUPA_KEY): print("ERR: no SUPABASE_SERVICE_ROLE_KEY / ARIA_SUPABASE_URL", file=sys.stderr) return [] since = (datetime.utcnow() - timedelta(hours=hours)).strftime("%Y-%m-%dT%H:%M:%S") params = urllib.parse.urlencode({ "select": "direction,message_text,created_at", "created_at": f"gt.{since}", "order": "created_at.asc", "limit": "1000", }) req = urllib.request.Request( f"{SUPA_URL}/rest/v1/aria_chat_log?{params}", headers={ "apikey": SUPA_KEY, "Authorization": f"Bearer {SUPA_KEY}", "Accept": "application/json", }, ) try: return json.loads(urllib.request.urlopen(req, timeout=30).read()) except Exception as e: print(f"ERR: Supabase fetch failed: {e}", file=sys.stderr) return [] # ─── Pattern Detection ──────────────────────────────── def normalize(text): """Normalize text fuer Aehnlichkeitsvergleich.""" text = text.lower().strip() text = re.sub(r'[^\w\s]', '', text) text = re.sub(r'\s+', ' ', text) return text[:80] def detect_recurring_questions(messages): """User-Messages die 3+ mal in aehnlicher Form vorkommen.""" user_msgs = [m for m in messages if m.get("direction") == "in"] normalized = Counter(normalize(m["message_text"]) for m in user_msgs if len(m.get("message_text", "")) > 5) recurring = [(text, count) for text, count in normalized.items() if count >= 3] return recurring[:10] def detect_corrections(messages): """User-Korrekturen: 'nein', 'falsch', 'anders', 'stop', 'nicht so'.""" correction_signals = ["nein,", "falsch", "anders", "nicht so", "stop ", "anders machen", "korrigier", "verbess"] user_msgs = [m for m in messages if m.get("direction") == "in"] corrections = [] for m in user_msgs: text = m.get("message_text", "").lower() if any(sig in text for sig in correction_signals): corrections.append({ "time": m.get("created_at", ""), "text": m.get("message_text", "")[:200], }) return corrections[:15] def detect_frustration(messages): """User-Frustration-Signale.""" signals = ["ich verarsch", "wieso", "warum", "ist nicht", "geht nicht", "tot", "kaputt", "verbug", "blöd"] user_msgs = [m for m in messages if m.get("direction") == "in"] frustration = [] for m in user_msgs: text = m.get("message_text", "").lower() if any(sig in text for sig in signals): frustration.append({ "time": m.get("created_at", ""), "text": m.get("message_text", "")[:200], }) return frustration[:15] # ─── Build Inbox Vorschlag ──────────────────────────── def build_proposal(messages): if not messages: return None recurring = detect_recurring_questions(messages) corrections = detect_corrections(messages) frustration = detect_frustration(messages) in_msgs = sum(1 for m in messages if m.get("direction") == "in") out_msgs = sum(1 for m in messages if m.get("direction") == "out") md = f"""--- title: Memory-Consolidation Vorschlag {TODAY} type: inbox tags: [memory, consolidation, sleep-time, auto] date: {TODAY} status: pending --- # Memory-Consolidation — {TODAY} > Auto-generiert via aria-memory-consolidate-night.py (Sleep-Time-Pattern, Letta/AutoSkill-inspired) > Letzte 24h aria_chat_log analysiert. Aria approved manuell — kein Auto-Write. ## Stats - Letzte 24h: **{len(messages)} Messages** ({in_msgs} eingehend, {out_msgs} ausgehend) - Avg Response-Time: noch nicht berechnet - Pattern-Detector: 3 Categories aktiv (Recurring, Corrections, Frustration) """ if recurring: md += "## 🔁 Wiederkehrende Fragen (3+x similar)\n\n" md += "_Kais fragt mehrfach Aehnliches → Memory-Eintrag oder Skill-Kandidat?_\n\n" for text, count in recurring: md += f"- **{count}x** — `{text}`\n" md += "\n**Aktion:** Pruefen ob ein Memory/Skill diese Fragen kuenftig autom. beantworten kann.\n\n" if corrections: md += "## ⚠️ Korrektur-Signale\n\n" md += "_User korrigiert Aria → CORRECTIONS-Eintrag-Kandidat?_\n\n" for c in corrections: md += f"- `{c['time']}` — _{c['text']}_\n" md += "\n**Aktion:** Zu jeder Korrektur pruefen ob sie ein neues feedback_*.md / CORRECTIONS-Pattern verdient.\n\n" if frustration: md += "## 😤 Frustrations-Signale\n\n" md += "_User-Frust → SELF-IMPROVEMENT-Kandidat oder System-Bug?_\n\n" for f in frustration: md += f"- `{f['time']}` — _{f['text']}_\n" md += "\n**Aktion:** Wiederholtes? → SELF-IMPROVEMENT-Regel. Einmaliger Bug? → fix.\n\n" if not (recurring or corrections or frustration): md += "## ✅ Keine Patterns gefunden\n\nLetzte 24h waren glatt. Keine Aktion noetig.\n\n" md += f"""--- ## Naechste Schritte (Kais) - [ ] Vorschlaege durchgehen - [ ] Approved → in entsprechendes Brain-File mergen (CORRECTIONS, SELF-IMPROVEMENT, feedback_*) - [ ] Diese Datei nach Approval verschieben nach 07-Memory-Consolidated/{TODAY}.md """ return md # ─── Notify Telegram ────────────────────────────────── def notify_telegram(summary): if not (TG_TOKEN and TG_CHAT): return try: payload = json.dumps({"chat_id": TG_CHAT, "text": summary}).encode() urllib.request.urlopen( urllib.request.Request( f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage", data=payload, headers={"Content-Type": "application/json"}, ), timeout=10, ) except Exception as e: print(f"Telegram notify failed: {e}", file=sys.stderr) # ─── Main ──────────────────────────────────────────── def main(): INBOX.mkdir(parents=True, exist_ok=True) print(f"Sleep-Time Memory Consolidation — {datetime.now().isoformat()}") messages = fetch_recent_chats(24) print(f"Fetched {len(messages)} messages") proposal = build_proposal(messages) if not proposal: print("No messages, no proposal") return OUT_FILE.write_text(proposal, encoding="utf-8") print(f"Wrote {OUT_FILE} ({OUT_FILE.stat().st_size} bytes)") # Telegram Summary line_count = proposal.count("\n- ") if line_count > 0: msg = ( f"🌙 Memory-Consolidation — {TODAY}\n\n" f"{line_count} Vorschlaege in 00-Inbox/memory-consolidation-{TODAY}.md.\n" f"Approval-Pflicht (kein Auto-Write)." ) notify_telegram(msg) if __name__ == "__main__": _log.event("script_start") main()