mission-control-v2/backend/services/agent_aktivitaet.py

"""Werkzeug-Verlauf des Agenten (v3-Umbau P6).

WARUM ES DAS GIBT: Lucys Arbeit war eine Blackbox mit Statuswort. Der Blueprint sah
dafür eine „Live Agent Matrix" mit Denkstrom vor (§4.4) — die ist so nicht baubar: Die
Denkschritte entstehen im Hermes-Prozess, und `AGENTS.md` verbietet es, dessen Quellcode
zu patchen.

Beim Nachsehen zeigte sich aber, dass die HÄLFTE davon längst offen daliegt: Hermes
protokolliert jeden Werkzeug-Ruf nach `~/.hermes/logs/agent.log`. Eine Log-Datei zu lesen
ist kein Patchen. Was dadurch sichtbar wird — welches Werkzeug, wie lange, mit welchem
Ergebnis, in welchem Lauf — ist für die Frage „was tut sie gerade und wo hängt es" oft
nützlicher als der Fließtext ihrer Gedanken.

WAS NICHT GEHT (und hier auch nicht so tut): der Denkstrom selbst und die Argumente eines
Werkzeug-Rufs. Beides steht nicht im Log. Die Ansicht verspricht deshalb nur, was sie
halten kann.

ES HAT SICH SOFORT GELOHNT: Beim ersten Lesen fiel auf, dass `web_extract` seit
mindestens dem 25.08. JEDEN Morgen um 07:00 scheitert (der Suchanbieter kann keine
Seiten abrufen). Vier Tage lang, ohne dass es irgendwo aufgefallen wäre.
"""

import logging
import os
import re
from datetime import datetime
from pathlib import Path

log = logging.getLogger(__name__)

LOG_PFAD = Path(os.path.expanduser(
    os.environ.get("MC_HERMES_AGENT_LOG", "~/.hermes/logs/agent.log")))

# Nur das Ende der Datei lesen. Sie wächst auf mehrere MB und wird rotiert; für einen
# Verlauf der letzten Stunden reicht der Schwanz — und er kostet nichts.
LESE_BYTES = 512 * 1024

# Die drei Formen, in denen Hermes einen Werkzeug-Ruf notiert (am Log gemessen, nicht
# geraten). Die Lauf-Kennung in eckigen Klammern fehlt bei Nicht-Cron-Läufen.
#
#   INFO [cron_…] agent.tool_executor: tool terminal completed (1.40s, 53 chars)
#   INFO          agent.tool_executor: tool web_search completed (2.61s, 2944 chars)
#   WARNING [cron_…] agent.tool_executor: Tool web_extract returned error (0.17s): {…}
#   INFO          agent.tool_executor: tool web_extract failed (0.17s): {…}
_ZEIT = r"(?P<zeit>\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}),\d+"
_LAUF = r"(?:\[(?P<lauf>[^\]]+)\] )?"

_FERTIG = re.compile(
    _ZEIT + r" \w+ " + _LAUF + r"agent\.tool_executor: [Tt]ool (?P<werkzeug>\S+) completed "
    r"\((?P<dauer>[\d.]+)s, (?P<zeichen>\d+) chars\)")

_FEHLER = re.compile(
    _ZEIT + r" \w+ " + _LAUF + r"agent\.tool_executor: [Tt]ool (?P<werkzeug>\S+) "
    r"(?:failed|returned error) \((?P<dauer>[\d.]+)s\)(?::\s*(?P<detail>.*))?")


def _schwanz(pfad: Path, n: int) -> list[str]:
    """Die letzten n Bytes als Zeilen. Die erste Zeile kann angeschnitten sein und
    wird verworfen — ein halber Zeitstempel passt auf kein Muster, aber sicher ist besser."""
    try:
        groesse = pfad.stat().st_size
        with pfad.open("rb") as f:
            f.seek(max(0, groesse - n))
            roh = f.read()
        zeilen = roh.decode("utf-8", errors="replace").splitlines()
        return zeilen[1:] if groesse > n and zeilen else zeilen
    except OSError:
        return []


def _kurz(detail: str | None) -> str | None:
    """Fehlertext des Werkzeugs auf einen lesbaren Satz eindampfen. Hermes legt dort ein
    JSON ab; interessant ist daran nur das `error`-Feld."""
    if not detail:
        return None
    treffer = re.search(r'"error"\s*:\s*"([^"]{1,300})"', detail)
    text = treffer.group(1) if treffer else detail.strip()
    return (text[:297] + "…") if len(text) > 300 else text


def rufe(limit: int = 60) -> list[dict]:
    """Die jüngsten Werkzeug-Rufe, ältester zuerst."""
    ergebnis: list[dict] = []
    for zeile in _schwanz(LOG_PFAD, LESE_BYTES):
        m = _FERTIG.match(zeile)
        if m:
            ergebnis.append({
                "zeit": _iso(m.group("zeit")),
                "lauf": m.group("lauf"),
                "werkzeug": m.group("werkzeug"),
                "dauer_s": float(m.group("dauer")),
                "zeichen": int(m.group("zeichen")),
                "ok": True,
                "fehler": None,
            })
            continue
        m = _FEHLER.match(zeile)
        if m:
            ergebnis.append({
                "zeit": _iso(m.group("zeit")),
                "lauf": m.group("lauf"),
                "werkzeug": m.group("werkzeug"),
                "dauer_s": float(m.group("dauer")),
                "zeichen": None,
                "ok": False,
                "fehler": _kurz(m.group("detail")),
            })
    return ergebnis[-limit:]


def _iso(s: str) -> str:
    """`2026-08-28 07:03:18` → ISO. Die Box läuft in Europe/Berlin; die Zeit bleibt naiv,
    weil der Klient sie ohnehin nur anzeigt und nicht rechnet."""
    try:
        return datetime.strptime(s, "%Y-%m-%d %H:%M:%S").isoformat()
    except ValueError:
        return s


def uebersicht(limit: int = 60) -> dict:
    """Was die Agent-Ansicht braucht: die Rufe selbst, je Werkzeug eine Bilanz und die
    wiederkehrenden Fehler zusammengefasst.

    Die Bilanz ist der eigentliche Nutzen: Ein Werkzeug, das IMMER scheitert, verschwindet
    in einer Zeitleiste — in einer Zeile „web_extract · 4 Rufe · 4 Fehler" nicht."""
    liste = rufe(limit)
    if not LOG_PFAD.exists():
        return {"verfuegbar": False, "pfad": str(LOG_PFAD), "rufe": [],
                "werkzeuge": [], "laeufe": []}

    bilanz: dict[str, dict] = {}
    for r in liste:
        b = bilanz.setdefault(r["werkzeug"], {
            "werkzeug": r["werkzeug"], "rufe": 0, "fehler": 0,
            "dauer_summe": 0.0, "letzter_fehler": None,
        })
        b["rufe"] += 1
        b["dauer_summe"] += r["dauer_s"]
        if not r["ok"]:
            b["fehler"] += 1
            b["letzter_fehler"] = r["fehler"]

    werkzeuge = sorted(
        ({**b, "dauer_schnitt_s": round(b["dauer_summe"] / max(b["rufe"], 1), 2)}
         for b in bilanz.values()),
        key=lambda b: (-b["fehler"], -b["rufe"]),
    )

    # Läufe in der Reihenfolge ihres ersten Auftretens (nicht sortiert nach Kennung —
    # die trägt zwar ein Datum, aber darauf sollte sich niemand verlassen).
    laeufe: list[dict] = []
    gesehen: dict[str, dict] = {}
    for r in liste:
        schluessel = r["lauf"] or "(interaktiv)"
        if schluessel not in gesehen:
            gesehen[schluessel] = {"lauf": schluessel, "von": r["zeit"], "bis": r["zeit"],
                                   "rufe": 0, "fehler": 0}
            laeufe.append(gesehen[schluessel])
        e = gesehen[schluessel]
        e["bis"] = r["zeit"]
        e["rufe"] += 1
        if not r["ok"]:
            e["fehler"] += 1

    return {"verfuegbar": True, "pfad": str(LOG_PFAD), "rufe": liste,
            "werkzeuge": werkzeuge, "laeufe": laeufe}