ARIA-AGENT/aria-brain/safeguard.py

"""Anthropics Sicherheitsfilter — erkennen, ausweichen, Verlauf heil halten.

Der Filter ("... safeguards flagged this message", `Details: [cyber]`) meldet
sich NICHT als HTTP-Fehler. Die CLI liefert ein voellig normales Ergebnis mit
is_error=true, und der Fehlertext steht im `content` — also genau dort, wo
sonst ARIAs Antwort steht. Ungeprueft passiert deshalb Folgendes:

  1. Der Fehlertext landet per conversation.add("assistant", ...) im Verlauf.
  2. Jeder Folge-Turn schickt den Verlauf samt der ausloesenden Nachricht wieder
     mit — der Filter greift erneut.
  3. Das Projekt ist tot: jede weitere Frage scheitert an derselben Stelle.

Dagegen zwei Rettungsstufen:

  Stufe 1 — Modellwechsel. Der Filter ist modellspezifisch ("Opus 5's
  safeguards ..."), Anthropic empfiehlt selbst "change your model". Derselbe
  Turn wird mit dem Ausweich-Modell neu gefragt. Der ausloesende Turn wird
  vermerkt, damit die naechsten Turns des Projekts nicht erst wieder gegen die
  Wand laufen. Der Vermerk verfaellt von selbst, sobald der Turn aus dem
  Rolling Window gerollt ist — nichts muss aufgeraeumt werden.

  Stufe 2 — Quarantaene. Blockt auch das Ausweich-Modell, liegt es am Verlauf
  selbst. Dann wird der ausloesende Turn aus dem Prompt-Fenster genommen.
  NICHT geloescht: er bleibt in conversation.jsonl, bleibt in der Diagnostic
  sichtbar und ist mit einem Klick zurueckholbar. Das Projekt ist damit im
  naechsten Turn wieder benutzbar.

Der Zustand liegt in /data/safeguard.json — bewusst NICHT in
conversation.jsonl: ein Markieren wuerde dort ein Rewrite erzwingen, und das
schreibt nur das In-Memory-Window zurueck (alles schon Destillierte waere weg).
"""

from __future__ import annotations

import json
import logging
import os
import re
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Optional

logger = logging.getLogger(__name__)

STATE_FILE = Path(os.environ.get("SAFEGUARD_STATE_FILE", "/data/safeguard.json"))
RUNTIME_CONFIG_FILE = Path(os.environ.get("RUNTIME_CONFIG_FILE", "/shared/config/runtime.json"))

# Wie viele Vorfaelle wir aufbewahren. Reicht fuer die Diagnostic-Anzeige und
# haelt die Datei klein.
MAX_INCIDENTS = 200

_lock = threading.Lock()


class Blocked(Exception):
    """Auch das Ausweich-Modell wurde blockiert.

    Dann hilft kein Modellwechsel mehr — es liegt am Verlauf selbst. Der Agent
    faengt das und geht auf Stufe 2 (Quarantaene).
    """

    def __init__(self, models, text: str = ""):
        self.models = [m for m in models if m]
        self.text = text or ""
        super().__init__("Sicherheitsfilter blockiert: " + ", ".join(self.models))

# Textmarker des Filters. Englisch und stabil — der Text kommt von Anthropic,
# nicht von uns. Mehrere, weil die Formulierung sich aendern kann.
_MARKERS = (
    "safeguards flagged this message",
    "anthropic.com/legal/aup",
    "can't respond to this message with",
    "cannot respond to this message with",
    "try rephrasing the request in a new session",
)

_DETAIL_RE = re.compile(r"details:\s*`?\[([a-z0-9_-]+)\]`?", re.IGNORECASE)


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def norm_model(model: str) -> str:
    """Modellnamen vergleichbar machen (Gross/Klein, CLI-Prefix, Leerzeichen)."""
    m = (model or "").strip().lower()
    if m.startswith("claude-code-cli/"):
        m = m[len("claude-code-cli/"):]
    return m


# ── Erkennung ───────────────────────────────────────────────────────

def looks_like_block(text: str, is_error: bool = False) -> bool:
    """True, wenn dieser Antworttext der Sicherheitsfilter ist.

    Zwei Wege, absichtlich unterschiedlich streng:

    * is_error=true (die CLI sagt selbst "das war ein Fehler") — ein Marker
      genuegt.
    * Ohne is_error muss der Text mit "API Error" ANFANGEN. Sonst wuerde die
      Erkennung auch dann zuschlagen, wenn ARIA ueber den Filter *spricht*
      ("der Fehler hiess: safeguards flagged ...") — eine Antwort ueber den
      Fehler ist kein Fehler.
    """
    t = (text or "").strip()
    if not t:
        return False
    low = t.lower()
    hits = sum(1 for m in _MARKERS if m in low)
    if hits == 0:
        return False
    if is_error:
        return True
    return low.startswith("api error") and len(t) <= 2000


def detail_code(text: str) -> str:
    """Der Klassifikator-Code aus "Details: `[cyber]`" — nur fuer die Anzeige."""
    m = _DETAIL_RE.search(text or "")
    return m.group(1) if m else ""


# ── Ausweich-Modell ─────────────────────────────────────────────────

def _runtime_cfg() -> dict:
    try:
        if RUNTIME_CONFIG_FILE.exists():
            return json.loads(RUNTIME_CONFIG_FILE.read_text(encoding="utf-8")) or {}
    except Exception as exc:
        logger.warning("[safeguard] runtime.json lesen fehlgeschlagen: %s", exc)
    return {}


def fallback_for(model: str) -> str:
    """Welches Modell beim naechsten Versuch.

    Vorrang hat `brainFallbackModel` aus runtime.json (Diagnostic). Zeigt das
    auf dasselbe Modell, ist es wertlos — dann faellt es auf die automatische
    Wahl zurueck: ein ANDERES Tier, denn der Filter haengt am Modell.

    Pro Aufruf gelesen, nicht beim Import: eine Aenderung in der Diagnostic
    greift so ohne Brain-Neustart.
    """
    want = str(_runtime_cfg().get("brainFallbackModel") or "").strip()
    primary = norm_model(model)
    if want and norm_model(want) != primary:
        return want
    if "sonnet" in primary:
        return "opus"
    # Opus, Fable, Haiku und alles Unbekannte → Sonnet. Sonnet ist das
    # breiteste Tier im Max-Abo und ein anderer Filter-Kontext als Opus.
    return "sonnet"


# ── Zustand ─────────────────────────────────────────────────────────

def _load() -> dict:
    try:
        if STATE_FILE.exists():
            data = json.loads(STATE_FILE.read_text(encoding="utf-8")) or {}
            if isinstance(data, dict):
                data.setdefault("incidents", [])
                data.setdefault("quarantine", [])
                return data
    except Exception as exc:
        logger.warning("[safeguard] Zustand lesen fehlgeschlagen: %s", exc)
    return {"incidents": [], "quarantine": []}


def _save(data: dict) -> None:
    try:
        STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
        tmp = STATE_FILE.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp.replace(STATE_FILE)
    except Exception as exc:
        logger.warning("[safeguard] Zustand schreiben fehlgeschlagen: %s", exc)


def record_block(project_id: str, turn_ts: str, model: str, fallback: str,
                 text: str = "", recovered: bool = False) -> None:
    """Einen Vorfall vermerken. turn_ts = der User-Turn, der mitlief."""
    with _lock:
        data = _load()
        data["incidents"].append({
            "at": _now(),
            "project_id": project_id or "",
            "turn_ts": turn_ts or "",
            "model": model or "",
            "fallback": fallback or "",
            "recovered": bool(recovered),
            "code": detail_code(text),
            "snippet": (text or "").strip()[:300],
        })
        data["incidents"] = data["incidents"][-MAX_INCIDENTS:]
        _save(data)


def mark_recovered(turn_ts: str, model: str, fallback_used: str) -> None:
    """Nachtragen, dass der Ausweichversuch geklappt hat (nur Anzeige)."""
    with _lock:
        data = _load()
        for inc in reversed(data["incidents"]):
            if inc.get("turn_ts") == turn_ts and norm_model(inc.get("model", "")) == norm_model(model):
                inc["recovered"] = True
                inc["fallback"] = fallback_used
                break
        _save(data)


def blocked_models(project_id: str, turn_timestamps: Iterable[str]) -> set:
    """Modelle, die an den Turns DIESES Fensters schon gescheitert sind.

    Der Abgleich laeuft ueber die Turn-Zeitstempel: rollt der ausloesende Turn
    aus dem Fenster, taucht er hier nicht mehr auf und das Projekt darf wieder
    aufs Hauptmodell. Kein Aufraeumen, kein Ablaufdatum von Hand.
    """
    want = set(turn_timestamps or ())
    if not want:
        return set()
    pid = project_id or ""
    out = set()
    for inc in _load()["incidents"]:
        if (inc.get("project_id") or "") != pid:
            continue
        ts = inc.get("turn_ts") or ""
        if ts and ts in want:
            m = norm_model(inc.get("model", ""))
            if m:
                out.add(m)
    return out


# ── Quarantaene ─────────────────────────────────────────────────────

def quarantine(turn_ts: str, project_id: str = "", reason: str = "") -> None:
    if not turn_ts:
        return
    with _lock:
        data = _load()
        if any(q.get("turn_ts") == turn_ts for q in data["quarantine"]):
            return
        data["quarantine"].append({
            "turn_ts": turn_ts, "project_id": project_id or "",
            "at": _now(), "reason": reason or "",
        })
        _save(data)
    logger.warning("[safeguard] Turn %s in Quarantaene (Projekt %r)", turn_ts, project_id or "(main)")


def unquarantine(turn_ts: str) -> bool:
    with _lock:
        data = _load()
        before = len(data["quarantine"])
        data["quarantine"] = [q for q in data["quarantine"] if q.get("turn_ts") != turn_ts]
        changed = len(data["quarantine"]) != before
        if changed:
            _save(data)
    if changed:
        logger.info("[safeguard] Quarantaene aufgehoben: %s", turn_ts)
    return changed


def quarantined_ts() -> set:
    return {q.get("turn_ts") for q in _load()["quarantine"] if q.get("turn_ts")}


def filter_window(turns: list) -> list:
    """Quarantaenierte Turns aus dem Prompt-Fenster nehmen.

    Greift NUR auf dem Weg zum Modell. Gespeichert bleibt alles — die
    Diagnostic und jeder Export sehen den Turn weiter.
    """
    q = quarantined_ts()
    if not q:
        return turns
    return [t for t in turns if getattr(t, "ts", "") not in q]


# ── Diagnostic ──────────────────────────────────────────────────────

def status(limit: int = 25) -> dict:
    data = _load()
    inc = data["incidents"][-limit:][::-1]
    return {
        "incidents": inc,
        "total": len(data["incidents"]),
        "quarantine": data["quarantine"],
        "fallback_model": str(_runtime_cfg().get("brainFallbackModel") or "").strip(),
        "state_file": str(STATE_FILE),
    }


def clear(project_id: Optional[str] = None) -> int:
    """Vorfaelle loeschen (Anzeige + Taint). Quarantaene bleibt — die wird
    einzeln aufgehoben, damit nicht ein Klick alle Blockaden zurueckholt."""
    with _lock:
        data = _load()
        before = len(data["incidents"])
        if project_id is None:
            data["incidents"] = []
        else:
            pid = project_id or ""
            data["incidents"] = [i for i in data["incidents"]
                                 if (i.get("project_id") or "") != pid]
        removed = before - len(data["incidents"])
        _save(data)
    return removed