mission-control-v2/deploy/agent-hooks/no-progress-bremse.py

#!/usr/bin/env python3
"""No-Progress-Bremse (Phase 2 des Drei-Welten-Plans, 19.07.2026).

Generische, turn-UEBERGREIFENDE Schleifenbremse — ergaenzt Hermes' native
tool_loop_guardrails, die (a) pro Turn zaehlen und (b) nur als Fehler
klassifizierte Ergebnisse sehen. Die teuer gelernte Luecke (SSH-Marathon 18.07.,
36 min / 75 Schleifen): ein Befehl laeuft "erfolgreich" durch, die AUSGABE sagt
aber immer dasselbe ("Permission denied") — der Agent variiert Befehle gegen
eine unsichtbare Wand, ueber Turn-Grenzen hinweg, und nichts stoppt ihn.

Prinzip (KEINE hartkodierten Fehler-Strings — Leitplanke des Commanders:
"wenn sich was aendert, muss es das SELBST erkennen"):
  post_tool_call  → Ergebnis normalisieren (Zahlen/Hex-IDs raus) → Muster-Hash.
                    Gleicher Hash beim gleichen Tool in Folge = Zaehler hoch.
                    Unbekannte Muster landen in neue-signaturen.jsonl —
                    Futter fuer Traum/Radar, um daraus Playbooks zu verdichten.
  pre_tool_call   → Zaehler >= 3 fuer dieses Tool: EINMAL mechanisch blocken
                    (Diagnose-Anweisung statt naechster Blindversuch), Zaehler
                    zuruecksetzen. Der Folgeversuch ist frei; laeuft er wieder
                    3x gegen dasselbe Muster, blockt es erneut.

Gezaehlt wird nur, wo Wiederholung wirklich Stillstand heisst: terminal IMMER
(dort verstecken sich die "erfolgreichen" Fehler), andere Tools nur bei
status=error. Lese-Tools deckt die native idempotent-Bremse ab.

Aufruf (hooks in config.yaml): `no-progress-bremse.py pre` bzw. `... post`.
State: ~/.hermes/state/no-progress/<session>.json (48 h Selbst-Aufraeumen).
Fail-open: jeder eigene Fehler -> {} (die Bremse darf nie selbst zur Wand werden).
"""
import hashlib
import json
import os
import re
import sys
import time

STATE_DIR = os.path.expanduser("~/.hermes/state/no-progress")
SIG_LOG = os.path.join(STATE_DIR, "neue-signaturen.jsonl")
SEEN_FILE = os.path.join(STATE_DIR, "seen-sigs.json")
THRESHOLD = 3
STATE_TTL = 48 * 3600

BLOCK_REASON = (
    "NO-PROGRESS-BREMSE: '{tool}' hat {n}x in Folge dasselbe Ergebnismuster geliefert — "
    "das ist Stillstand, egal wie sehr die Befehle variieren. STOPP. "
    "1) Benenne das Hindernis in EINEM Satz (was genau meldet das Ergebnis?). "
    "2) Diagnose statt Variation — bei SSH/Zugriff ZUERST: Schluessel-Passphrase pruefen "
    "(`ssh-keygen -y -f <key> </dev/null` — stilles 'Permission denied' = verschluesselter Key), "
    "dann Gegenseite (Auth-Log, Dienststatus, Erreichbarkeit) LESEN statt raten. "
    "3) Bleibt es unklar: Aufgabe als blocked/Karte mit dem EXAKTEN Ergebnistext an den "
    "Commander uebergeben. Dein naechster Versuch ist frei — nutze ihn erst NACH der Diagnose."
)

_NUM_RE = re.compile(r"\d+")
_HEX_RE = re.compile(r"\b[0-9a-f]{8,}\b")
_WS_RE = re.compile(r"\s+")


def _norm_sig(tool: str, text: str) -> str:
    t = (text or "")[:2000].lower()
    t = _HEX_RE.sub("#", _NUM_RE.sub("#", t))
    t = _WS_RE.sub(" ", t).strip()[:400]
    return hashlib.sha1(f"{tool}|{t}".encode()).hexdigest()[:16]


def _session_key(p: dict) -> str:
    sid = p.get("session_id") or p.get("parent_session_id") or ""
    if sid:
        return re.sub(r"[^A-Za-z0-9_.-]", "_", str(sid))[:80]
    return "cwd-" + hashlib.sha1(os.getcwd().encode()).hexdigest()[:12]


def _prune() -> None:
    now = time.time()
    for f in os.listdir(STATE_DIR):
        fp = os.path.join(STATE_DIR, f)
        if f.endswith(".json") and f != os.path.basename(SEEN_FILE):
            if now - os.path.getmtime(fp) > STATE_TTL:
                os.remove(fp)


def _load(path: str) -> dict:
    try:
        with open(path, encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:
        return {}


def _save(path: str, data: dict) -> None:
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(data, fh)
    os.replace(tmp, path)


def main() -> None:
    mode = sys.argv[1] if len(sys.argv) > 1 else "post"
    payload = json.load(sys.stdin)
    tool = str(payload.get("tool_name") or "")
    if not tool:
        print("{}")
        return
    os.makedirs(STATE_DIR, exist_ok=True)
    sfile = os.path.join(STATE_DIR, _session_key(payload) + ".json")
    state = _load(sfile)
    tools = state.setdefault("tools", {})

    if mode == "post":
        status = str(payload.get("status") or "ok")
        # terminal immer beobachten (dort tarnen sich Fehler als Erfolg),
        # sonst nur echte Fehler — Lese-Tools regelt die native Bremse.
        if tool != "terminal" and status != "error":
            print("{}")
            return
        sig = _norm_sig(tool, str(payload.get("result") or ""))
        entry = tools.get(tool) or {}
        entry["count"] = entry.get("count", 0) + 1 if entry.get("sig") == sig else 1
        entry["sig"] = sig
        tools[tool] = entry
        _save(sfile, state)
        _prune()
        seen = _load(SEEN_FILE)
        sigs = seen.setdefault("sigs", [])
        if sig not in sigs and status == "error" or (tool == "terminal" and sig not in sigs and entry["count"] >= 2):
            sigs.append(sig)
            del sigs[:-500]
            _save(SEEN_FILE, seen)
            sample = _WS_RE.sub(" ", str(payload.get("result") or ""))[:200]
            with open(SIG_LOG, "a", encoding="utf-8") as fh:
                fh.write(json.dumps({"ts": int(time.time()), "tool": tool,
                                     "sig": sig, "sample": sample}, ensure_ascii=False) + "\n")
        print("{}")
        return

    # mode == "pre"
    entry = tools.get(tool) or {}
    if entry.get("count", 0) >= THRESHOLD:
        entry["count"] = 0
        tools[tool] = entry
        _save(sfile, state)
        print(json.dumps({"decision": "block",
                          "reason": BLOCK_REASON.format(tool=tool, n=THRESHOLD)},
                         ensure_ascii=False))
        return
    print("{}")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        print("{}")