#!/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/.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 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 # Delegations-Ausnahme (20.07.2026): worker.sh / fremdblick.sh sind der # Fliessband-/Orchestrator-Motor — konzept-fliessband ruft sie pro Runde/Rolle # WIEDERHOLT auf (verschiedene Prompts, aehnliche Prosa-Ergebnisse). Die Bremse # bildet ihre Signatur aus dem Ergebnis-Text → sie hielt diese legitimen # Mehrrunden faelschlich fuer eine Schleife und blockte (Pomodoro-IDE-Test lief # 19 min dagegen an). Diese Helfer haben ihr EIGENES Runden-Limit (max 3) → # von der generischen Bremse ausnehmen. _ti = payload.get("tool_input") _cmd = _ti.get("command") if isinstance(_ti, dict) else (_ti if isinstance(_ti, str) else "") if tool == "terminal" and ("worker.sh" in str(_cmd) or "fremdblick.sh" in str(_cmd)): 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("{}")