feat(brain): Skill-Bypass-Detection + Bypass-Lehre als pinned Memory

Variante 3+ (Lerneffekt-Variante): Variante C scaffolded zwar Skills auto,
aber ARIA lernt nicht — sie wird beim naechsten Mal trotzdem zu Bash
greifen. Stefans Punkt: Lernen geht nur ueber Brain-Memory.

Mechanik:
1. api_heuristic.detect_recent_bypass(skills, since_sec=600):
   schaut letzte 10 Min im agent_stream.jsonl, findet Bash-curl gegen
   Hosts fuer die bereits ein matching Skill existiert. Returnt
   {host, skill_name, count, last_ts}.

2. api_heuristic.build_bypass_section(events):
   Drastischer Markdown-Block "## 🚨 SKILL-BYPASS ERKANNT" mit konkretem
   run_<skill>-Hint pro betroffenem Host. Landet direkt im System-Prompt
   noch VOR dem normalen API-Heuristik-Block.

3. agent.py._upsert_bypass_lesson(ev):
   Schreibt eine pinned type=rule Memory mit source=auto-feedback und
   migration_key=auto/skill-bypass/<skill_name>. Idempotent: bei
   Wiederholung wird die alte Memory ueberschrieben (Counter aktualisiert),
   keine Karteileichen. Content nennt konkret den run-Tool-Namen und
   Performance-Vergleich (3s Tool-Call vs 13-20s Bash-Wrapper).

Diese Memory ist permanent pinned → kommt bei jedem Chat-Turn,
cross-session, cross-restart als Hot-Memory durch. Damit lernt ARIA
es im wortlichen Sinne, nicht nur Reibung in der aktuellen Konversation.

Idempotenz wichtig: bei jedem Bypass-Detection-Lauf wird die Memory
upgedatet (nicht dupliziert). Stefan kann sie via Diagnostic-Gehirn-Tab
loeschen falls sie nervt.

Stefan-Frage beantwortet: 'sie wuerde es aber nur lernen wenn sie es
auch im gehirn speichert oder?' — exakt. Schimpfen im Prompt ist
Reibung dieser Session, pinned Memory ist permanenter Lerneffekt.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-30 00:37:40 +02:00
parent 298b2202a1
commit 210ce62ffe
3 changed files with 174 additions and 1 deletions
+93
View File
@@ -78,6 +78,99 @@ def invalidate_cache() -> None:
_cache.update(computed_at=0.0, hints=[])
def detect_recent_bypass(
existing_skills: list[dict],
since_sec: int = 600,
) -> list[dict]:
"""Findet Skill-Bypass-Vorfaelle: Bash-curl gegen einen Host fuer den
bereits ein matching Skill existiert. ARIA haette `run_<skill>` nutzen
sollen, hat aber gecurled. Das ist Drift — wir wollen es Brain merken.
Returns: liste {host, skill_name, count, last_ts} fuer Hosts wo ein
Bypass in den letzten `since_sec` Sekunden vorkam.
"""
if not AGENT_STREAM_LOG.exists() or not existing_skills:
return []
cutoff_ms = (time.time() - since_sec) * 1000
# Map host → matching skill_name
host_to_skill = {}
for s in existing_skills:
sname = (s.get("name") or "").lower()
if not sname:
continue
# Heuristik wie in _host_already_has_skill: stem des Skill-Namens
# mit Hostnamen verglichen. Fuer scaffolded skills nehmen wir den
# Skill-Namen als stem (z.B. "spotify" -> matched api.spotify.com)
host_to_skill[sname] = sname
bypass_events: dict[str, dict] = {}
try:
with AGENT_STREAM_LOG.open(encoding="utf-8") as f:
for line in f:
if not line.strip():
continue
try:
e = json.loads(line)
except Exception:
continue
if e.get("kind") != "tool_use":
continue
if (e.get("name") or "") != "Bash":
continue
ts = e.get("ts") or 0
if ts < cutoff_ms:
continue
for host in _extract_hosts_from_bash_input(e.get("input") or ""):
h = host.lower()
if h in _IGNORED_HOSTS:
continue
# Welcher Skill-Name matched diesen Host?
matched_skill = None
for skill_stem in host_to_skill:
if skill_stem in h:
matched_skill = host_to_skill[skill_stem]
break
if not matched_skill:
continue
entry = bypass_events.setdefault(h, {
"host": h, "skill_name": matched_skill,
"count": 0, "last_ts": 0,
})
entry["count"] += 1
if ts > entry["last_ts"]:
entry["last_ts"] = ts
except Exception as exc:
logger.warning("detect_recent_bypass: konnte log nicht lesen: %s", exc)
return []
return list(bypass_events.values())
def build_bypass_section(bypass_events: list[dict]) -> str:
"""Drastischer Block fuer den System-Prompt wenn ARIA gerade gegen einen
Host gecurled hat OBWOHL der Skill existiert. Inhalt soll sie spuerbar
ermahnen — wirkt nur in der aktuellen Session."""
if not bypass_events:
return ""
lines = [
"## 🚨 SKILL-BYPASS ERKANNT",
"",
"Du hast gerade — IN DEN LETZTEN MINUTEN — Bash-curl gegen Hosts "
"gemacht obwohl ein passender Skill existiert. Das ist Verschwendung: "
"5 Bash-Roundtrips à 3s statt 1 Tool-Call à 3s. Stefan wartet doppelt. "
"AB JETZT in diesem Chat:",
"",
]
for ev in bypass_events:
sname = ev["skill_name"]
host = ev["host"]
count = ev["count"]
lines.append(f"- gegen **{host}** ({count}x kuerzlich) → nutze "
f"`run_{sname.replace('-', '_')}(...)` statt curl. "
f"Der Skill ist da. Nutze ihn.")
lines.append("")
return "\n".join(lines)
def _extract_hosts_from_bash_input(input_str: str) -> list[str]:
"""Hostnames aus URLs in einem Bash-Command. Sehr robust — sucht `https?://host`."""
if not input_str: