feat(agents): Sprint 1.12 Phase 2 — Cookie-Policy v3 + ImpressumAgent v3 finetune

ImpressumAgent v3 (Refactor):
  - v3_engine: laedt direkt alle 75 doc_check_controls['impressum'] ohne
    Sidecar-Filter (Sidecar war zu streng, lieferte nur 3 von 75 MCs).
  - Layer 0 Boost prueft pass+fail_criteria gegen meine 12 Patterns mit
    erweiterten Initial-Seeds (User-Vorgabe 2026-06-09:
    manuelle Initial-Seeds OK, Auto-Learning erweitert zur Laufzeit).
  - ETO-Smoke: 75 DB-MCs · 7 Pattern-Boosts · 24 Boost-Overrides
    (versus 3 DB-MCs vorher).

CookiePolicyAgent v3 (Refactor):
  - cookie_policy/v3_engine.py + cookie_policy/regex_boost.py
  - Laedt direkt alle 381 Cookie-MCs aus doc_check_controls
  - Layer 0 mit 12 eigenen Patterns als Initial-Seed
  - KB-Layer (CMP-Vendor-Cross-Check) bleibt erhalten
  - agent_version='3.0'

Tests: 27/27 gruen (12 v3-impressum, 6 cookie-policy, 9 cross-placement).
Alte v2-cookie-tests umgeschrieben auf v3-Pipeline-Mock.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend-compliance/compliance/services/specialist_agents/impressum/regex_boost.py

@@ -29,49 +29,70 @@ logger = logging.getLogger(__name__)
 # Für jedes meiner field_id: welche Wörter erscheinen typisch in
 # der pass_criteria der zugehörigen DB-MCs? Wenn diese Wörter im
 # pass_criteria gefunden werden, ist es vermutlich derselbe MC.
+# Initial-Seed der Standard-Synonyme pro field_id. User-Vorgabe
+# 2026-06-09: manuelle Erweiterung als Initial-Seed ist OK; das
+# LLM-basierte Auto-Learning (Sprint 1.10/1.11) ergänzt zur Laufzeit
+# weitere Tail-Schreibweisen, sodass über die Zeit asymptotisch
+# weniger LLM-Calls nötig sind.
 BOOST_KEYWORDS: dict[str, tuple[str, ...]] = {
     "name_anbieter": (
-        "rechtsform", "anschrift", "anbieter", "firmensitz", "firmenname",
-        "diensteanbieter", "verantwortlich",
+        # Adresse / Anschrift
+        "anschrift", "adresse", "postadresse", "postalisch",
+        "geschäftsadresse", "geschäftssitz", "firmensitz",
+        "niederlassung", "niederlassungsort", "sitz", "ort",
+        "straße", "hausnummer", "plz",
+        # Firmenname / Rechtsform
+        "firma", "firmenname", "rechtsform", "kaufmann",
+        "anbieter", "diensteanbieter", "verantwortlich",
+        "anbieterkennzeichnung", "unternehmen",
     ),
     "kontakt_email": (
         "e-mail", "email", "elektronische", "kontaktmöglichkeit",
-        "mailadresse",
+        "kontaktdaten", "mailadresse", "e-mail-adresse",
     ),
     "kontakt_telefon": (
-        "telefon", "rufnummer", "telefonnummer", "phone", "kontaktdaten",
-        "telekommunikation",
+        "telefon", "rufnummer", "telefonnummer", "phone",
+        "kontaktdaten", "telekommunikation", "fax",
     ),
     "handelsregister": (
-        "handelsregister", "registergericht", "hrb", "registernummer",
+        "handelsregister", "registergericht", "hrb", "hra",
+        "registernummer", "registereintrag",
+        "handelsregisternummer", "handelsregisterauszug",
     ),
     "ust_id": (
-        "umsatzsteuer", "ust-id", "umsatzsteueridentifikation", "ust-idnr",
+        "umsatzsteuer", "ust-id", "ust-idnr",
+        "umsatzsteueridentifikation",
+        "umsatzsteueridentifikationsnummer", "vat",
     ),
     "vertretungsberechtigte": (
-        "geschäftsführer", "vorstand", "vertretungsberechtigt",
-        "vertretung", "gesellschafter",
+        "geschäftsführer", "geschäftsführung", "vorstand",
+        "vorsitzender", "vorstandsvorsitzender",
+        "vertretungsberechtigt", "vertretung", "vertreten",
+        "gesellschafter", "kaufmann", "inhaber",
     ),
     "vertretungsberechtigte_label_korrekt": (
-        "deutsche", "bezeichnung", "rechtsform",
+        "geschäftsführer", "vorstand", "deutsche", "bezeichnung",
+        "rechtsform",
     ),
     "aufsichtsbehoerde": (
-        "aufsichtsbehörde", "aufsicht", "behörde", "regulierungsbehörde",
+        "aufsichtsbehörde", "aufsicht", "behörde",
+        "regulierungsbehörde", "ihk", "bafin", "bnetza", "kba",
     ),
     "verantwortlicher_redaktion": (
         "redaktion", "verantwortlich", "rstv", "mstv",
-        "journalistisch", "publizistisch",
+        "journalistisch", "publizistisch", "v.i.s.d.p",
     ),
     "verbraucher_streitbeilegung": (
         "streitbeilegung", "vsbg", "verbraucherschlichtung",
-        "schlichtungsstelle",
+        "schlichtungsstelle", "verbraucherschlichtungsstelle",
     ),
     "berufsangaben": (
-        "berufsbezeichnung", "berufsordnung", "kammer", "berufsrecht",
+        "berufsbezeichnung", "berufsordnung", "kammer",
+        "berufsrecht", "berufsverband",
     ),
     "odr_link": (
         "online-streitbeilegung", "os-plattform", "odr",
-        "europäische kommission",
+        "europäische kommission", "ec.europa.eu",
     ),
 }
 
@@ -94,22 +115,36 @@ def compute_regex_boosts(text: str, business_scope: set[str]) -> set[str]:
     return hits
 
 
-def boost_matches_db_mc(boosts: set[str], pass_criteria: list) -> str | None:
+def boost_matches_db_mc(
+    boosts: set[str],
+    pass_criteria: list,
+    fail_criteria: list | None = None,
+) -> str | None:
     """Hat ein gebooster field_id genug Keyword-Überlapp mit den
-    pass_criteria einer DB-MC, um den MC zu boost'en?
+    pass_criteria + fail_criteria einer DB-MC, um den MC zu boost'en?
 
-    Returns: field_id (matched), oder None.
-    Vorsichtig: ≥2 Boost-Keywords müssen im pass_criteria-Text auftauchen,
-    sonst zu permissiv.
+    Returns: field_id (matched, mit höchstem Keyword-Match-Count), oder None.
+
+    Schwelle: ≥2 unique Boost-Keywords im kombinierten Text.
+    Beide criteria-Listen werden berücksichtigt — fail_criteria-Wörter
+    wie 'Keine Adresse angegeben' helfen das MC eindeutig zuzuordnen.
     """
-    if not boosts or not pass_criteria:
+    if not boosts:
         return None
-    crit_text = " ".join(
-        str(c) for c in pass_criteria if c
-    ).lower()
+    crit_parts: list[str] = []
+    for c in (pass_criteria or []):
+        if c:
+            crit_parts.append(str(c).lower())
+    for c in (fail_criteria or []):
+        if c:
+            crit_parts.append(str(c).lower())
+    if not crit_parts:
+        return None
+    crit_text = " ".join(crit_parts)
     best: tuple[int, str] | None = None
     for field_id in boosts:
         kws = BOOST_KEYWORDS.get(field_id) or ()
+        # zähle UNIQUE hits — gleiches keyword im selben Text zählt einmal
         match_count = sum(1 for kw in kws if kw in crit_text)
         if match_count >= 2:
             if best is None or match_count > best[0]:

backend-compliance/compliance/services/specialist_agents/impressum/v3_engine.py

@@ -43,44 +43,67 @@ async def run_v3_pipeline(
     logger.info("v3 Layer-0 boosts: %d hits — %s",
                 len(boost_field_ids), boost_field_ids)
 
-    # Layer 1+2: bestehender rag_document_checker (Keyword + Embedding)
-    try:
-        from compliance.services.rag_document_checker import (
-            check_document_with_controls,
-        )
-        results = await check_document_with_controls(
-            text=text,
-            doc_type="impressum",
-            doc_title="Impressum (Agent-Test)",
-            db_url=db_url,
-            max_controls=0,
-            use_agent=False,
-            business_scope=business_scope,
-        )
-    except Exception as e:
-        logger.warning("rag_document_checker failed: %s — using boosts only",
-                       e)
-        results = []
+    # Layer 1: lade ALLE 75 doc_check_controls für 'impressum' direkt
+    # aus DB. Sidecar-Klassifizierung wird bewusst übersprungen — der
+    # Agent soll auf der vollen MC-Liste arbeiten (Layer 3 LLM-Validator
+    # demoted Pattern-Misses zu LOW, sodass Breitenwirkung kein Risiko ist).
+    controls = await _load_impressum_mcs()
+    results: list[dict[str, Any]] = []
+    if controls:
+        try:
+            from compliance.services.rag_document_checker import (
+                _check_mc_deterministic,
+            )
+            text_lower = text.lower().replace("\xad", "")
+            for mc in controls:
+                r = _check_mc_deterministic(text_lower, mc)
+                if r:
+                    # pass_criteria im Result behalten für Boost-Layer
+                    r["_pass_criteria"] = mc.get("pass_criteria")
+                    r["_fail_criteria"] = mc.get("fail_criteria")
+                    results.append(r)
+        except Exception as e:
+            logger.warning("layer-1 keyword check failed: %s", e)
+            results = []
+
+    # Layer 2: Embedding-Match für die failed MCs
+    failed_for_embed = [c for c, r in zip(controls, results)
+                        if r and not r.get("passed")]
+    if failed_for_embed:
+        try:
+            from compliance.services.mc_embedding_matcher import (
+                ensure_mc_embeddings, embedding_match,
+            )
+            await ensure_mc_embeddings()
+            semantic_passes = await embedding_match(
+                text, failed_for_embed, doc_type="impressum",
+            )
+            if semantic_passes:
+                for r in results:
+                    cid = r.get("control_id")
+                    if cid in semantic_passes and not r.get("passed"):
+                        r["passed"] = True
+                        r["matched_text"] = "[layer-2 embedding match]"
+                        r["source"] = (r.get("source") or "") + "+embedding"
+        except Exception as e:
+            logger.warning("layer-2 embedding skipped: %s", e)
 
     layer_1_pass = sum(1 for r in results if r.get("passed"))
     layer_1_fail = sum(1 for r in results
                        if r.get("passed") is False)
 
-    # Layer 0 Override: failed MCs deren pass_criteria zu einem meiner
-    # gebooster field_ids passt → überschreiben zu PASS
+    # Layer 0 Override: failed MCs deren pass/fail_criteria zu einem meiner
+    # gebooster field_ids passen → überschreiben zu PASS. Wir haben
+    # pass_criteria + fail_criteria in r drin (Layer-1 hat sie behalten).
     boost_overrides = 0
     for r in results:
         if r.get("passed"):
             continue
-        # rag_document_checker nimmt pass_criteria intern weg vor
-        # dem Return; wir laden sie nochmal (oder bekommen sie via
-        # 'hint'). Hier rufen wir das per Helper.
-        crit = r.get("_pass_criteria") or []
-        if not crit:
-            # Fallback: aus dem Hint (= check_question) Boost-Match
-            # versuchen.
-            crit = [r.get("hint") or ""]
-        matched_field = boost_matches_db_mc(boosts, crit)
+        pass_crit = r.get("_pass_criteria") or []
+        fail_crit = r.get("_fail_criteria") or []
+        if not pass_crit and not fail_crit:
+            pass_crit = [r.get("hint") or r.get("label") or ""]
+        matched_field = boost_matches_db_mc(boosts, pass_crit, fail_crit)
         if matched_field:
             r["passed"] = True
             r["matched_text"] = (
@@ -102,3 +125,52 @@ async def run_v3_pipeline(
     }
     logger.info("v3 telemetry: %s", telemetry)
     return results, telemetry
+
+
+async def _load_impressum_mcs() -> list[dict]:
+    """Lädt alle Impressum-MCs aus compliance.doc_check_controls — ohne
+    Sidecar-Filter. v3_engine nimmt die volle Breite."""
+    try:
+        import json
+        from classroom_engine.database import SessionLocal
+        from sqlalchemy import text as _sa_text
+        db = SessionLocal()
+        try:
+            rows = db.execute(_sa_text(
+                "SELECT id, control_id, control_uuid, title, regulation, "
+                "       article, check_question, pass_criteria, "
+                "       fail_criteria, severity "
+                "FROM compliance.doc_check_controls "
+                "WHERE doc_type='impressum' "
+                "ORDER BY severity DESC, title"
+            )).fetchall()
+        finally:
+            db.close()
+        out: list[dict] = []
+        for r in rows:
+            def _parse(v):
+                if isinstance(v, list):
+                    return v
+                if isinstance(v, str):
+                    try:
+                        j = json.loads(v)
+                        return j if isinstance(j, list) else [v]
+                    except Exception:
+                        return [v]
+                return []
+            out.append({
+                "id": str(r[0]),
+                "control_id": r[1],
+                "control_uuid": str(r[2]) if r[2] else "",
+                "title": r[3] or "",
+                "regulation": r[4] or "",
+                "article": r[5] or "",
+                "check_question": r[6] or "",
+                "pass_criteria": _parse(r[7]),
+                "fail_criteria": _parse(r[8]),
+                "severity": r[9] or "MEDIUM",
+            })
+        return out
+    except Exception as e:
+        logger.warning("_load_impressum_mcs failed: %s", e)
+        return []