ARIA-AGENT/aria-brain/watcher.py

"""
Built-in Condition-Variablen + sicherer Mini-Parser fuer Watcher-Triggers.

Erlaubte Variablen kommen aus diesem Modul. Condition-Ausdruck ist ein
sicheres Subset von Python (kein eval, kein exec): nur Vergleiche und
Boolean-Operatoren, nur die hier deklarierten Variablen, nur Zahlen +
String-Literale als rechte Seite.

Beispiele:
  disk_free_gb < 5
  hour_of_day == 8 and day_of_week == "mon"
  rvs_connected == False
  (disk_free_pct < 10 and uptime_sec > 3600)
"""

from __future__ import annotations

import ast
import logging
import os
import shutil
import time
from datetime import datetime
from pathlib import Path
from typing import Any

logger = logging.getLogger(__name__)


# ─── Variablen-Quellen ──────────────────────────────────────────────

def _disk_stats() -> tuple[float, float]:
    """Returns (free_gb, free_pct). Schaut /shared (geteiltes Volume) — sonst /."""
    target = "/shared" if os.path.exists("/shared") else "/"
    try:
        st = shutil.disk_usage(target)
        free_gb = st.free / (1024 ** 3)
        free_pct = 100.0 * st.free / st.total if st.total else 0.0
        return free_gb, free_pct
    except Exception as e:
        logger.warning("disk_usage: %s", e)
        return 0.0, 0.0


def _uptime_sec() -> int:
    try:
        with open("/proc/uptime", "r") as f:
            return int(float(f.read().split()[0]))
    except Exception:
        return 0


def _rvs_connected() -> bool:
    """Liest /shared/config/runtime.json oder ein Bridge-State-File.
    Aktuell: wir koennen das nicht zuverlaessig aus dem Brain-Container
    bestimmen — gibt False als sicheren Default zurueck.
    Spaeter: Bridge schreibt einen Heartbeat-File den wir hier lesen."""
    return False


_DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"]


def collect_variables() -> dict[str, Any]:
    """Liefert aktuellen Snapshot aller Built-in-Variablen."""
    free_gb, free_pct = _disk_stats()
    now = datetime.now()
    # Memory-Count aus der Vector-DB (importiert lazy um zirkulaere Imports
    # zu vermeiden — beim Modul-Load gibt's noch keinen Store)
    memory_count = 0
    try:
        from main import store  # type: ignore
        s = store()
        memory_count = s.count()
    except Exception:
        pass

    return {
        "disk_free_gb":   round(free_gb, 2),
        "disk_free_pct":  round(free_pct, 1),
        "uptime_sec":     _uptime_sec(),
        "hour_of_day":    now.hour,
        "day_of_week":    _DAYS[now.weekday()],
        "rvs_connected":  _rvs_connected(),
        "memory_count":   memory_count,
    }


def describe_variables() -> list[dict]:
    """Liste der verfuegbaren Variablen + Beschreibung — fuer System-Prompt + UI."""
    return [
        {"name": "disk_free_gb",   "type": "number", "desc": "freier Plattenplatz in GB (auf /shared)"},
        {"name": "disk_free_pct",  "type": "number", "desc": "freier Plattenplatz in Prozent"},
        {"name": "uptime_sec",     "type": "number", "desc": "Sekunden seit Brain-Start"},
        {"name": "hour_of_day",    "type": "number", "desc": "0..23, lokale Zeit"},
        {"name": "day_of_week",    "type": "string", "desc": "mon|tue|wed|thu|fri|sat|sun"},
        {"name": "rvs_connected",  "type": "bool",   "desc": "True wenn RVS-Verbindung steht"},
        {"name": "memory_count",   "type": "number", "desc": "Anzahl Memories in der Vector-DB"},
    ]


# ─── Sicherer Condition-Parser ──────────────────────────────────────

_ALLOWED_NODES = (
    ast.Expression, ast.BoolOp, ast.UnaryOp, ast.Compare,
    ast.Name, ast.Constant, ast.Load,
    ast.And, ast.Or, ast.Not,
    ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.Gt, ast.GtE,
)


def parse_condition(expr: str) -> ast.Expression:
    """Parst einen Condition-Ausdruck und validiert ihn gegen das Safe-Subset.
    Wirft ValueError bei verbotenen Konstrukten."""
    expr = (expr or "").strip()
    if not expr:
        raise ValueError("Leere Condition")
    if len(expr) > 500:
        raise ValueError("Condition zu lang (>500 Zeichen)")
    try:
        tree = ast.parse(expr, mode="eval")
    except SyntaxError as e:
        raise ValueError(f"Condition Syntax-Fehler: {e}")
    # Whitelist-Walk
    allowed_names = {v["name"] for v in describe_variables()}
    for node in ast.walk(tree):
        if not isinstance(node, _ALLOWED_NODES):
            raise ValueError(f"Verbotener Ausdruck: {type(node).__name__}")
        if isinstance(node, ast.Name):
            if node.id not in allowed_names and node.id not in ("True", "False"):
                raise ValueError(f"Unbekannte Variable: {node.id}")
        if isinstance(node, ast.Constant):
            if not isinstance(node.value, (int, float, str, bool)) and node.value is not None:
                raise ValueError(f"Verbotener Konstant-Typ: {type(node.value).__name__}")
    return tree


def evaluate(expr: str, variables: dict[str, Any] | None = None) -> bool:
    """Evaluiert die Condition gegen die aktuellen Variablen.
    Returns bool. Bei Fehler in Variablen → False (defensiv)."""
    tree = parse_condition(expr)
    vars_ = variables if variables is not None else collect_variables()
    code = compile(tree, "<condition>", "eval")
    # Globals leer, locals nur die erlaubten Variablen → kein Builtin-Zugriff
    try:
        result = eval(code, {"__builtins__": {}}, vars_)
    except Exception as e:
        logger.warning("Condition '%s' eval-Fehler: %s", expr, e)
        return False
    return bool(result)