ARIA-AGENT/aria-brain/context_model.py

"""Welches Modell gilt in welchem Chat.

Drei Ebenen, von unten nach oben:

  1. Einstellung — `brainModel` aus runtime.json (Diagnostic → Sprachmodell).
     Gilt fuer alles, was keine eigene Wahl hat.
  2. Pro Chat — ein Override je Kontext ("" = Hauptchat, sonst project_id).
     Damit kann ein Projekt bewusst auf einem anderen Modell laufen, ohne das
     Hauptmodell fuer alles umzustellen.
  3. Ausweich — Anthropics Sicherheitsfilter hat das gewaehlte Modell an einem
     Turn blockiert, der noch im Fenster liegt (safeguard.py). Das sticht alles,
     ist aber nur so lange aktiv, wie der Turn im Fenster steht.

Warum ueberhaupt ein eigener Speicher und kein Feld im Projekt-Manifest: der
Hauptchat IST kein Projekt. Ein Manifest-Feld haette ihn aussen vor gelassen,
und genau dort faellt eine Modellwahl am ehesten an.
"""

from __future__ import annotations

import json
import logging
import os
import threading
from pathlib import Path
from typing import Tuple

logger = logging.getLogger(__name__)

STORE_FILE = Path(os.environ.get("CONTEXT_MODELS_FILE",
                                 "/shared/config/context_models.json"))

# Begruendungen, wie sie die Oberflaechen anzeigen. Bewusst kurze Schluessel —
# App und Diagnostic uebersetzen sie selbst, damit die Texte dort hingehoeren,
# wo sie gelesen werden.
REASON_SETTING = "einstellung"
REASON_PER_CHAT = "pro-chat"
REASON_FALLBACK = "ausweich"

_lock = threading.Lock()


def _load() -> dict:
    try:
        if STORE_FILE.exists():
            data = json.loads(STORE_FILE.read_text(encoding="utf-8")) or {}
            if isinstance(data, dict):
                # Nur String→String; alles andere waere ein Schreibfehler von
                # aussen und wuerde sonst still als Modellname durchgehen.
                return {str(k): str(v) for k, v in data.items() if isinstance(v, str)}
    except Exception as exc:
        logger.warning("[context-model] lesen fehlgeschlagen: %s", exc)
    return {}


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


def all_overrides() -> dict:
    return _load()


def get(context_id: str) -> str:
    """Der Override dieses Kontexts, oder "" wenn keiner gesetzt ist."""
    return _load().get(context_id or "", "")


def set(context_id: str, model: str) -> str:
    """Override setzen. Leerer Modellname loescht ihn (= wieder Einstellung)."""
    cid = context_id or ""
    m = (model or "").strip()
    with _lock:
        data = _load()
        if m:
            data[cid] = m
        else:
            data.pop(cid, None)
        _save(data)
    logger.info("[context-model] %s → %s", cid or "(Hauptchat)", m or "(Einstellung)")
    return m


def resolve(context_id: str, default_model: str) -> Tuple[str, str]:
    """(Modell, Begruendung) ohne Sicherheitsfilter — den kennt nur der Agent."""
    ov = get(context_id)
    if ov:
        return ov, REASON_PER_CHAT
    return default_model, REASON_SETTING