""" System-Seed-Regeln — werden bei jedem Brain-Boot idempotent in die Vector-DB geschrieben (pinned, source="seed"). Im Gegensatz zu aria-data/brain-import/ (User-Saatgut, manuell via Diagnostic-Klick migriert) ist das hier System-Regeln, die zum Brain-Code gehoeren und mit jedem Deploy ausgerollt werden. Idempotenz: Punkte mit gleicher `migration_key` werden vor dem Schreiben geloescht. Editieren = Zeile aendern, Brain neu starten, fertig. """ from __future__ import annotations import logging import uuid from datetime import datetime, timezone from typing import List from memory import Embedder, VectorStore from memory.vector_store import COLLECTION from qdrant_client.http import models as qm logger = logging.getLogger(__name__) # Jede Regel = ein eigener Memory-Punkt. Klein halten, klar formulieren — # ARIA sieht das in jedem Chat-Turn als pinned Hot Memory. SEED_RULES: List[dict] = [ { "migration_key": "seed/skill-rule/list-before-create", "type": "rule", "title": "Skill-Regel: skill_list vor skill_create", "category": "skills", "content": ( "Bevor du einen neuen Skill mit `skill_create` anlegst, ruf IMMER " "zuerst `skill_list` auf. Schau dir die Namen und Descriptions an. " "Wenn ein passender Skill existiert: verwende ihn oder verbessere " "ihn mit `skill_update`. Lege keinen Duplikat-Skill an." ), }, { "migration_key": "seed/skill-rule/no-version-suffix", "type": "rule", "title": "Skill-Regel: keine Versions-Suffixe im Namen", "category": "skills", "content": ( "Skill-Namen muessen permanent und beschreibend sein. NIEMALS " "Suffixe wie `-v2`, `_v3`, `-new`, `-fixed`, `-aria`, `-ctl` " "anhaengen, um eine neue Variante zu bauen. Wenn ein Skill kaputt " "ist oder verbessert werden soll: `skill_update`. Versionsverwaltung " "macht das System intern (Rollback ueber `skill_rollback`)." ), }, { "migration_key": "seed/skill-rule/update-not-recreate", "type": "rule", "title": "Skill-Regel: kaputten Skill reparieren, nicht neu bauen", "category": "skills", "content": ( "Wenn ein vorhandener Skill nicht wie erwartet funktioniert, lies " "zuerst Code + Logs (`skill_get`, `skill_logs`). Repariere ihn dann " "mit `skill_update` (entry_code, readme oder pip_packages patchen). " "Baue NIEMALS einen zweiten Skill mit aehnlichem Namen — das gibt " "Skill-Friedhof und Stefan muss aufraeumen." ), }, { "migration_key": "seed/skill-rule/no-hardcoded-credentials", "type": "rule", "title": "Skill-Regel: keine hardcoded Credentials", "category": "skills", "content": ( "Schreibe NIEMALS API-Keys, Tokens, Passwoerter, client_id oder " "client_secret direkt in den Skill-Code. Fuer OAuth-Services " "(Spotify, Google, GitHub etc.) nutze das Brain-Tool " "`oauth_get_token('')` — das macht Auto-Refresh und " "haelt den Token frisch. Stefan muss sich sonst alle 60 Minuten " "manuell neu einloggen, das nervt." ), }, { "migration_key": "seed/skill-rule/config-schema-for-settings", "type": "rule", "title": "Skill-Regel: konfigurierbare Werte ueber config_schema", "category": "skills", "content": ( "Wenn dein Skill konfigurierbare Werte braucht (User-IDs, " "Default-Geraete, Endpoints, nicht-OAuth-API-Keys), deklariere " "sie im `config_schema`-Feld der skill.json. Stefan setzt sie " "dann in der Diagnostic-UI; der Skill bekommt die Werte zur " "Laufzeit als Environment-Variable `CFG_`. NICHT als " "Argument, NICHT hardcoded." ), }, { "migration_key": "seed/skill-rule/brain-internal-url", "type": "rule", "title": "Skill-Regel: BRAIN_INTERNAL_URL ist deine Brain-Schnittstelle", "category": "skills", "content": ( "Jeder Skill bekommt die ENV-Variable BRAIN_INTERNAL_URL " "(Default http://localhost:8080). Damit kann der Skill das Brain " "aufrufen — kein hardcoden noetig:\n" " - GET {BRAIN_INTERNAL_URL}/oauth//token -> access_token " "(mit Auto-Refresh) fuer jeden OAuth-Service\n" " - GET {BRAIN_INTERNAL_URL}/memory/search?q=...&k=5 -> " "Stefans Memories semantisch durchsuchen\n" " - GET {BRAIN_INTERNAL_URL}/memory/pinned -> Hot Memory (Identitaet, Regeln)\n" " - GET {BRAIN_INTERNAL_URL}/skills/list -> verfuegbare Skills\n" "Mehr Endpoints siehe Brain main.py. Lies die URL IMMER aus " "os.environ['BRAIN_INTERNAL_URL'] — hardcoden waere kaputt sobald " "der Port wechselt. Beispiel: ein Wetter-Skill kann Stefans " "Standort per /memory/search holen statt ihn als Arg zu erwarten." ), }, { "migration_key": "seed/skill-rule/external-api-auth-strategy", "type": "rule", "title": "Skill-Regel: Auth-Strategie fuer externe APIs", "category": "skills", "content": ( "Wenn dein Skill mit einer externen API redet (Spotify, Google, " "Reddit, GitHub, OpenWeather, OpenAI, …), entscheide IMMER bewusst " "die Auth-Strategie in dieser Reihenfolge:\n" " 1. OAuth2? (Spotify, Google, GitHub, Reddit, Discord, Twitch, " "Microsoft, …) -> nutze `oauth_register_provider` falls der " "Provider noch nicht da ist, dann `oauth_authorize` fuer " "Initial-Login. Im Skill: Token via " "BRAIN_INTERNAL_URL/oauth//token holen — Brain macht " "Auto-Refresh, Stefan muss sich nicht alle 60min neu einloggen.\n" " 2. Statischer API-Key / Bearer-Token? (OpenWeather, OpenAI, " "Twilio, SendGrid, …) -> in skill.json `config_schema` " "deklarieren. Stefan setzt den Wert in Diagnostic, Skill bekommt " "ihn als CFG_ ENV.\n" " 3. NIEMALS hardcoden — egal wie 'temporaer' es ist.\n" "Wenn Du nicht sicher bist welche Strategie ein Service nutzt: " "in der API-Doku des Services nachsehen ('OAuth' oder " "'API Key' im Auth-Kapitel). Nicht raten." ), }, ] def apply(store: VectorStore, embedder: Embedder) -> dict: """Schreibt alle SEED_RULES idempotent in die DB. Vorgehen: erst alle Punkte mit `source=seed` UND passender migration_key loeschen, dann frisch upserten. So koennen Regeln editiert/entfernt werden indem die SEED_RULES-Liste angepasst wird. """ if not SEED_RULES: return {"written": 0} migration_keys = [r["migration_key"] for r in SEED_RULES] # Alte Versionen entfernen (nur die mit unserer migration_key — andere # source=seed Punkte aus zukuenftigen seed-Files sind sicher) try: store.client.delete( collection_name=COLLECTION, points_selector=qm.FilterSelector(filter=qm.Filter(must=[ qm.FieldCondition(key="migration_key", match=qm.MatchAny(any=migration_keys)) ])), ) except Exception as exc: logger.warning("seed_rules: delete-by-migration_key fehlgeschlagen (%s) — wahrscheinlich erster Run", exc) # Frisch einbetten + schreiben texts = [r["content"] for r in SEED_RULES] vectors = embedder.embed_batch(texts) now = datetime.now(timezone.utc).isoformat() written = 0 for rule, vec in zip(SEED_RULES, vectors): payload = { "type": rule["type"], "title": rule["title"], "content": rule["content"], "pinned": True, "category": rule.get("category", ""), "source": "seed", "tags": [], "created_at": now, "updated_at": now, "migration_key": rule["migration_key"], "attachments": [], } store.client.upsert( collection_name=COLLECTION, points=[qm.PointStruct(id=str(uuid.uuid4()), vector=vec, payload=payload)], ) written += 1 logger.info("seed_rules: %d Regeln in DB geschrieben", written) return {"written": written, "keys": migration_keys}