ARIA-AGENT/aria-brain/prompts.py

"""
System-Prompt-Bau aus Memory-Punkten.

Strategie:
  1. Alle pinned Punkte (Hot Memory) — gruppiert nach Type — in den
     System-Prompt schreiben. IMMER drin.
  2. Top-K semantisch aehnliche Punkte (Cold Memory) zur aktuellen
     User-Nachricht — als "Moeglicherweise relevant" eingehaengt.
  3. Aktive Skills als kompakte Liste (nur Name + Description) — damit
     ARIA weiss was sie hat.

Phase B Punkt 1: nur Hot-Memory-Bau, Skills + Cold-Search kommen
mit dem Conversation-Loop in spaeteren Phasen.
"""

from __future__ import annotations

from datetime import datetime, timezone, timedelta
from typing import List

from memory import MemoryPoint


def build_time_section() -> str:
    """Aktueller Zeitstempel — damit ARIA Timer korrekt anlegen kann
    und Watcher-Conditions mit hour_of_day etc. einordenbar bleiben."""
    now_utc = datetime.now(timezone.utc)
    # Europa/Berlin: Sommerzeit CEST = UTC+2, Winterzeit CET = UTC+1.
    # Wir nehmen den simplen Fall (kein zoneinfo-Import noetig im Brain-Image):
    # Stefans VM laeuft auf UTC, die Bridge in der Wohnung — Anzeige reicht.
    local_offset_h = 2 if 3 <= now_utc.month <= 10 else 1
    local = now_utc + timedelta(hours=local_offset_h)
    lines = [
        "## Aktuelle Zeit",
        f"- UTC: {now_utc.isoformat(timespec='seconds')}",
        f"- Lokal (Europa/Berlin, UTC+{local_offset_h}): "
        f"{local.strftime('%Y-%m-%d %H:%M:%S')} ({local.strftime('%A')})",
        "",
        "Nutze das fuer Trigger-Timestamps und um Watcher-Conditions wie "
        "`hour_of_day == 8` einzuordnen. Fuer relative Angaben "
        "('in 10min', 'in 2 Stunden') nutze beim `trigger_timer` den "
        "`in_seconds`-Parameter — Server rechnet dann selbst.",
    ]
    return "\n".join(lines)

TYPE_HEADINGS = {
    "identity": "## Wer du bist",
    "rule": "## Sicherheitsregeln & Prinzipien",
    "preference": "## Benutzer-Praeferenzen",
    "tool": "## Tool-Freigaben",
    "skill": "## Deine Skills",
}


def build_hot_memory_section(pinned: List[MemoryPoint]) -> str:
    """Baue den 'IMMER-im-Prompt'-Block aus pinned Punkten."""
    grouped: dict[str, List[MemoryPoint]] = {}
    for p in pinned:
        grouped.setdefault(p.type, []).append(p)

    parts: List[str] = []
    # Sortier-Reihenfolge: identity → rule → preference → tool → skill → Rest
    order = ["identity", "rule", "preference", "tool", "skill"]
    for t in order:
        items = grouped.pop(t, [])
        if not items:
            continue
        parts.append(TYPE_HEADINGS.get(t, f"## {t}"))
        for p in items:
            parts.append(f"### {p.title}")
            parts.append(p.content.strip())
            parts.append("")

    # uebrige Types (falls jemand was anderes als pinned markiert)
    for t, items in grouped.items():
        parts.append(f"## {t}")
        for p in items:
            parts.append(f"### {p.title}")
            parts.append(p.content.strip())
            parts.append("")

    return "\n".join(parts).strip()


def build_cold_memory_section(matches: List[MemoryPoint]) -> str:
    """Baue 'Moeglicherweise relevant'-Block aus Search-Treffern."""
    if not matches:
        return ""
    lines = ["## Moeglicherweise relevant (aus Gedaechtnis)"]
    for p in matches:
        score = f" [score={p.score:.2f}]" if p.score is not None else ""
        lines.append(f"- **{p.title}**{score}")
        lines.append(f"  {p.content.strip()}")
    return "\n".join(lines)


def build_skills_section(skills: List[dict]) -> str:
    """Listet alle Skills (aktiv + deaktiviert) damit ARIA weiss was es gibt
    und keine doppelt baut. Plus klare Schwelle wann ein Skill sich lohnt."""
    lines = ["## Deine Skills"]
    if skills:
        for s in skills:
            active = s.get("active", True)
            marker = "" if active else " [DEAKTIVIERT — kann nicht aufgerufen werden]"
            lines.append(f"- **{s.get('name', '?')}**{marker} — {s.get('description', '(ohne Beschreibung)')}")
        lines.append("")
        lines.append("Wenn ein vorhandener Skill zur Aufgabe passt: nutze ihn via Tool-Call.")
    else:
        lines.append("(noch keine Skills vorhanden)")

    lines.append("")
    lines.append("### Wann lohnt sich ein neuer Skill?")
    lines.append("")
    lines.append("**Skills sind IMMER Python** — eigene venv pro Skill mit den noetigen "
                 "pip-Paketen. Kein apt im Skill, kein systemweiter Install. Python deckt "
                 "in der Regel alles ab (yt-dlp, requests, pypdf, pillow, openpyxl, "
                 "static-ffmpeg, beautifulsoup4, …). Falls etwas WIRKLICH nur via apt geht: "
                 "Stefan fragen ob es ins Brain-Dockerfile soll.")
    lines.append("")
    lines.append("**Harte Regel — IMMER Skill anlegen wenn:** die Loesung erfordert eine "
                 "pip-Library. Begruendung: Brain-Container hat keinen persistenten State "
                 "ausser /data/skills/. Ohne Skill wuerde der Install bei jedem "
                 "Container-Restart wiederholt.")
    lines.append("")
    lines.append("**Sonst — Skill nur wenn alle vier zutreffen:**")
    lines.append("")
    lines.append("1. **Wiederkehrend** — die Aufgabe wird realistisch nochmal gestellt. "
                 "Einmal-Faelle (\"wie spaet ist es jetzt\") kein Skill.")
    lines.append("2. **Nicht-trivial** — mehrere Schritte. Ein einzelner Shell-Befehl "
                 "(`date`, `hostname`, `ls`) ist KEIN Skill — das macht Bash direkt.")
    lines.append("3. **Parametrisierbar** — der Skill nimmt Eingaben (URL, Datei, Suchbegriff) "
                 "und gibt ein nuetzliches Ergebnis zurueck.")
    lines.append("4. **Wiederverwendbar als ganzes** — Stefan wuerde es zukuenftig per Name "
                 "ansprechen (\"mach mir den YouTube zu MP3\") statt jedes Mal zu erklaeren.")
    lines.append("")
    lines.append("Wenn nichts installiert werden muss UND nicht alle vier zutreffen: einfach "
                 "die Aufgabe loesen ohne Skill anzulegen. Stefan kann jederzeit sagen "
                 "'bau daraus einen Skill'.")
    return "\n".join(lines)


def build_triggers_section(
    triggers: List[dict],
    condition_vars: List[dict],
    condition_funcs: List[dict] | None = None,
) -> str:
    """Triggers (passive Aufweck-Quellen) + verfuegbare Condition-Variablen + Funktionen."""
    lines = ["## Trigger (passive Aufweck-Quellen)"]
    lines.append("")
    lines.append("Trigger sind ANDERS als Skills: das System ruft DICH wenn ein Event passiert. "
                 "Du legst sie an wenn Stefan sagt 'erinner mich an X' oder 'sag bescheid wenn Y'.")
    lines.append("")
    if triggers:
        lines.append("### Aktuelle Trigger")
        for t in triggers:
            active = t.get("active", True)
            mark = "" if active else " [INAKTIV]"
            if t["type"] == "timer":
                lines.append(f"- **{t['name']}**{mark} (timer) feuert {t.get('fires_at')}: \"{t.get('message','')[:80]}\"")
            elif t["type"] == "watcher":
                lines.append(f"- **{t['name']}**{mark} (watcher) cond=`{t.get('condition')}`: \"{t.get('message','')[:80]}\"")
        lines.append("")
    lines.append("### Verfuegbare Condition-Variablen (fuer Watcher)")
    for v in condition_vars:
        lines.append(f"- `{v['name']}` ({v['type']}) — {v['desc']}")
    if condition_funcs:
        lines.append("")
        lines.append("### Verfuegbare Funktionen in Conditions")
        for fn in condition_funcs:
            lines.append(f"- `{fn['signature']}` — {fn['desc']}")
    lines.append("")
    lines.append("Operatoren in Conditions: `<` `>` `<=` `>=` `==` `!=` `and` `or` `not`. "
                 "Beispiele: `disk_free_gb < 5 and hour_of_day >= 8`, "
                 "`day_of_week == \"mon\"`, `near(53.123, 7.456, 500)`. "
                 "Funktionen nur mit Konstanten als Argumenten (keine Variablen, "
                 "keine geschachtelten Funktionen).")
    lines.append("")
    lines.append("### Wann welcher Typ?")
    lines.append("- **Timer** fuer einmalige Erinnerungen mit konkreter Zeit ('in 10min', 'um 14:30').")
    lines.append("- **Watcher** fuer 'wenn X passiert' (Disk voll, bestimmte Tageszeit, GPS-Naehe).")
    lines.append("- ARIA legt Trigger NUR auf Stefan-Wunsch an, nicht eigenmaechtig.")
    lines.append("")
    lines.append("### GPS-Watcher mit near()")
    lines.append(
        "Wenn du einen Watcher mit `near()` anlegst: die App sendet GPS-Position "
        "nur kontinuierlich wenn Tracking AN ist (Default: AUS, Akku-Schutz). "
        "Rufe dafuer `request_location_tracking(on=true, reason=\"...\")` auf "
        "bevor oder gleich nach dem trigger_watcher. Sonst hat current_lat/lon "
        "veraltete Werte und der Watcher feuert nie. "
        "Beim Loeschen des letzten GPS-Watchers (trigger_cancel) wieder "
        "`request_location_tracking(on=false)` aufrufen.")
    return "\n".join(lines)


def build_system_prompt(
    pinned: List[MemoryPoint],
    cold: List[MemoryPoint] | None = None,
    skills: List[dict] | None = None,
    triggers: List[dict] | None = None,
    condition_vars: List[dict] | None = None,
    condition_funcs: List[dict] | None = None,
) -> str:
    """Kompletter System-Prompt: Hot + Cold + Skills + Triggers."""
    parts = [build_hot_memory_section(pinned), "", build_time_section()]
    if skills:
        parts.append("")
        parts.append(build_skills_section(skills))
    if condition_vars:
        parts.append("")
        parts.append(build_triggers_section(triggers or [], condition_vars, condition_funcs))
    if cold:
        parts.append("")
        parts.append(build_cold_memory_section(cold))
    return "\n".join(parts).strip()