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>

README.md

@@ -402,7 +402,7 @@ jedem Chat-Turn im Hot-Memory-Block auf:
 - **oauth-reauth-reflex** — bei 401: ZUERST `oauth_get_token` (Auto-Refresh), nur bei dessen Fehler `oauth_authorize`
 - **no-skill-drift** — kein Drift vom Skill zu Ad-hoc-Bash-Befehlen. Skill kaputt? `skill_logs` + `skill_update`. Niemals nur SAGEN „ich baue dir einen Skill", wenn `skill_create` nicht wirklich gefeuert wird
 - **runtime-topology** (architektur) — ARIA laeuft als `claude`-CLI-Subprocess IM aria-proxy Container (alpine — kein python3/jq), NICHT im aria-brain. `/data/skills/` und `BRAIN_INTERNAL_URL` existieren dort nicht. Brain-Resources via Brain-Tools (`oauth_get_token`, `memory_search`, `run_<skill>` …), nicht via Bash. SSH zur VM-Host via `ssh aria@host` (Key liegt im Proxy)
-- **scaffold-reflex** — Brain trackt cross-session welche externen Hosts via Bash-curl wiederholt (≥3× in 24h) ohne passenden Skill aufgerufen wurden. Ergebnis landet als `## API-Heuristik`-Block im System-Prompt mit konkretem `skill_scaffold(...)`-Vorschlag → ARIA scaffolded statt zu curlen. Data-Source: `agent_stream.jsonl`, Cache 5 min
+- **scaffold-reflex** — Brain trackt cross-session welche externen Hosts via Bash-curl wiederholt (≥3× in 24h) ohne passenden Skill aufgerufen wurden. Ergebnis landet als `## API-Heuristik`-Block im System-Prompt. **Auto-Scaffold**: bei bekannten Hosts (Spotify, GitHub, OpenAI etc.) legt Brain den Skill automatisch an — ARIA findet ihn beim nächsten Turn vor (author=`aria-auto`) und nutzt `run_<name>` statt curlen. Toggle via ENV `BRAIN_AUTO_SCAFFOLD=false`. **Bypass-Lehre**: wenn ARIA trotz vorhandenem Skill weiter curlt (Skill-Bypass), erkennt Brain das im agent_stream und (1) injiziert einen drastischen `🚨 SKILL-BYPASS`-Hint im aktuellen System-Prompt und (2) speichert ein pinned `type=rule, source=auto-feedback` Memory mit Skill+Host (idempotent via migration_key `auto/skill-bypass/<skill>`) — damit lernt sie es cross-session, nicht nur in der aktuellen Konversation. Data-Source: `agent_stream.jsonl`, Cache 5 min
 - **external-api-auth-strategy** — OAuth2 → `oauth_get_token`, sonst `config_schema`, NIEMALS hardcoden
 
 ### Skill-Scaffold (Templates)

aria-brain/agent.py

@@ -829,6 +829,65 @@ class Agent:
         self._pending_events = []
         return events
 
+    def _upsert_bypass_lesson(self, ev: dict) -> None:
+        """Speichert die Lehre aus einem Skill-Bypass als pinned Memory.
+        Idempotent ueber migration_key — bei Wiederholung wird der vorhandene
+        Punkt aktualisiert (Counter hoeher). So lernt ARIA cross-session,
+        nicht nur in der aktuellen Konversation."""
+        from datetime import datetime, timezone
+        import uuid as _uuid
+        from memory.vector_store import COLLECTION
+        from qdrant_client.http import models as _qm
+
+        skill_name = ev["skill_name"]
+        host = ev["host"]
+        count = ev["count"]
+        migration_key = f"auto/skill-bypass/{skill_name}"
+        title = f"Skill '{skill_name}' nutzen, nicht curl"
+        run_tool = f"run_{skill_name.replace('-', '_')}"
+        content = (
+            f"WICHTIG fuer Performance + Stefans Wartezeit: "
+            f"Skill '{skill_name}' existiert und deckt {host} ab. "
+            f"Nutze `{run_tool}(...)` als Brain-Tool, NICHT Bash-curl gegen {host}. "
+            f"Brain hat {count}× erkannt dass dieser Skill umgangen wurde "
+            f"(letzter Vorfall: heute). Ein Skill-Aufruf = 1 Tool-Call (~3s) "
+            f"vs. Bash-Wrapper = 3-5 Tool-Calls (~13-20s)."
+        )
+
+        # Alte Version mit gleicher migration_key entfernen (Counter-Update)
+        try:
+            self.store.client.delete(
+                collection_name=COLLECTION,
+                points_selector=_qm.FilterSelector(filter=_qm.Filter(must=[
+                    _qm.FieldCondition(key="migration_key",
+                                       match=_qm.MatchValue(value=migration_key))
+                ])),
+            )
+        except Exception:
+            pass
+
+        vec = self.embedder.embed(content)
+        now = datetime.now(timezone.utc).isoformat()
+        payload = {
+            "type": "rule",
+            "title": title,
+            "content": content,
+            "pinned": True,
+            "category": "skills",
+            "source": "auto-feedback",
+            "tags": [],
+            "created_at": now,
+            "updated_at": now,
+            "migration_key": migration_key,
+            "attachments": [],
+        }
+        self.store.client.upsert(
+            collection_name=COLLECTION,
+            points=[_qm.PointStruct(id=str(_uuid.uuid4()), vector=vec, payload=payload)],
+        )
+        logger.info("bypass-lesson upserted: skill=%s host=%s count=%d",
+                    skill_name, host, count)
+
     # ── Hauptpfad: ein User-Turn → Tool-Loop → finaler Reply ──
 
     MAX_TOOL_ITERATIONS = 8  # Schutz vor Endlos-Loops
@@ -887,7 +946,76 @@ class Agent:
         try:
             import api_heuristic as _ah
             hints = _ah.compute_hints(existing_skills=all_skills)
+
+            # AUTO-SCAFFOLD (Variante C): wenn ein Hinweis ein konkretes
+            # (name, template, params) hat UND der Skill noch nicht existiert,
+            # legt Brain ihn JETZT an — bevor ARIA wieder Bash-curl macht.
+            # ARIA findet den Skill in den naechsten Tool-Listen vor und
+            # nutzt ihn direkt via `run_<name>`. Toggle via ENV.
+            auto_scaffold = os.environ.get("BRAIN_AUTO_SCAFFOLD", "true").strip().lower() != "false"
+            if auto_scaffold and hints:
+                existing_names = {s.get("name") for s in all_skills}
+                scaffolded_any = False
+                for hint in hints:
+                    sug = hint.get("suggestion")
+                    if not sug:
+                        continue
+                    sname, stpl, sparams = sug
+                    if sname in existing_names:
+                        continue
+                    try:
+                        new_manifest = skills_mod.scaffold_skill(
+                            name=sname, template=stpl, params=sparams, author="aria-auto",
+                        )
+                        logger.info("auto_scaffold: '%s' aus '%s' angelegt (trigger: %s mit %d Calls)",
+                                    sname, stpl, hint["host"], hint["count"])
+                        self._pending_events.append({
+                            "type": "skill_created",
+                            "skill": {
+                                "name": new_manifest["name"],
+                                "description": new_manifest.get("description", ""),
+                                "execution": new_manifest.get("execution", ""),
+                                "active": new_manifest.get("active", True),
+                                "setup_error": new_manifest.get("setup_error"),
+                                "auto_scaffolded": True,
+                                "from_template": stpl,
+                                "trigger_host": hint["host"],
+                                "trigger_count": hint["count"],
+                            },
+                        })
+                        scaffolded_any = True
+                    except Exception as exc:
+                        logger.warning("auto_scaffold '%s' fehlgeschlagen: %s", sname, exc)
+                if scaffolded_any:
+                    # Skills-Liste refresh damit der frische Skill im Prompt sichtbar ist
+                    all_skills = skills_mod.list_skills(active_only=False)
+                    active_skills = [s for s in all_skills if s.get("active", True)]
+                    _ah.invalidate_cache()
+                    # Heuristik neu rechnen — die scaffold-targets sind jetzt weg
+                    hints = _ah.compute_hints(existing_skills=all_skills, force=True)
+
             api_heuristic_section = _ah.build_section(hints)
+
+            # BYPASS-DETECTION (Variante 3 / Lerneffekt):
+            # Hat ARIA in den letzten ~10min Bash-curl gegen einen Host
+            # gemacht OBWOHL der Skill existiert? → drastischer Hint im
+            # Prompt JETZT + pinned Memory speichern, damit's beim
+            # naechsten Turn / naechster Session weiter sichtbar ist
+            # ("echtes Lernen via Brain-Memory").
+            bypass_events = _ah.detect_recent_bypass(all_skills, since_sec=600)
+            if bypass_events:
+                bypass_section = _ah.build_bypass_section(bypass_events)
+                if bypass_section:
+                    api_heuristic_section = (
+                        (bypass_section + "\n\n" + api_heuristic_section)
+                        if api_heuristic_section else bypass_section
+                    )
+                # Pinned-Memory pro Skill speichern, idempotent ueber migration_key
+                for ev in bypass_events:
+                    try:
+                        self._upsert_bypass_lesson(ev)
+                    except Exception as exc:
+                        logger.warning("bypass-lesson upsert fehlgeschlagen: %s", exc)
         except Exception as exc:
             logger.warning("api_heuristic fehlgeschlagen: %s", exc)
 

aria-brain/api_heuristic.py

@@ -72,6 +72,105 @@ _SUGGESTIONS: dict[str, tuple[str, str, dict]] = {
 _cache: dict = {"computed_at": 0.0, "hints": []}
 
 
+def invalidate_cache() -> None:
+    """Cache leeren — sinnvoll nach skill_create / scaffold damit der neue
+    Skill sofort beim naechsten Aufruf erkannt wird."""
+    _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:

aria-brain/seed_rules.py

@@ -216,11 +216,22 @@ SEED_RULES: List[dict] = [
         "content": (
             "Brain trackt server-side wie oft Du in den letzten 24h dieselbe "
             "externe API per Bash-curl angerufen hast (Cross-Session-Counter, "
-            "siehe '## API-Heuristik'-Block im System-Prompt). Sobald da "
-            "ein Eintrag steht: das ist KEINE Empfehlung sondern eine "
-            "Aufforderung. RUFE als ALLERERSTES `skill_scaffold` mit dem "
-            "vorgeschlagenen Template und params auf, BEVOR Du wieder Bash-"
-            "curl machst. Dann nutze den frischen Skill via `run_<name>`.\n"
+            "siehe '## API-Heuristik'-Block im System-Prompt). \n"
+            "\n"
+            "AUTO-SCAFFOLD: Brain legt fuer wiederkehrende Hosts mit "
+            "bekanntem Template (Spotify, GitHub, OpenAI, OpenWeather, …) "
+            "automatisch einen Skill an — Du siehst ihn dann in `## Skills` "
+            "ohne dass Du ihn selbst gebaut hast (Markierung "
+            "`author=aria-auto`). NUTZE diesen Skill via `run_<name>` "
+            "direkt, NICHT mehr Bash-curl gegen den Host. Beispiel: wenn "
+            "`spotify` plotzlich in der Skill-Liste auftaucht → "
+            "`run_spotify({method:'GET', path:'/v1/me/player'})` statt "
+            "Token holen + curl.\n"
+            "\n"
+            "Wenn die API-Heuristik einen Eintrag OHNE Suggestion zeigt "
+            "(unbekannter Host): rufe selbst `skill_scaffold` mit dem "
+            "passenden Template (oauth-api / apikey-api / file-process), "
+            "BEVOR Du wieder Bash-curl machst.\n"
             "\n"
             "Warum: jede Chat-Anfrage ist eine eigene Claude-CLI-Session — "
             "Du siehst nicht dass Du gestern auch schon 10x Spotify gecurled "