fix(ucca): Cross-Reg 0070 — beide Domaenen im Router-Top-K (#47 )

style(ucca): gocritic equalFold in balanceByRegulation (go-lint gruen)
strings.EqualFold(code, cv) statt code==strings.ToUpper(cv) — behebt den einzigen gocritic-Befund auf der neuen Zeile (CI go-lint, new-from-merge-base). Verhalten unveraendert (case-insensitive exakter regulation_code-Match); Unit + 0070-e2e bleiben gruen. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-30 13:42:28 +00:00 · 2026-06-30 15:30:58 +02:00 · 2026-06-30 15:08:18 +02:00 · 2026-06-30 12:26:53 +00:00 · 2026-06-30 14:13:09 +02:00 · 2026-06-30 10:01:59 +00:00
33 changed files with 1360 additions and 146 deletions
@@ -0,0 +1,200 @@
 'use client'
 // ETO / Onboarding-Advisor — thin operator surface over POST /api/compliance/onboarding/advisor-start.
 // Certifications + target + scanner findings -> Silent Pass -> Advisor. NOT the regulation gap engine
 // (/sdk/gap-analysis is a different flow: product -> applicable regulations). This tests the cert->delta
 // case: "TISAX/ISO27001 -> CRA, what is auto-detected, what stays an open question?". No new backend.
 import React, { useEffect, useState } from 'react'
 const CERTS = ['ISO27001', 'TISAX', 'ISO9001', 'IEC62443', 'ISO13485', 'ISO14001', 'ASPICE', 'IATF16949']
 // label -> {signal_id, source_type} — demonstrates all three signal KINDS (observation / partial / requirement)
 const FINDINGS: Array<{ label: string; signal_id: string; source_type: string; kind: string }> = [
  { label: 'SBOM im Repo (CycloneDX/SPDX)', signal_id: 'cyclonedx_found', source_type: 'repository', kind: 'observation' },
  { label: 'security.txt / CVD-Policy veröffentlicht', signal_id: 'security_txt', source_type: 'website', kind: 'observation' },
  { label: 'Signierte Releases', signal_id: 'signed_releases', source_type: 'repository', kind: 'observation' },
  { label: 'Produkt-Risikobewertung (Dokument)', signal_id: 'risk_assessment_pdf', source_type: 'document', kind: 'observation' },
  { label: 'CI-Pipeline vorhanden (nur Indikation)', signal_id: 'github_actions_ci', source_type: 'repository', kind: 'partial' },
  { label: 'Cloud-/vernetztes Produkt', signal_id: 'cloud_hosted', source_type: 'product', kind: 'observation' },
  { label: 'Ausschreibung FORDERT SBOM (Requirement)', signal_id: 'requires_sbom', source_type: 'tender', kind: 'requirement' },
  { label: 'OEM FORDERT PSIRT (Requirement)', signal_id: 'supplier_requires_psirt', source_type: 'oem', kind: 'requirement' },
 ]
 interface Question { capability_id: string; question_intent: string; why: string; information_value: number; priority: string }
 interface Inferred { certification: string; capabilities: string[]; statement: string }
 interface Rejected { certification?: string; statement: string; reason: string }
 interface Measure { capability_id: string; leverage: number; closes: string[] }
 interface AdvisorResponse {
  silent_intake_summary: string; headline: string; auto_detected: string[]; indications: string[]
  inferred_assumptions: Inferred[]; rejected_assumptions: Rejected[]; top_5_questions: Question[]
  capability_delta: string[]; top_measures: Measure[]; evidence_requests: string[]
  unsupported_domains: string[]; completeness_summary: string; capability_labels: Record<string, string>
 }
 const PROXY = '/api/sdk/v1/compliance/onboarding'
 function Chips({ items, tone }: { items: string[]; tone: string }) {
  if (!items.length) return <span className="text-gray-400 text-sm">—</span>
  return (
    <div className="flex flex-wrap gap-2">
      {items.map(c => <span key={c} className={`px-2.5 py-1 rounded-full text-xs font-medium ${tone}`}>{c}</span>)}
    </div>
  )
 }
 function Section({ title, hint, children }: { title: string; hint?: string; children: React.ReactNode }) {
  return (
    <div className="bg-white rounded-xl border border-gray-200 p-5">
      <h3 className="font-semibold text-gray-900">{title}</h3>
      {hint && <p className="text-xs text-gray-500 mt-0.5 mb-2">{hint}</p>}
      <div className="mt-2">{children}</div>
    </div>
  )
 }
 export default function OnboardingAdvisorPage() {
  const [targets, setTargets] = useState<string[]>([])
  const [company, setCompany] = useState('Beispiel Maschinenbau')
  const [industry, setIndustry] = useState('machine_builder')
  const [certs, setCerts] = useState<string[]>(['ISO27001', 'ISO9001'])
  const [target, setTarget] = useState('CRA')
  const [findings, setFindings] = useState<string[]>(['cyclonedx_found', 'github_actions_ci', 'requires_sbom'])
  const [knownEvidence, setKnownEvidence] = useState('CE-Prozess')
  const [result, setResult] = useState<AdvisorResponse | null>(null)
  const [loading, setLoading] = useState(false)
  const [error, setError] = useState('')
  useEffect(() => {
    fetch(`${PROXY}/targets`).then(r => r.json()).then(d => {
      if (Array.isArray(d.targets)) { setTargets(d.targets); if (!d.targets.includes('CRA') && d.targets[0]) setTarget(d.targets[0]) }
    }).catch(() => {})
  }, [])
  const toggle = (list: string[], set: (v: string[]) => void, v: string) =>
    set(list.includes(v) ? list.filter(x => x !== v) : [...list, v])
  const lbl = (id: string) => result?.capability_labels?.[id] || id.replace(/_/g, ' ')
  const run = async () => {
    setLoading(true); setError(''); setResult(null)
    try {
      const scanner_findings = FINDINGS.filter(f => findings.includes(f.signal_id))
        .map(f => ({ signal_id: f.signal_id, source_type: f.source_type }))
      const res = await fetch(`${PROXY}/advisor-start`, {
        method: 'POST', headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          company, industry, products: [], markets: ['EU'], certifications: certs,
          known_evidence: knownEvidence ? knownEvidence.split(',').map(s => s.trim()).filter(Boolean) : [],
          target, scanner_findings,
        }),
      })
      if (!res.ok) throw new Error(await res.text())
      setResult(await res.json())
    } catch (e) {
      setError(e instanceof Error ? e.message : 'Advisor fehlgeschlagen')
    } finally { setLoading(false) }
  }
  // auto-recompute when certifications / target / scanner signals change (no button click needed)
  useEffect(() => { if (certs.length) run() }, [certs, target, findings])  // eslint-disable-line react-hooks/exhaustive-deps
  return (
    <div className="min-h-screen bg-gray-50 py-8">
      <div className="max-w-5xl mx-auto px-4">
        <h1 className="text-3xl font-bold text-gray-900">ETO / Onboarding-Advisor</h1>
        <p className="text-gray-600 mt-2 mb-6">
          Zertifikate + Ziel + Scanner-Signale → Silent Pass → Capability-Delta + nächste beste Fragen.
          Welt-1: ein Zertifikat <em>legt nahe</em>, beweist nichts (Verifikation erforderlich).
        </p>
        <div className="grid md:grid-cols-2 gap-4 mb-6">
          <Section title="Unternehmen & Ziel">
            <label className="block text-sm text-gray-600">Unternehmen
              <input value={company} onChange={e => setCompany(e.target.value)} className="mt-1 w-full border rounded-lg px-3 py-2" /></label>
            <label className="block text-sm text-gray-600 mt-3">Branche
              <input value={industry} onChange={e => setIndustry(e.target.value)} className="mt-1 w-full border rounded-lg px-3 py-2" /></label>
            <label className="block text-sm text-gray-600 mt-3">Ziel
              <select value={target} onChange={e => setTarget(e.target.value)} className="mt-1 w-full border rounded-lg px-3 py-2">
                {(targets.length ? targets : ['CRA']).map(t => <option key={t} value={t}>{t}</option>)}
              </select></label>
            <label className="block text-sm text-gray-600 mt-3">Vorhandene Nachweise (kommagetrennt)
              <input value={knownEvidence} onChange={e => setKnownEvidence(e.target.value)} className="mt-1 w-full border rounded-lg px-3 py-2" /></label>
          </Section>
          <Section title="Zertifizierungen">
            <div className="flex flex-wrap gap-2">
              {CERTS.map(c => (
                <button key={c} onClick={() => toggle(certs, setCerts, c)}
                  className={`px-3 py-1.5 rounded-lg text-sm border ${certs.includes(c) ? 'bg-blue-600 text-white border-blue-600' : 'bg-white text-gray-700 border-gray-300'}`}>{c}</button>
              ))}
            </div>
          </Section>
        </div>
        <Section title="Scanner-Signale (Silent Pass)" hint="observation = gesehen · partial = Indikation · requirement = gefordert (≠ vorhanden)">
          <div className="grid sm:grid-cols-2 gap-2">
            {FINDINGS.map(f => (
              <label key={f.signal_id} className="flex items-center gap-2 text-sm text-gray-700">
                <input type="checkbox" checked={findings.includes(f.signal_id)} onChange={() => toggle(findings, setFindings, f.signal_id)} />
                <span>{f.label}</span>
                <span className={`ml-auto text-[10px] px-1.5 py-0.5 rounded ${f.kind === 'requirement' ? 'bg-purple-100 text-purple-700' : f.kind === 'partial' ? 'bg-amber-100 text-amber-700' : 'bg-emerald-100 text-emerald-700'}`}>{f.kind}</span>
              </label>
            ))}
          </div>
        </Section>
        <button onClick={run} disabled={loading || !certs.length}
          className="mt-6 w-full py-3 bg-blue-600 text-white rounded-xl font-medium hover:bg-blue-700 disabled:opacity-50">
          {loading ? 'Analysiere…' : 'Advisor starten'}
        </button>
        {error && <div className="mt-6 bg-red-50 border border-red-200 rounded-lg p-4 text-red-700 text-sm whitespace-pre-wrap">{error}</div>}
        {result && (
          <div className="mt-8 space-y-4">
            <div className="bg-blue-600 text-white rounded-xl p-5">
              <div className="text-lg font-semibold">{result.headline}</div>
              <div className="text-blue-100 text-sm mt-1">{result.silent_intake_summary}</div>
            </div>
            <div className="grid md:grid-cols-2 gap-4">
              <Section title="Automatisch erkannt" hint="konkrete Artefakte – nicht mehr gefragt"><Chips items={result.auto_detected.map(lbl)} tone="bg-emerald-100 text-emerald-800" /></Section>
              <Section title="Indikationen" hint="erhöht Annahmestärke – trotzdem gefragt"><Chips items={result.indications.map(lbl)} tone="bg-amber-100 text-amber-800" /></Section>
            </div>
            <Section title="Nächste beste Fragen" hint="max 5, jede erklärt sich selbst">
              {result.top_5_questions.length ? (
                <ol className="space-y-3">
                  {result.top_5_questions.map((q, i) => (
                    <li key={q.capability_id} className="border-l-2 border-blue-300 pl-3">
                      <div className="font-medium text-gray-900">{i + 1}. {lbl(q.capability_id)}</div>
                      <div className="text-sm text-gray-600">{q.why}</div>
                    </li>
                  ))}
                </ol>
              ) : <span className="text-gray-400 text-sm">—</span>}
            </Section>
            <div className="grid md:grid-cols-2 gap-4">
              <Section title="Wahrscheinlich abgedeckt (Welt-1)" hint="Zertifikat legt nahe – Verifikation erforderlich">
                {result.inferred_assumptions.length ? result.inferred_assumptions.map(a => (
                  <div key={a.certification} className="mb-2"><span className="font-medium">{a.certification}</span>: {a.capabilities.map(lbl).join(', ')}</div>
                )) : <span className="text-gray-400 text-sm">—</span>}
              </Section>
              <Section title="Nicht relevant" hint="relevance(evidence, target) = 0">
                {result.rejected_assumptions.length ? result.rejected_assumptions.map((a, i) => (
                  <div key={i} className="mb-1 text-sm text-gray-700">{a.statement}</div>
                )) : <span className="text-gray-400 text-sm">—</span>}
              </Section>
            </div>
            <div className="grid md:grid-cols-2 gap-4">
              <Section title="Offene Lücken (Delta)"><Chips items={result.capability_delta.map(lbl)} tone="bg-gray-100 text-gray-700" /></Section>
              <Section title="Geforderte Nachweise"><Chips items={result.evidence_requests} tone="bg-gray-100 text-gray-700" /></Section>
            </div>
            <Section title="Vollständigkeit" hint={result.unsupported_domains.length ? `nicht abgedeckt: ${result.unsupported_domains.join(', ')}` : undefined}>
              <span className="text-sm text-gray-700">{result.completeness_summary || '—'}</span>
            </Section>
          </div>
        )}
      </div>
    </div>
  )
 }
@@ -51,8 +51,8 @@ describe('advisor-rag', () => {
    })
  })
-  describe('queryAdvisorRAG', () => {
+  describe('queryAdvisorRAG (Authority Router)', () => {
-    it('fragt alle 6 Collections ab und formatiert die Treffer', async () => {
+    it('ruft den Router EINMAL auf und formatiert die Treffer', async () => {
      mockFetch.mockResolvedValue({
        ok: true,
        json: async () => ({ results: [{ text: 'Inhalt A', regulation_short: 'DSGVO', score: 0.9 }] }),
@@ -60,19 +60,19 @@ describe('advisor-rag', () => {
      const result = await mod.queryAdvisorRAG('Was ist eine DSFA?')
      expect(result).toContain('[Quelle 1: DSGVO]')
      expect(result).toContain('Inhalt A')
-      expect(mockFetch).toHaveBeenCalledTimes(mod.COMPLIANCE_COLLECTIONS.length)
+      expect(mockFetch).toHaveBeenCalledTimes(1)
    })
-    it('ruft die ai-sdk /sdk/v1/rag/search mit collection + top_k auf', async () => {
+    it('ruft /sdk/v1/rag/retrieve mit query + top_k (ohne collection) auf', async () => {
      mockFetch.mockResolvedValue({ ok: true, json: async () => ({ results: [] }) })
      await mod.queryAdvisorRAG('test')
      expect(mockFetch).toHaveBeenCalledWith(
-        expect.stringContaining('/sdk/v1/rag/search'),
+        expect.stringContaining('/sdk/v1/rag/retrieve'),
        expect.objectContaining({ method: 'POST' }),
      )
      const body = JSON.parse(mockFetch.mock.calls[0][1].body)
-      expect(body).toMatchObject({ query: 'test', top_k: 3 })
+      expect(body).toMatchObject({ query: 'test', top_k: 8 })
-      expect(mod.COMPLIANCE_COLLECTIONS).toContain(body.collection)
+      expect(body.collection).toBeUndefined()
    })
    it('liefert leeren String wenn das RAG-Backend nicht erreichbar ist (graceful)', async () => {
@@ -80,10 +80,5 @@ describe('advisor-rag', () => {
      const result = await mod.queryAdvisorRAG('test')
      expect(result).toBe('')
    })
    it('umfasst genau die 6 Compliance-Collections', () => {
      expect(mod.COMPLIANCE_COLLECTIONS).toHaveLength(6)
      expect(mod.COMPLIANCE_COLLECTIONS).toContain('bp_compliance_recht')
    })
  })
 })
@@ -1,12 +1,13 @@
 /**
 * Compliance-Advisor RAG-Suche.
 *
- * Fragt die ai-compliance-sdk (`/sdk/v1/rag/search`) ab statt des frueheren
+ * Fragt den Authority Router der ai-compliance-sdk (`/sdk/v1/rag/retrieve`) mit NUR der
- * `rag-service:8097` (auf prod nicht erreichbar). Die ai-sdk embeddet die Query
+ * Query ab — der Router waehlt selbst die Collections (Broad-Authority-Base + KB-2026.1-Slice
- * mit bge-m3 (prod: ollama-embed) und sucht in den Qdrant-Compliance-Collections
+ * bei in-scope), embeddet mit bge-m3 (prod: ollama-embed), merged + authority-ranked. Der
- * — damit profitiert der Advisor vom reicheren Embedding.
+ * Advisor bleibt damit collection-agnostisch (Vertrag: Compiler -> Collections -> Retriever
 * -> Advisor); die fruehere Multi-Collection-Logik liegt jetzt im Retriever.
 *
- * Fehler je Collection werden geschluckt (graceful: Antwort ohne diesen Treffer).
+ * Fehler werden geschluckt (graceful: Antwort ohne RAG-Kontext).
 * Fundstellen via article_label sind live ab dem Prod-Re-Ingest 2026-06.
 */
@@ -17,16 +18,6 @@ const DEFAULT_USER = '00000000-0000-0000-0000-000000000001'
 const DEFAULT_TENANT =
  process.env.DEFAULT_TENANT_ID || '9282a473-5c95-4b3a-bf78-0ecc0ec71d3e'
 // Compliance-relevante Collections (ai-sdk-Whitelist `AllowedCollections`).
 export const COMPLIANCE_COLLECTIONS = [
  'bp_compliance_gesetze',
  'bp_compliance_ce',
  'bp_compliance_datenschutz',
  'bp_dsfa_corpus',
  'bp_compliance_recht',
  'bp_legal_templates',
 ] as const
 interface SdkRagResult {
  text?: string
  regulation_code?: string
@@ -68,39 +59,36 @@ export function mapSdkResults(results: SdkRagResult[] | undefined): ScoredPassag
    .filter((p) => p.content)
 }
-async function searchCollection(collection: string, query: string): Promise<ScoredPassage[]> {
+/**
 * Authority Router: EIN collection-agnostischer Aufruf an die ai-sdk (`/sdk/v1/rag/retrieve`).
 * Der Router waehlt die Collections (Broad-Authority-Base + KB-2026.1-Slice bei in-scope),
 * merged + authority-ranked sie und liefert die Top-Passagen. Der Advisor weiss damit nichts
 * mehr ueber einzelne Collections — die fruehere Multi-Collection-Logik liegt jetzt im Retriever.
 * Fehler werden geschluckt (graceful: Antwort ohne RAG-Kontext).
 */
 export async function queryAdvisorRAG(query: string): Promise<string> {
  let passages: ScoredPassage[] = []
  try {
-    const res = await fetch(`${SDK_URL}/sdk/v1/rag/search`, {
+    const res = await fetch(`${SDK_URL}/sdk/v1/rag/retrieve`, {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'X-User-ID': DEFAULT_USER,
        'X-Tenant-ID': DEFAULT_TENANT,
      },
-      body: JSON.stringify({ query, collection, top_k: 3 }),
+      body: JSON.stringify({ query, top_k: 8 }),
-      signal: AbortSignal.timeout(10000),
+      signal: AbortSignal.timeout(15000),
    })
-    if (!res.ok) return []
+    if (res.ok) {
-    const data = await res.json()
+      const data = await res.json()
-    return mapSdkResults(data.results)
+      passages = mapSdkResults(data.results)
    }
  } catch {
-    return []
+    // graceful: keine Verbindung -> Antwort ohne RAG-Kontext
  }
-}
+  // Der Router liefert bereits authority-geordnete Top-K; Reihenfolge bewahren.
-
+  if (passages.length === 0) return ''
-/**
+  return passages
 * Fragt alle Compliance-Collections parallel ab und liefert die Top-8-Passagen
 * als formatierten Kontextblock (oder '' wenn nichts erreichbar/gefunden).
 */
 export async function queryAdvisorRAG(query: string): Promise<string> {
  const settled = await Promise.all(
    COMPLIANCE_COLLECTIONS.map((c) => searchCollection(c, query)),
  )
  const all = settled.flat()
  if (all.length === 0) return ''
  all.sort((a, b) => b.score - a.score)
  return all
    .slice(0, 8)
    .map((r, i) => `[Quelle ${i + 1}: ${r.source}]\n${r.content}`)
    .join('\n\n---\n\n')
 }
@@ -82,6 +82,43 @@ func (h *RAGHandlers) Search(c *gin.Context) {
 	})
 }
 // RetrieveRequest is the Authority Router request: a query only, no collection — the router decides
 // which collections to query (broad authority base + the in-scope KB-2026.1 slice).
 type RetrieveRequest struct {
 	Query string `json:"query" binding:"required"`
 	TopK  int    `json:"top_k,omitempty"`
 }
 // Retrieve is the Authority Router endpoint. The Advisor calls this with ONLY a query and stays
 // collection-agnostic; the router fans out over the authority base + the in-scope slice, merges by
 // authority score, and returns the unified top-K. Response shape matches Search (query/results/
 // count/assessment) so existing consumers parse it unchanged.
 // POST /sdk/v1/rag/retrieve
 func (h *RAGHandlers) Retrieve(c *gin.Context) {
 	var req RetrieveRequest
 	if err := c.ShouldBindJSON(&req); err != nil {
 		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
 		return
 	}
 	if req.TopK <= 0 || req.TopK > 20 {
 		req.TopK = 8
 	}
 	results, err := h.ragClient.Retrieve(c.Request.Context(), req.Query, req.TopK)
 	if err != nil {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": "RAG retrieve failed: " + err.Error()})
 		return
 	}
 	c.JSON(http.StatusOK, gin.H{
 		"query":      req.Query,
 		"results":    results,
 		"count":      len(results),
 		"assessment": ucca.Assess(results),
 	})
 }
 // ListRegulations returns the list of available regulations in the corpus.
 // GET /sdk/v1/rag/regulations
 func (h *RAGHandlers) ListRegulations(c *gin.Context) {
@@ -159,6 +159,7 @@ func registerRAGRoutes(v1 *gin.RouterGroup, h *handlers.RAGHandlers) {
 	ragRoutes := v1.Group("/rag")
 	{
 		ragRoutes.POST("/search", h.Search)
 		ragRoutes.POST("/retrieve", h.Retrieve)
 		ragRoutes.GET("/regulations", h.ListRegulations)
 		ragRoutes.GET("/corpus-status", h.CorpusStatus)
 		ragRoutes.GET("/corpus-versions/:collection", h.CorpusVersionHistory)
@@ -358,7 +359,6 @@ func registerWhistleblowerRoutes(v1 *gin.RouterGroup, h *handlers.WhistleblowerH
 	}
 }
 func registerMaximizerRoutes(v1 *gin.RouterGroup, h *handlers.MaximizerHandlers) {
 	m := v1.Group("/maximizer")
 	{
@@ -0,0 +1,73 @@
 package iace
 // P3: pin accepted proposer decisions into the GT gate.
 //
 // When a human accepts a proposal from the offline proposer (a dedup
 // supersession, a foreign-framing gate, a vocab→tag mapping, a coverage hazard),
 // they record an AcceptedPin. A pin is a tiny, machine-scoped invariant — "this
 // pattern MUST (or must NOT) fire for this machine" — that a test re-checks on
 // every run. This is what makes the library's growth COMPOUND into the gate
 // instead of silently eroding it: a future change that re-introduces a dropped
 // duplicate, un-gates a foreign pattern, or removes a coverage hazard breaks the
 // pin and fails CI.
 //
 // A single boolean covers all four proposal types:
 //   - dedup supersession accepted → DropPattern MustFire=false
 //   - foreign-framing gate accepted → foreign pattern MustFire=false
 //   - vocab→tag / coverage hazard accepted → the enabled pattern MustFire=true
 // AcceptedPin is one regression invariant for an accepted proposal.
 type AcceptedPin struct {
 	Pattern      string `json:"pattern"`
 	MustFire     bool   `json:"must_fire"`
 	Reason       string `json:"reason"`
 	FromProposal string `json:"from_proposal,omitempty"`
 }
 // PinSet is the accepted-pin registry for one machine (testdata/accepted_pins_*.json).
 type PinSet struct {
 	Machine string        `json:"machine"`
 	Pins    []AcceptedPin `json:"pins"`
 }
 // PinResult is the verdict for one pin against an engine run.
 type PinResult struct {
 	Pin    AcceptedPin
 	OK     bool
 	Detail string
 }
 // VerifyPins checks every pin against the set of pattern IDs the engine actually
 // fired for the machine. A pin holds iff the pattern's presence equals MustFire.
 func VerifyPins(pins []AcceptedPin, firedPatternIDs []string) []PinResult {
 	fired := make(map[string]bool, len(firedPatternIDs))
 	for _, id := range firedPatternIDs {
 		fired[id] = true
 	}
 	out := make([]PinResult, 0, len(pins))
 	for _, p := range pins {
 		got := fired[p.Pattern]
 		ok := got == p.MustFire
 		detail := "ok"
 		if !ok {
 			if p.MustFire {
 				detail = "expected to fire but did NOT — coverage/mapping regressed"
 			} else {
 				detail = "expected to be suppressed but FIRED — gate/supersession regressed"
 			}
 		}
 		out = append(out, PinResult{Pin: p, OK: ok, Detail: detail})
 	}
 	return out
 }
 // GenerateDedupPin turns an accepted (verdict=duplicate) dedup candidate into the
 // pin that protects the supersession: the dropped pattern must no longer fire.
 func GenerateDedupPin(c DedupCandidate) AcceptedPin {
 	return AcceptedPin{
 		Pattern:      c.DropPattern,
 		MustFire:     false,
 		Reason:       "accepted duplicate of " + c.KeepPattern + " (" + c.Category + ")",
 		FromProposal: "dedup " + c.DropPattern + " -> " + c.KeepPattern,
 	}
 }
@@ -0,0 +1,63 @@
 package iace
 import (
 	"encoding/json"
 	"os"
 	"path/filepath"
 	"testing"
 )
 func TestVerifyPins(t *testing.T) {
 	pins := []AcceptedPin{
 		{Pattern: "HPa", MustFire: true},
 		{Pattern: "HPb", MustFire: false},
 	}
 	res := VerifyPins(pins, []string{"HPa", "HPb"})
 	if !res[0].OK {
 		t.Errorf("HPa must_fire=true and it fired -> should be OK")
 	}
 	if res[1].OK {
 		t.Errorf("HPb must_fire=false but it fired -> should be VIOLATED")
 	}
 	res2 := VerifyPins(pins, []string{})
 	if res2[0].OK || !res2[1].OK {
 		t.Errorf("expected HPa violated + HPb ok, got %+v", res2)
 	}
 }
 func TestGenerateDedupPin(t *testing.T) {
 	pin := GenerateDedupPin(DedupCandidate{KeepPattern: "HP144", DropPattern: "HP013", Category: "electrical_hazard"})
 	if pin.Pattern != "HP013" || pin.MustFire {
 		t.Fatalf("want pin {HP013, must_fire=false}, got %+v", pin)
 	}
 }
 // TestWarewashing_AcceptedPins re-checks every accepted P1 supersession against the
 // live warewashing engine output. A future change that un-suppresses HP013/016/018
 // or drops HP2201/HP144 breaks a pin here — the gate compounds, not erodes.
 func TestWarewashing_AcceptedPins(t *testing.T) {
 	raw, err := os.ReadFile(filepath.Join("testdata", "accepted_pins_warewashing.json"))
 	if err != nil {
 		t.Fatalf("read pins: %v", err)
 	}
 	var ps PinSet
 	if err := json.Unmarshal(raw, &ps); err != nil {
 		t.Fatalf("parse pins: %v", err)
 	}
 	_, _, kept := warewashingEngineOutput()
 	firedIDs := make([]string, 0, len(kept))
 	for _, pm := range kept {
 		firedIDs = append(firedIDs, pm.PatternID)
 	}
 	ok := 0
 	for _, r := range VerifyPins(ps.Pins, firedIDs) {
 		if r.OK {
 			ok++
 			continue
 		}
 		t.Errorf("PIN VIOLATED: %s (must_fire=%v) — %s [%s]", r.Pin.Pattern, r.Pin.MustFire, r.Detail, r.Pin.Reason)
 	}
 	t.Logf("accepted pins for %q: %d/%d hold", ps.Machine, ok, len(ps.Pins))
 }
@@ -0,0 +1,10 @@
 {
  "machine": "Gewerbliche Untertisch-Geschirrspuelmaschine (vernetzt)",
  "pins": [
    {"pattern": "HP016", "must_fire": false, "reason": "generic hot-surface (Formwerkzeuge/Auspuffleitung framing) superseded by HP2201", "from_proposal": "P1 thermal supersession"},
    {"pattern": "HP018", "must_fire": false, "reason": "actuator-burn superseded by HP2201", "from_proposal": "P1 thermal supersession"},
    {"pattern": "HP013", "must_fire": false, "reason": "stored-energy Batterie/USV framing superseded by HP144", "from_proposal": "P1 stored-energy supersession"},
    {"pattern": "HP2201", "must_fire": true, "reason": "warewashing hot-surface (Boiler/Tank/Spuelkammer) must remain — it is the clean equivalent that replaces HP016/HP018", "from_proposal": "P1 thermal supersession"},
    {"pattern": "HP144", "must_fire": true, "reason": "residual-voltage (Frequenzumrichter/Zwischenkreis) must remain — clean equivalent that replaces HP013", "from_proposal": "P1 stored-energy supersession"}
  ]
 }
@@ -0,0 +1,129 @@
 package ucca
 import (
 	"context"
 	"os"
 	"sort"
 	"strings"
 	"sync"
 )
 // routerBaseCollections is the broad authority base the Authority Router fans out over. It mirrors
 // the Advisor's historical multi-collection set; the KB-2026.1 slice is added separately when the
 // query is in scope. Override via RAG_ROUTER_COLLECTIONS (comma-separated) per environment.
 func (c *LegalRAGClient) routerBaseCollections() []string {
 	if v := strings.TrimSpace(os.Getenv("RAG_ROUTER_COLLECTIONS")); v != "" {
 		var out []string
 		for _, p := range strings.Split(v, ",") {
 			if s := strings.TrimSpace(p); s != "" {
 				out = append(out, s)
 			}
 		}
 		if len(out) > 0 {
 			return out
 		}
 	}
 	return []string{
 		"bp_compliance_gesetze",
 		"bp_compliance_ce",
 		"bp_compliance_datenschutz",
 		"bp_dsfa_corpus",
 		"bp_compliance_recht",
 		"bp_legal_templates",
 	}
 }
 const routerPerCollectionTopK = 3
 // Retrieve is the Authority Router entry point: callers (the Advisor) pass ONLY a query and stay
 // collection-agnostic. The router fans out over the broad authority base and ADDS the KB-2026.1
 // slice when the query is in scope (inKBScope), then merges all hits, deduplicates, and returns the
 // top-K by authority score. This moves the former Advisor-side collection fan-out into the retrieval
 // layer (the "Retriever" tier of the quality pyramid), so the proven KB-2026.1 slice gain reaches
 // the product path without the Advisor knowing about individual collections.
 //
 // The merged set is ordered by the per-collection authority score that rerankByAuthority already
 // produced inside searchInternal — i.e. binding-vs-guidance ordering is preserved across the merge.
 // Per-collection failures (e.g. a collection absent on an environment) degrade gracefully.
 func (c *LegalRAGClient) Retrieve(ctx context.Context, query string, topK int) ([]LegalSearchResult, error) {
 	if topK <= 0 {
 		topK = 8
 	}
 	collections := c.routerBaseCollections()
 	if c.kbScopeRoutingEnabled && c.kbSliceCollection != "" && inKBScope(query) {
 		collections = append(collections, c.kbSliceCollection)
 	}
 	// Cross-regulation queries (>=2 explicitly named regulations) get a larger per-collection budget
 	// so each collection's multi-regulation search isn't truncated down to the keyword-dominant
 	// domain; the final per-regulation balancing then guarantees every named domain in the top-K.
 	regs := detectRegulations(query)
 	perColl := routerPerCollectionTopK
 	if len(regs) >= 2 {
 		perColl = routerPerCollectionTopK * len(regs)
 	}
 	// Warm the full-text indexes sequentially first so the concurrent fan-out below only READS the
 	// shared textIndexEnsured map (the writes happen here, serialized) — closes the cold-start map
 	// race deterministically. Best-effort: a missing collection just stays un-indexed (hybrid then
 	// falls back to dense, or the per-collection search degrades to nothing).
 	if c.hybridEnabled {
 		for _, coll := range collections {
 			_ = c.ensureTextIndex(ctx, coll)
 		}
 	}
 	out := make([][]LegalSearchResult, len(collections))
 	var wg sync.WaitGroup
 	for i, coll := range collections {
 		wg.Add(1)
 		go func(i int, coll string) {
 			defer wg.Done()
 			if res, err := c.searchInternal(ctx, coll, query, nil, perColl); err == nil {
 				out[i] = res
 			}
 		}(i, coll)
 	}
 	wg.Wait()
 	merged := make([]LegalSearchResult, 0, len(collections)*perColl)
 	for _, r := range out {
 		merged = append(merged, r...)
 	}
 	merged = dedupResults(merged)
 	sort.SliceStable(merged, func(a, b int) bool { return merged[a].Score > merged[b].Score })
 	// Cross-regulation: guarantee every named domain is represented (0070-class fix) instead of
 	// letting a global score-sort starve the non-dominant domain.
 	if len(regs) >= 2 {
 		return balanceByRegulation(merged, regs, topK), nil
 	}
 	if len(merged) > topK {
 		merged = merged[:topK]
 	}
 	return merged, nil
 }
 // dedupResults removes duplicate passages that can appear when collections overlap, keeping the
 // highest-scoring occurrence. Identity = regulation_code + article_label + a text prefix.
 func dedupResults(in []LegalSearchResult) []LegalSearchResult {
 	pos := make(map[string]int, len(in))
 	out := make([]LegalSearchResult, 0, len(in))
 	for _, r := range in {
 		text := r.Text
 		if len(text) > 80 {
 			text = text[:80]
 		}
 		key := r.RegulationCode + "|" + r.ArticleLabel + "|" + text
 		if idx, ok := pos[key]; ok {
 			if r.Score > out[idx].Score {
 				out[idx] = r
 			}
 			continue
 		}
 		pos[key] = len(out)
 		out = append(out, r)
 	}
 	return out
 }
@@ -0,0 +1,164 @@
 package ucca
 import (
 	"context"
 	"encoding/json"
 	"os"
 	"strconv"
 	"strings"
 	"testing"
 )
 type benchQ struct {
 	ID       string `json:"id"`
 	Document string `json:"document"`
 	Question string `json:"question"`
 }
 // docTokens maps a bench question's expected document to acceptable regulation_code/label substrings.
 func docTokens(document string) []string {
 	d := strings.ToUpper(document)
 	var t []string
 	for _, wp := range []string{"WP243", "WP248", "WP260"} {
 		if strings.Contains(d, wp) {
 			t = append(t, wp)
 		}
 	}
 	dns := strings.ReplaceAll(d, " ", "")
 	for _, gl := range []struct{ key, tok string }{{"07/2020", "GL07"}, {"05/2020", "GL05"}, {"09/2022", "GL09"}} {
 		if strings.Contains(dns, gl.key) {
 			t = append(t, gl.tok)
 		}
 	}
 	if strings.Contains(d, "TDDDG") {
 		t = append(t, "TDDDG")
 	}
 	if strings.Contains(d, "DSGVO") || strings.Contains(d, "ART. 13") || strings.Contains(d, "ART. 14") {
 		t = append(t, "DSGVO")
 	}
 	if strings.Contains(d, "BDSG") {
 		t = append(t, "BDSG")
 	}
 	if strings.Contains(d, "CRA") {
 		t = append(t, "CRA")
 	}
 	if strings.Contains(d, "MASCH") {
 		t = append(t, "MASCH", "MACHINERY", "MVO")
 	}
 	return t
 }
 func hitDoc(results []LegalSearchResult, toks []string) bool {
 	for _, r := range results {
 		s := strings.ReplaceAll(strings.ToUpper(r.RegulationCode+" "+r.ArticleLabel), " ", "")
 		for _, tk := range toks {
 			if strings.Contains(s, strings.ReplaceAll(tk, " ", "")) {
 				return true
 			}
 		}
 	}
 	return false
 }
 // TestMultiReg0070E2E (RUN_E2E=1) is the 0070 regression: a cross-regulation query (CRA + MaschVO)
 // must return BOTH domains through the real Retrieve(), not just the keyword-dominant CRA.
 func TestMultiReg0070E2E(t *testing.T) {
 	if os.Getenv("RUN_E2E") != "1" {
 		t.Skip("set RUN_E2E=1 + QDRANT_URL/OLLAMA_URL/QDRANT_API_KEY")
 	}
 	c := NewLegalRAGClient()
 	q := "Wie greifen CRA und Maschinenverordnung bei einer vernetzten Maschine ineinander?"
 	res, err := c.Retrieve(context.Background(), q, 8)
 	if err != nil {
 		t.Fatalf("retrieve: %v", err)
 	}
 	var hasCRA, hasMasch bool
 	var codes []string
 	for _, r := range res {
 		u := strings.ToUpper(r.RegulationCode)
 		codes = append(codes, u)
 		if strings.Contains(u, "CRA") {
 			hasCRA = true
 		}
 		if strings.Contains(u, "MASCH") || strings.Contains(u, "MACHIN") || u == "MVO" {
 			hasMasch = true
 		}
 	}
 	t.Logf("0070 top-8 codes: %v", codes)
 	if !hasCRA || !hasMasch {
 		t.Errorf("0070 must return BOTH domains via Retrieve(): CRA=%v MaschVO=%v", hasCRA, hasMasch)
 	}
 }
 // TestAuthorityRouterCB100 (RUN_E2E=1) drives the REAL Retrieve() over the ComplianceBench-100 against
 // the live collections: NEW (scope routing on → slice added for in-scope queries) vs OLD (routing off
 // → broad base only). It is the regression gate that the router actually delivers the proven slice
 // gain (+28/0-regr in the offline simulation) through the production Go code path.
 func TestAuthorityRouterCB100(t *testing.T) {
 	if os.Getenv("RUN_E2E") != "1" {
 		t.Skip("set RUN_E2E=1 + QDRANT_URL/OLLAMA_URL/QDRANT_API_KEY + BENCH_PATH")
 	}
 	path := os.Getenv("BENCH_PATH")
 	if path == "" {
 		path = "/tmp/compliance_bench.json"
 	}
 	raw, err := os.ReadFile(path)
 	if err != nil {
 		t.Fatalf("bench read: %v", err)
 	}
 	var doc struct {
 		Questions []benchQ `json:"questions"`
 	}
 	if err := json.Unmarshal(raw, &doc); err != nil {
 		t.Fatalf("bench parse: %v", err)
 	}
 	// BENCH_STRIDE samples every Kth question (stratified across DS/CRA/MaschVO) so the gate stays
 	// tractable against the remote dev Qdrant; default 1 = full CB-100.
 	stride := 1
 	if s := os.Getenv("BENCH_STRIDE"); s != "" {
 		if n, err := strconv.Atoi(s); err == nil && n > 0 {
 			stride = n
 		}
 	}
 	c := NewLegalRAGClient()
 	ctx := context.Background()
 	var n, oldHit, newHit, gain, regr int
 	for i, q := range doc.Questions {
 		if i%stride != 0 {
 			continue
 		}
 		n++
 		toks := docTokens(q.Document)
 		c.kbScopeRoutingEnabled = false
 		oldRes, _ := c.Retrieve(ctx, q.Question, 8)
 		c.kbScopeRoutingEnabled = true
 		newRes, _ := c.Retrieve(ctx, q.Question, 8)
 		oh, nh := hitDoc(oldRes, toks), hitDoc(newRes, toks)
 		if oh {
 			oldHit++
 		}
 		if nh {
 			newHit++
 		}
 		flip := "="
 		switch {
 		case !oh && nh:
 			gain++
 			flip = "GAIN"
 		case oh && !nh:
 			regr++
 			flip = "REGR"
 		}
 		t.Logf("%-9s [%-14s] OLD=%-5v NEW=%-5v %s", q.ID, q.Document, oh, nh, flip)
 	}
 	t.Logf("CB-100 sample (stride=%d) via Retrieve(): N=%d | OLD-hit %d | NEW-hit %d | GAIN %d | REGR %d",
 		stride, n, oldHit, newHit, gain, regr)
 	if newHit <= oldHit || gain < 3 {
 		t.Errorf("router must add slice gains: NEW(%d) must exceed OLD(%d), gain=%d", newHit, oldHit, gain)
 	}
 	if regr > 2 {
 		t.Errorf("too many regressions through the router: %d", regr)
 	}
 }
@@ -0,0 +1,99 @@
 package ucca
 import (
 	"os"
 	"testing"
 )
 func TestRouterBaseCollections(t *testing.T) {
 	c := &LegalRAGClient{}
 	os.Unsetenv("RAG_ROUTER_COLLECTIONS")
 	def := c.routerBaseCollections()
 	if len(def) != 6 || def[1] != "bp_compliance_ce" {
 		t.Fatalf("default base collections unexpected: %v", def)
 	}
 	os.Setenv("RAG_ROUTER_COLLECTIONS", " bp_compliance_ce , kb_2026_1_build ,, ")
 	defer os.Unsetenv("RAG_ROUTER_COLLECTIONS")
 	got := c.routerBaseCollections()
 	if len(got) != 2 || got[0] != "bp_compliance_ce" || got[1] != "kb_2026_1_build" {
 		t.Fatalf("env override parse failed (trim/empty): %v", got)
 	}
 }
 func TestRouterSliceSelection(t *testing.T) {
 	// The router appends the slice exactly when the query is in scope (inKBScope) and routing is on.
 	// Mirror the selection logic so a regression in either is caught without a live Qdrant.
 	c := &LegalRAGClient{kbSliceCollection: "kb_2026_1_build", kbScopeRoutingEnabled: true}
 	sel := func(q string) bool {
 		colls := c.routerBaseCollections()
 		if c.kbScopeRoutingEnabled && c.kbSliceCollection != "" && inKBScope(q) {
 			colls = append(colls, c.kbSliceCollection)
 		}
 		for _, x := range colls {
 			if x == c.kbSliceCollection {
 				return true
 			}
 		}
 		return false
 	}
 	if !sel("Welche neun Kriterien nennt WP248 fuer ein hohes Risiko?") {
 		t.Error("in-scope guidance query must include the slice")
 	}
 	if sel("Was sagt NIST SP 800-53 zu Access Control?") {
 		t.Error("out-of-scope query must NOT include the slice")
 	}
 	c.kbScopeRoutingEnabled = false
 	if sel("Welche Kriterien nennt WP248?") {
 		t.Error("routing disabled => slice never included")
 	}
 }
 func TestBalanceByRegulation(t *testing.T) {
 	regs := []detectedRegulation{
 		{Canonical: "CRA", CodeValues: []string{"CRA"}},
 		{Canonical: "MaschVO", CodeValues: []string{"MASCHVO", "MVO", "MACHINERY"}},
 	}
 	// CRA dominates by score; without balancing the top-4 would be all CRA + NIST.
 	pool := []LegalSearchResult{
 		{RegulationCode: "CRA", Score: 0.99},
 		{RegulationCode: "CRA", Score: 0.98},
 		{RegulationCode: "CRA", Score: 0.97},
 		{RegulationCode: "NIST", Score: 0.96},
 		{RegulationCode: "MACHINERY", Score: 0.70},
 		{RegulationCode: "MVO", Score: 0.65},
 	}
 	out := balanceByRegulation(pool, regs, 4)
 	var hasCRA, hasMasch bool
 	for _, r := range out {
 		switch r.RegulationCode {
 		case "CRA":
 			hasCRA = true
 		case "MACHINERY", "MVO":
 			hasMasch = true
 		}
 	}
 	if !hasCRA || !hasMasch {
 		t.Errorf("both named domains must be represented: CRA=%v MaschVO=%v out=%v", hasCRA, hasMasch, out)
 	}
 	if out[0].RegulationCode != "CRA" || !(out[1].RegulationCode == "MACHINERY" || out[1].RegulationCode == "MVO") {
 		t.Errorf("round-robin should alternate domains, got %s then %s", out[0].RegulationCode, out[1].RegulationCode)
 	}
 }
 func TestDedupResults(t *testing.T) {
 	in := []LegalSearchResult{
 		{RegulationCode: "EDPB WP248", ArticleLabel: "III.B", Text: "lorem", Score: 0.7},
 		{RegulationCode: "EDPB WP248", ArticleLabel: "III.B", Text: "lorem", Score: 0.9}, // dup, higher score
 		{RegulationCode: "DSGVO", ArticleLabel: "Art. 35", Text: "ipsum", Score: 0.8},
 	}
 	out := dedupResults(in)
 	if len(out) != 2 {
 		t.Fatalf("expected 2 deduped, got %d", len(out))
 	}
 	for _, r := range out {
 		if r.RegulationCode == "EDPB WP248" && r.Score != 0.9 {
 			t.Errorf("dedup must keep highest score, got %v", r.Score)
 		}
 	}
 }
@@ -0,0 +1,52 @@
 package ucca
 import "strings"
 // kbScopeTopics are high-precision data-protection / compliance topic markers that place a query in
 // the KB-2026.1 authoritative slice even when it does NOT name a regulation. Conservative by design:
 // an unmatched query falls back to the broad CE default (no regression) — the slice is only used when
 // the query is confidently in-scope.
 var kbScopeTopics = []string{
 	// DP-Guidance-Marker, die IN der Slice liegen (EDPB/DSK/WP/GL) — bewusst NICHT die generischen
 	// Verben aus guidanceIntentSignals (sagt/laut/empfiehlt/auslegung) und NICHT enisa/bsi/nist/owasp
 	// (die liegen im breiten CE-Pool, nicht in der Slice).
 	"edpb", "dsk", "datenschutzausschuss", "orientierungshilfe",
 	"wp2", "wp 2", "wp29", "working paper", "gl 0",
 	"datenschutz", "dsgvo", "gdpr", "dsfa", "folgenabschätzung", "folgenabschaetzung",
 	"einwilligung", "auftragsverarbeit", "betroffenenrecht", "auskunftsrecht",
 	"verarbeitungsverzeichnis", "datenschutzbeauftragt", "verzeichnis von verarbeitung",
 	"cookie", "tracking", "transparenzpflicht", "datenpanne", "meldepflicht",
 	"technische und organisatorische maßnahmen",
 	"cyber resilience", "schwachstelle", "vulnerability", "sicherheitsupdate",
 	"maschinensicherheit", "wesentliche veränderung", "wesentliche veraenderung",
 	"konformitätsbewertung", "konformitaetsbewertung", "ce-kennzeichnung",
 }
 // inKBScope reports whether the query belongs to the KB-2026.1 authoritative slice. True when it
 // names an in-slice regulation (detectRegulations), asks for guidance (EDPB/DSK/WP/GL), or hits a
 // data-protection / compliance topic marker.
 func inKBScope(query string) bool {
 	if len(detectRegulations(query)) > 0 {
 		return true
 	}
 	q := strings.ToLower(query)
 	for _, t := range kbScopeTopics {
 		if strings.Contains(q, t) {
 			return true
 		}
 	}
 	return false
 }
 // resolveCollection applies the Blue-Green „authoritative slice promotion" routing. An explicitly
 // requested collection is honoured unchanged; the DEFAULT (empty) request is routed to the KB-2026.1
 // slice when the query is in-scope, else to the broad CE default. Disable via RAG_KB_SCOPE_ROUTING=false.
 func (c *LegalRAGClient) resolveCollection(query, requested string) string {
 	if requested != "" {
 		return requested
 	}
 	if c.kbScopeRoutingEnabled && c.kbSliceCollection != "" && inKBScope(query) {
 		return c.kbSliceCollection
 	}
 	return c.collection
 }
@@ -0,0 +1,101 @@
 package ucca
 import (
 	"context"
 	"fmt"
 	"os"
 	"strings"
 	"testing"
 )
 func TestInKBScope(t *testing.T) {
 	inScope := []string{
 		"Welche neun Kriterien nennt WP248 fuer ein hohes Risiko?",
 		"Wie greifen CRA und Maschinenverordnung bei einer vernetzten Maschine ineinander?",
 		"Wann ist eine Datenschutz-Folgenabschaetzung erforderlich?",
 		"Welche Anforderungen stellt die DSGVO an die Einwilligung?",
 		"Brauche ich einen Datenschutzbeauftragten?",
 		"Wann muss eine aktiv ausgenutzte Schwachstelle gemeldet werden?",
 	}
 	outScope := []string{
 		"Welche OWASP-Kontrollen gibt es fuer Authentifizierung?",
 		"Was sagt NIST SP 800-53 zu Access Control?",
 		"Wie funktioniert ISO 27001 Zertifizierung?",
 		"Welche IFRS-Standards gelten fuer Leasing?",
 	}
 	for _, q := range inScope {
 		if !inKBScope(q) {
 			t.Errorf("inKBScope(%q) = false, want true", q)
 		}
 	}
 	for _, q := range outScope {
 		if inKBScope(q) {
 			t.Errorf("inKBScope(%q) = true, want false", q)
 		}
 	}
 }
 func TestResolveCollection(t *testing.T) {
 	c := &LegalRAGClient{collection: "bp_compliance_ce", kbSliceCollection: "kb_2026_1_build", kbScopeRoutingEnabled: true}
 	if got := c.resolveCollection("Welche Kriterien nennt WP248?", ""); got != "kb_2026_1_build" {
 		t.Errorf("in-scope default -> %s, want kb_2026_1_build", got)
 	}
 	if got := c.resolveCollection("Was sagt NIST SP 800-53?", ""); got != "bp_compliance_ce" {
 		t.Errorf("out-of-scope default -> %s, want bp_compliance_ce", got)
 	}
 	if got := c.resolveCollection("Welche Kriterien nennt WP248?", "explicit_coll"); got != "explicit_coll" {
 		t.Errorf("explicit request must be honoured -> %s", got)
 	}
 	c.kbScopeRoutingEnabled = false
 	if got := c.resolveCollection("Welche Kriterien nennt WP248?", ""); got != "bp_compliance_ce" {
 		t.Errorf("disabled routing -> %s, want bp_compliance_ce", got)
 	}
 }
 // TestKBScopeRoutingE2E (RUN_E2E=1) verifies the routing against the REAL collections: a default
 // Search() of an in-scope query must hit the KB-2026.1 slice (WP248/MaschVO live there but NOT in
 // the broad CE pool = clean discriminator); an out-of-scope query stays on CE.
 func TestKBScopeRoutingE2E(t *testing.T) {
 	if os.Getenv("RUN_E2E") != "1" {
 		t.Skip("set RUN_E2E=1 + QDRANT_URL/OLLAMA_URL/QDRANT_API_KEY")
 	}
 	c := NewLegalRAGClient()
 	cases := []struct {
 		q         string
 		wantToken string // expected in top-8 when routed to the slice
 		wantInKB  bool
 	}{
 		{"Welche neun Kriterien nennt WP248 fuer ein voraussichtlich hohes Risiko?", "WP248", true},
 		{"Welche grundlegenden Sicherheits- und Gesundheitsschutzanforderungen enthaelt Anhang III der Maschinenverordnung?", "MASCH", true},
 		{"Wie greifen CRA und Maschinenverordnung bei einer vernetzten Maschine ineinander?", "MASCH", true},
 		{"Was sagt NIST SP 800-53 zu Access Control?", "", false},
 	}
 	for _, tc := range cases {
 		routed := c.resolveCollection(tc.q, "")
 		res, err := c.Search(context.Background(), tc.q, nil, 8)
 		if err != nil {
 			t.Fatalf("%q: %v", tc.q, err)
 		}
 		codes := map[string]bool{}
 		for _, r := range res {
 			codes[strings.ToUpper(r.RegulationCode)] = true
 		}
 		hit := false
 		if tc.wantToken != "" {
 			for cd := range codes {
 				if strings.Contains(cd, tc.wantToken) {
 					hit = true
 					break
 				}
 			}
 		}
 		col := make([]string, 0, len(codes))
 		for cd := range codes {
 			col = append(col, cd)
 		}
 		fmt.Printf("inKB=%-5v routed=%-16s wantTok=%-6s found=%-5v | %v\n", tc.wantInKB, routed, tc.wantToken, hit, col)
 		if tc.wantInKB && tc.wantToken != "" && !hit {
 			t.Errorf("%q routed to %s but %s not in top-8 (slice not active?)", tc.q, routed, tc.wantToken)
 		}
 	}
 }
@@ -21,6 +21,12 @@ type LegalRAGClient struct {
 	textIndexEnsured map[string]bool
 	hybridEnabled    bool
 	graphEnabled     bool
 	// Blue-Green „authoritative slice promotion" (additiv, KEIN CE-Ersatz): faellt eine Query
 	// in den KB-2026.1-Scope (DP/CRA/MaschVO/NIS2/DataAct/DORA/AIAct + EDPB/DSK-Guidance), wird
 	// die hochwertige Slice-Collection abgefragt; sonst bleibt der breite Default (bp_compliance_ce).
 	kbSliceCollection     string
 	kbScopeRoutingEnabled bool
 }
 // NewLegalRAGClient creates a new Legal RAG client using Ollama bge-m3 embeddings.
@@ -45,15 +51,25 @@ func NewLegalRAGClient() *LegalRAGClient {
 	// zur Begruendung/Vollstaendigkeit genutzt, nicht zur Pool-Expansion (Default).
 	graphEnabled := os.Getenv("RAG_GRAPH_EXPANSION") == "true"
 	// KB-2026.1 authoritative slice (Blue-Green, additiv). Routing default AN; Rollback ohne
 	// Redeploy ueber RAG_KB_SCOPE_ROUTING=false (dann faellt alles auf den CE-Default zurueck).
 	kbSlice := os.Getenv("RAG_KB_SLICE_COLLECTION")
 	if kbSlice == "" {
 		kbSlice = "kb_2026_1_build"
 	}
 	kbScopeRouting := os.Getenv("RAG_KB_SCOPE_ROUTING") != "false"
 	return &LegalRAGClient{
-		qdrantURL:        qdrantURL,
+		qdrantURL:             qdrantURL,
-		qdrantAPIKey:     qdrantAPIKey,
+		qdrantAPIKey:          qdrantAPIKey,
-		ollamaURL:        ollamaURL,
+		ollamaURL:             ollamaURL,
-		embeddingModel:   "bge-m3",
+		embeddingModel:        "bge-m3",
-		collection:       "bp_compliance_ce",
+		collection:            "bp_compliance_ce",
-		textIndexEnsured: make(map[string]bool),
+		textIndexEnsured:      make(map[string]bool),
-		hybridEnabled:    hybridEnabled,
+		hybridEnabled:         hybridEnabled,
-		graphEnabled:     graphEnabled,
+		graphEnabled:          graphEnabled,
 		kbSliceCollection:     kbSlice,
 		kbScopeRoutingEnabled: kbScopeRouting,
 		httpClient: &http.Client{
 			Timeout: 60 * time.Second,
 		},
@@ -63,15 +79,13 @@ func NewLegalRAGClient() *LegalRAGClient {
 // SearchCollection queries a specific Qdrant collection for relevant passages.
 // If collection is empty, it falls back to the default collection (bp_compliance_ce).
 func (c *LegalRAGClient) SearchCollection(ctx context.Context, collection string, query string, regulationIDs []string, topK int) ([]LegalSearchResult, error) {
-	if collection == "" {
+	return c.searchInternal(ctx, c.resolveCollection(query, collection), query, regulationIDs, topK)
 		collection = c.collection
 	}
 	return c.searchInternal(ctx, collection, query, regulationIDs, topK)
 }
-// Search queries the compliance CE corpus for relevant passages.
+// Search queries the compliance corpus for relevant passages. The target collection is resolved by
 // the Blue-Green slice routing: the KB-2026.1 slice for in-scope queries, else the broad CE default.
 func (c *LegalRAGClient) Search(ctx context.Context, query string, regulationIDs []string, topK int) ([]LegalSearchResult, error) {
-	return c.searchInternal(ctx, c.collection, query, regulationIDs, topK)
+	return c.searchInternal(ctx, c.resolveCollection(query, ""), query, regulationIDs, topK)
 }
 // searchInternal performs the actual search against a given collection.
@@ -20,7 +20,9 @@ var regulationCatalog = []struct {
 	CodeValues []string
 }{
 	{"CRA", []string{"cra", "cyber resilience"}, []string{"CRA"}},
-	{"MaschVO", []string{"maschinenverordnung", "maschvo", "machinery regulation"}, []string{"MASCHVO", "MaschVO"}},
+	// MaschVO heisst je Collection anders: Slice MASCHVO · gesetze MVO · ce MACHINERY/MASCHINENVO.
 	// Alle Varianten als CodeValues, sonst findet der per-Reg-Filter MaschVO nur in der Slice (0070).
 	{"MaschVO", []string{"maschinenverordnung", "maschvo", "machinery regulation"}, []string{"MASCHVO", "MaschVO", "MVO", "MASCHINENVO", "MACHINERY"}},
 	{"NIS2", []string{"nis2", "nis-2", "nis 2"}, []string{"NIS2"}},
 	{"DORA", []string{"dora"}, []string{"DORA"}},
 	{"Data Act", []string{"data act", "datengesetz"}, []string{"DATA ACT", "DataAct"}},
@@ -53,6 +55,62 @@ func detectRegulations(query string) []detectedRegulation {
 func hitID(h qdrantSearchHit) string { return fmt.Sprintf("%v", h.ID) }
 // balanceByRegulation builds the final top-K so EVERY explicitly-named regulation with hits is
 // represented, instead of letting the keyword-dominant domain (e.g. CRA) crowd out the other
 // (e.g. MaschVO) in a cross-regulation query. The input pool must already be score-ordered;
 // results are grouped by exact regulation_code match against each regulation's CodeValues, then
 // taken round-robin across the named domains (highest-scored first within each), with any
 // remaining slots filled by the leftover pool in score order. Generic; no doc-specific logic.
 func balanceByRegulation(pool []LegalSearchResult, regs []detectedRegulation, topK int) []LegalSearchResult {
 	if topK <= 0 {
 		topK = 8
 	}
 	byReg := make([][]LegalSearchResult, len(regs))
 	matched := make([]bool, len(pool))
 	for ri, r := range regs {
 		for pi := range pool {
 			if matched[pi] {
 				continue
 			}
 			code := strings.TrimSpace(pool[pi].RegulationCode)
 			for _, cv := range r.CodeValues {
 				if strings.EqualFold(code, cv) {
 					byReg[ri] = append(byReg[ri], pool[pi])
 					matched[pi] = true
 					break
 				}
 			}
 		}
 	}
 	out := make([]LegalSearchResult, 0, topK)
 	idx := make([]int, len(regs))
 	for len(out) < topK {
 		progressed := false
 		for ri := range regs {
 			if idx[ri] < len(byReg[ri]) {
 				out = append(out, byReg[ri][idx[ri]])
 				idx[ri]++
 				progressed = true
 				if len(out) >= topK {
 					break
 				}
 			}
 		}
 		if !progressed {
 			break
 		}
 	}
 	for pi := range pool {
 		if len(out) >= topK {
 			break
 		}
 		if !matched[pi] {
 			out = append(out, pool[pi])
 		}
 	}
 	return out
 }
 // searchMultiRegulation retrieves each explicitly-named regulation SEPARATELY (per-regulation
 // filter) and merges, so a cross-regulation query ("Wie greifen CRA und MaschVO ineinander?")
 // returns BOTH domains in the prompt instead of only the keyword-dominant one. Generic over any
@@ -162,7 +162,7 @@ async def update_ai_system(
    db: Session = Depends(get_db),
 ):
    """Update an AI system."""
-    from datetime import datetime
+    from datetime import datetime, timezone
    system = db.query(AISystemDB).filter(AISystemDB.id == system_id).first()
    if not system:
@@ -226,7 +226,7 @@ async def assess_ai_system(
    db: Session = Depends(get_db),
 ):
    """Run AI Act risk assessment for an AI system."""
-    from datetime import datetime
+    from datetime import datetime, timezone
    system = db.query(AISystemDB).filter(AISystemDB.id == system_id).first()
    if not system:
@@ -47,6 +47,8 @@ from compliance.services.canonical_control_service import (
    _control_row,  # re-exported for legacy test imports
 )
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/v1/canonical", tags=["canonical-controls"])
@@ -14,7 +14,7 @@ Endpoints:
 """
 import logging
-from datetime import datetime, date, timedelta
+from datetime import datetime, date, timedelta, timezone
 from calendar import month_abbr
 from typing import Optional, Dict, Any, List
 from decimal import Decimal
@@ -26,10 +26,11 @@ versions). Module-level helpers re-exported for legacy tests.
 import logging
 from typing import Any, List, Optional
-from fastapi import APIRouter, Depends, Query
+from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel
 from fastapi.responses import Response
 from sqlalchemy.orm import Session
 from sqlalchemy import text
 from classroom_engine.database import get_db
 from compliance.api._http_errors import translate_domain_errors
@@ -484,6 +485,7 @@ async def list_dsfas(
 async def create_dsfa(
    request: DSFACreate,
    tenant_id: Optional[str] = Query(None),
    db: Session = Depends(get_db),
    service: DSFAService = Depends(get_dsfa_service),
 ) -> dict[str, Any]:
    """Neue DSFA erstellen."""
@@ -16,6 +16,11 @@ from the legacy path.
 """
 import logging
 import os
 import json
 import hashlib
 import uuid as uuid_module
 from datetime import datetime, timedelta
 from typing import Any, Optional
 from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile
@@ -30,14 +35,15 @@ from ..db import (
    EvidenceConfidenceEnum,
    EvidenceTruthStatusEnum,
 )
-from ..db.models import EvidenceDB, ControlDB, AuditTrailDB
+from ..db.models import EvidenceDB, AuditTrailDB
 from ..services.auto_risk_updater import AutoRiskUpdater
-from ..services.evidence_service import EvidenceService
+from ..services.evidence_service import EvidenceService, _update_risks as _update_risks_impl
 from .schemas import (
    EvidenceCreate, EvidenceResponse, EvidenceListResponse,
    EvidenceRejectRequest,
 )
 from .audit_trail_utils import log_audit_trail
 from ._http_errors import translate_domain_errors
 logger = logging.getLogger(__name__)
 router = APIRouter(tags=["compliance-evidence"])
@@ -146,6 +152,7 @@ async def list_evidence(
    status: Optional[str] = None,
    page: Optional[int] = Query(None, ge=1, description="Page number (1-based)"),
    limit: Optional[int] = Query(None, ge=1, le=500, description="Items per page"),
    db: Session = Depends(get_db),
    service: EvidenceService = Depends(get_evidence_service),
 ) -> EvidenceListResponse:
    """List evidence with optional filters and pagination."""
@@ -186,9 +193,11 @@ async def list_evidence(
@router.post("/evidence", response_model=EvidenceResponse)
 async def create_evidence(
    evidence_data: EvidenceCreate,
    db: Session = Depends(get_db),
    service: EvidenceService = Depends(get_evidence_service),
 ) -> EvidenceResponse:
    """Create new evidence record."""
    dsms_cid = None
    repo = EvidenceRepository(db)
    # Get control UUID
@@ -257,6 +266,7 @@ async def create_evidence(
@router.delete("/evidence/{evidence_id}")
 async def delete_evidence(
    evidence_id: str,
    db: Session = Depends(get_db),
    service: EvidenceService = Depends(get_evidence_service),
 ) -> dict[str, Any]:
    """Delete an evidence record."""
@@ -275,6 +285,7 @@ async def upload_evidence(
    title: str = Query(...),
    file: UploadFile = File(...),
    description: Optional[str] = Query(None),
    db: Session = Depends(get_db),
    service: EvidenceService = Depends(get_evidence_service),
 ) -> EvidenceResponse:
    """Upload evidence file."""
@@ -674,6 +685,7 @@ async def collect_ci_evidence(
 async def get_ci_evidence_status(
    control_id: Optional[str] = Query(None, description="Filter by control ID"),
    days: int = Query(30, description="Look back N days"),
    db: Session = Depends(get_db),
    service: EvidenceService = Depends(get_evidence_service),
 ) -> dict[str, Any]:
    """Get CI/CD evidence collection status overview."""
@@ -681,70 +693,8 @@ async def get_ci_evidence_status(
        return service.ci_status(control_id, days)
-# ----------------------------------------------------------------------------
+# (Alte CI-Status-Implementierung entfernt — unerreichbarer Code nach `return
-# Legacy re-exports for tests that import helpers directly.
+# service.ci_status(...)`; durch den Service ersetzt, `query` war nie initialisiert.)
 # ----------------------------------------------------------------------------
    if control_id:
        ctrl_repo = ControlRepository(db)
        control = ctrl_repo.get_by_control_id(control_id)
        if control:
            query = query.filter(EvidenceDB.control_id == control.id)
    evidence_list = query.order_by(EvidenceDB.collected_at.desc()).limit(100).all()
    # Group by control and calculate stats
    control_stats = defaultdict(lambda: {
        "total": 0,
        "valid": 0,
        "failed": 0,
        "last_collected": None,
        "evidence": [],
    })
    for e in evidence_list:
        # Get control_id string
        control = db.query(ControlDB).filter(ControlDB.id == e.control_id).first()
        ctrl_id = control.control_id if control else "unknown"
        stats = control_stats[ctrl_id]
        stats["total"] += 1
        if e.status:
            if e.status.value == "valid":
                stats["valid"] += 1
            elif e.status.value == "failed":
                stats["failed"] += 1
        if not stats["last_collected"] or e.collected_at > stats["last_collected"]:
            stats["last_collected"] = e.collected_at
        # Add evidence summary
        stats["evidence"].append({
            "id": e.id,
            "type": e.evidence_type,
            "status": e.status.value if e.status else None,
            "collected_at": e.collected_at.isoformat() if e.collected_at else None,
            "ci_job_id": e.ci_job_id,
        })
    # Convert to list and sort
    result = []
    for ctrl_id, stats in control_stats.items():
        result.append({
            "control_id": ctrl_id,
            "total_evidence": stats["total"],
            "valid_count": stats["valid"],
            "failed_count": stats["failed"],
            "last_collected": stats["last_collected"].isoformat() if stats["last_collected"] else None,
            "recent_evidence": stats["evidence"][:5],
        })
    result.sort(key=lambda x: x["last_collected"] or "", reverse=True)
    return {
        "period_days": days,
        "total_evidence": len(evidence_list),
        "controls": result,
    }
 # ============================================================================
@@ -772,6 +722,7 @@ async def review_evidence(
    approval_status='first_approved'. A second (different) reviewer then
    sets second_reviewer and approval_status='approved'.
    """
    dsms_cid = None
    evidence = db.query(EvidenceDB).filter(EvidenceDB.id == evidence_id).first()
    if not evidence:
        raise HTTPException(status_code=404, detail=f"Evidence {evidence_id} not found")
@@ -851,6 +802,7 @@ async def reject_evidence(
    db: Session = Depends(get_db),
 ):
    """Reject evidence (sets approval_status='rejected')."""
    dsms_cid = None
    evidence = db.query(EvidenceDB).filter(EvidenceDB.id == evidence_id).first()
    if not evidence:
        raise HTTPException(status_code=404, detail=f"Evidence {evidence_id} not found")
@@ -8,7 +8,7 @@ This adds NO new reasoning logic. It exposes the already-built, tested orchestra
 """
 import logging
-from typing import List, Optional
+from typing import Dict, List, Optional
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
@@ -20,7 +20,7 @@ from compliance.onboarding import (
    ProducedSignal,
    RejectedAssumption,
 )
-from compliance.services.onboarding_service import run_advisor, supported_targets
+from compliance.services.onboarding_service import labels_for, run_advisor, supported_targets
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/onboarding", tags=["onboarding"])
@@ -50,6 +50,7 @@ class AdvisorResponse(BaseModel):
    evidence_requests: List[str] = Field(default_factory=list)
    unsupported_domains: List[str] = Field(default_factory=list)
    completeness_summary: str = ""
    capability_labels: Dict[str, str] = Field(default_factory=dict)   # capability_id -> human label (DE)
@router.get("/targets")
@@ -65,10 +66,17 @@ def advisor_start_endpoint(req: OnboardingAdvisorRequest) -> AdvisorResponse:
        company=req.company, certifications=req.certifications, target=req.target,
        signals=req.scanner_findings, known_evidence=req.known_evidence,
        products=req.products, markets=req.markets, industry=req.industry or "")
    surfaced = [
        *result.auto_detected, *result.indications, *result.capability_delta,
        *(q.capability_id for q in result.next_best_questions),
        *(c for a in result.inferred_assumptions for c in a.capabilities),
        *(m.capability_id for m in result.top_measures),
    ]
    return AdvisorResponse(
        silent_intake_summary=si_summary, headline=result.headline, auto_detected=result.auto_detected,
        indications=result.indications,
        inferred_assumptions=result.inferred_assumptions, rejected_assumptions=result.rejected_assumptions,
        top_5_questions=result.next_best_questions, capability_delta=result.capability_delta,
        top_measures=result.top_measures, evidence_requests=result.evidence_requests,
-        unsupported_domains=result.unsupported_domains, completeness_summary=result.completeness_summary)
+        unsupported_domains=result.unsupported_domains, completeness_summary=result.completeness_summary,
        capability_labels=labels_for(surfaced))
@@ -24,6 +24,7 @@ from fastapi.responses import FileResponse
 from sqlalchemy.orm import Session
 from classroom_engine.database import get_db
 from ..db.models import EvidenceDB
 from .audit_trail_utils import log_audit_trail
 from ..db import (
@@ -310,6 +311,7 @@ async def list_controls_paginated(
 )
 async def get_control(
    control_id: str,
    db: Session = Depends(get_db),
    svc: ControlExportService = Depends(get_ctrl_export_service),
 ) -> ControlResponse:
    """Get a specific control by control_id."""
@@ -354,6 +356,7 @@ async def get_control(
 async def update_control(
    control_id: str,
    update: ControlUpdate,
    db: Session = Depends(get_db),
    svc: ControlExportService = Depends(get_ctrl_export_service),
 ) -> ControlResponse:
    """Update a control."""
@@ -443,6 +446,7 @@ async def update_control(
 async def review_control(
    control_id: str,
    review: ControlReviewRequest,
    db: Session = Depends(get_db),
    svc: ControlExportService = Depends(get_ctrl_export_service),
 ) -> ControlResponse:
    """Mark a control as reviewed with new status."""
@@ -21,7 +21,7 @@ Phase 1 Step 4 refactor: handlers delegate to VVTService.
 import logging
 from typing import Any, List, Optional
-from fastapi import APIRouter, Depends, Query, Request
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
 from fastapi.responses import StreamingResponse
 from sqlalchemy.orm import Session
@@ -21,6 +21,14 @@ from .observations import (
    empirical_distribution,
    reviewed,
 )
 from .observation_log import (
    HypothesisStats,
    ObservationRecord,
    aggregate_by_hypothesis,
    append_observation,
    load_observations,
    review_queue,
 )
 from .signals import (
    ProducedSignal,
    SignalVocabularyEntry,
@@ -69,4 +77,10 @@ __all__ = [
    "ProducedSignal",
    "SignalVocabularyEntry",
    "normalize_signals",
    "ObservationRecord",
    "HypothesisStats",
    "append_observation",
    "load_observations",
    "aggregate_by_hypothesis",
    "review_queue",
 ]
@@ -143,8 +143,8 @@ def advisor_start(
        next_best_questions=next_q, capability_delta=delta, top_measures=measures,
        evidence_requests=evidence, unsupported_domains=unsupported,
        completeness_summary=rep.completeness_summary,
-        headline="%d Anforderungen erkannt · %d automatisch erkannt (Intake) · %d wahrscheinlich (Zertifikate) · %d zu klären"
+        headline="%d von %d Anforderungen offen · %d automatisch erkannt (Intake) · %d wahrscheinlich (Zertifikate) · %d zu klären"
-        % (len(assess.coverage), len(auto_detected), len(probably), len(next_q)))
+        % (len(delta), len(assess.coverage), len(auto_detected), len(probably), len(next_q)))
 def apply_answer(known_capabilities: Sequence[str], capability_id: str, answer: str) -> List[str]:
@@ -0,0 +1,108 @@
 """Observation Log — append-only JSONL store for empirical calibration events (Task 59b v1).
 Observations are NOT business data and NOT product-DB data — they are CALIBRATION events for the
 knowledge base ("ISO27001 -> SDL confirmed", "TISAX -> supplier security refuted"). So they live with the
 other versioned knowledge artifacts (hypotheses, transition patterns, vocabulary), NOT in the product
 database: an append-only JSONL log under `knowledge/observations/`. NO migration, NO DB. The empirical
 DISTRIBUTION and CONFIDENCE are COMPUTED from this log on demand (computed-not-stored) — a hypothesis is
 NEVER auto-updated; only REVIEWED observations calibrate (the review gate, enforced in observations.py).
 Append-only: each line is one ObservationRecord and lines are NEVER modified in place. A later review is
 a NEW line with the same observation_id and reviewed=true; load_observations() reconciles to the latest
 per id. You can `rm` the log and recompute, `git diff` it over months, or rebuild confidence under a new
 policy. Anonymisation is MANDATORY: customer_archetype is a sector/cert archetype, NEVER a real company
 name (this file is committed to git). Time is stamped by the CALLER (no hidden clock) for determinism.
 I/O only at the append/load boundary; statistics are pure. Python 3.9 compatible.
 """
 from __future__ import annotations
 import json
 import os
 from typing import Dict, List, Optional, Sequence
 from pydantic import BaseModel, Field
 from .observations import Observation, empirical_confidence, empirical_distribution
 _DEFAULT_LOG = os.path.join(
    os.path.dirname(__file__), "..", "..", "knowledge", "observations", "observations.jsonl")
 class ObservationRecord(Observation):
    """A persisted observation line: an Observation (with its review gate + observation_type) plus log
    metadata. `observation_id` is stable — a review re-appends the SAME id with reviewed=true."""
    observation_id: str                                  # stable id; a review re-appends the same id
    timestamp: str = ""                                  # ISO 8601, stamped by the CALLER (no hidden clock)
    customer_archetype: str = ""                         # sector/cert archetype — NEVER a real company name
    evidence: str = ""                                   # what backs the answer (reference, not the artifact)
    provenance: str = ""                                 # where the answer came from (audit trail)
    knowledge_version: str = ""                          # hypotheses/vocabulary version observed under
 class HypothesisStats(BaseModel):
    """Per-hypothesis empirical rollup — all COMPUTED from the log, nothing stored on the hypothesis."""
    hypothesis_id: str
    distribution: Dict[str, int] = Field(default_factory=dict)   # reviewed counts per observation_type
    confidence: Optional[float] = None                           # None until a for/against obs is reviewed
    reviewed_count: int = 0
    total_count: int = 0
 def append_observation(record: ObservationRecord, path: str = _DEFAULT_LOG) -> None:
    """Append ONE record as a JSON line. Append-only — existing lines are never rewritten."""
    os.makedirs(os.path.dirname(path), exist_ok=True)
    line = json.dumps(record.model_dump(mode="json"), ensure_ascii=False, sort_keys=True)
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(line + "\n")
 def load_observations(path: str = _DEFAULT_LOG, reconcile: bool = True) -> List[ObservationRecord]:
    """Read all records — a single `.jsonl` file or a directory of monthly `.jsonl` files. With
    reconcile, the LATEST record per observation_id wins (a later reviewed=true supersedes the original).
    Returns deterministic order (by observation_id when reconciled, else append order)."""
    files: List[str] = []
    if os.path.isdir(path):
        files = sorted(os.path.join(path, f) for f in os.listdir(path) if f.endswith(".jsonl"))
    elif os.path.exists(path):
        files = [path]
    records: List[ObservationRecord] = []
    for fpath in files:
        with open(fpath, encoding="utf-8") as fh:
            for raw in fh:
                raw = raw.strip()
                if raw:
                    records.append(ObservationRecord(**json.loads(raw)))
    if not reconcile:
        return records
    latest: Dict[str, ObservationRecord] = {}
    for r in records:                                    # file/append order -> later lines win
        latest[r.observation_id] = r
    return [latest[k] for k in sorted(latest)]
 def aggregate_by_hypothesis(records: Sequence[ObservationRecord]) -> List[HypothesisStats]:
    """Per-hypothesis distribution + confidence. The review gate applies inside empirical_distribution/
    empirical_confidence (reviewed-only), so unreviewed observations are counted in total but never
    calibrate. Deterministic order (by hypothesis id)."""
    by_hyp: Dict[str, List[ObservationRecord]] = {}
    for r in records:
        by_hyp.setdefault(r.hypothesis_id, []).append(r)
    out: List[HypothesisStats] = []
    for hyp in sorted(by_hyp):
        obs = by_hyp[hyp]
        out.append(HypothesisStats(
            hypothesis_id=hyp,
            distribution=empirical_distribution(obs),    # reviewed-only (the gate)
            confidence=empirical_confidence(obs),        # None until reviewed for/against
            reviewed_count=sum(1 for o in obs if o.reviewed),
            total_count=len(obs)))
    return out
 def review_queue(records: Sequence[ObservationRecord]) -> List[ObservationRecord]:
    """The reviewer's worklist: observations not yet reviewed. Calibration ignores these until a reviewer
    accepts them (Observation -> Review -> Accepted -> Knowledge recomputed), never Observation -> conf++."""
    return [r for r in records if not r.reviewed]
@@ -9,7 +9,7 @@ It adds NO new reasoning logic — it only exposes what exists. No DB, no persis
 from __future__ import annotations
 import os
-from typing import Any, Dict, List, Sequence, Tuple
+from typing import Any, Dict, Iterable, List, Sequence, Tuple
 import yaml
@@ -37,6 +37,13 @@ def _load(*parts: str) -> Any:
 _HYP_LIB = [CapabilityHypothesis(**h) for h in _load("certification_hypotheses", "hypotheses.yaml")["hypotheses"]]
 _VOCAB = [SignalVocabularyEntry(**v) for v in _load("onboarding", "signal_vocabulary.yaml")["signals"]]
 _SIGNAL_MAP = [SignalMapping(**m) for m in _load("onboarding", "intake_signal_map.yaml")["mappings"]]
 _LABELS: Dict[str, str] = _load("onboarding", "capability_labels.yaml")["labels"]
 def labels_for(capability_ids: Iterable[str]) -> Dict[str, str]:
    """Human labels (DE) for the given capability ids — presentation only. Ids without a curated label
    are omitted (the frontend falls back to a prettified id). Deduped, deterministic."""
    return {c: _LABELS[c] for c in dict.fromkeys(capability_ids) if c in _LABELS}
 # target id -> transition pattern that defines its required capabilities (curated registry)
 _TARGET_PATTERNS = {
@@ -53,9 +60,10 @@ def supported_targets() -> List[str]:
 def _target(target_id: str) -> Tuple[List[TargetRequirement], Dict[str, List[str]]]:
    pat = _load("transition_patterns", _TARGET_PATTERNS[target_id])
-    reqs = [TargetRequirement(capability_id=a["capability"]) for a in pat["likely_covered"]]
+    reqs = [TargetRequirement(capability_id=a["capability"], rationale=a.get("reviewable_claim", "")) for a in pat["likely_covered"]]
    reqs += [TargetRequirement(capability_id=d["capability"], question_intent=d.get("needed_information", "verify_existence"),
-                               expected_evidence=d.get("expected_evidence", [])) for d in pat["delta_requirements"]]
+                               rationale=d.get("why_asked", ""), expected_evidence=d.get("expected_evidence", []))
             for d in pat["delta_requirements"]]
    covers = {d["capability"]: d.get("covers_targets", []) for d in pat["delta_requirements"]}
    return reqs, covers
@@ -104,7 +104,8 @@ def assess_transition(
        )
        buckets[status].append(req.capability_id)
        if status in _REQUESTABLE:
-            reason, prio = _REQUESTABLE[status]
+            default_reason, prio = _REQUESTABLE[status]
            reason = req.rationale or default_reason   # curated human text wins over the generic fallback
            requests.append(
                TransitionQuestionRequest(
                    capability_id=req.capability_id,
@@ -70,6 +70,7 @@ class TargetRequirement(BaseModel):
    capability_id: str  # MCAP-...
    question_intent: str = "verify_existence"  # passed through to the request, not rendered
    rationale: str = ""  # curated human text (e.g. why_asked / reviewable_claim) — surfaced as the request reason
    expected_evidence: List[str] = Field(default_factory=list)
    source_control_id: Optional[str] = None
    supports_obligations: List[str] = Field(default_factory=list)
@@ -0,0 +1,2 @@
 # Append-only observation log (Task 59b). Real lines (observations.jsonl / YYYY-MM.jsonl) are written at
 # runtime via compliance/onboarding/observation_log.py. Anonymised archetypes only — NEVER real company names.
@@ -0,0 +1,45 @@
 # Human-readable capability labels (DE) — presentation only, reusable across all targets.
 # A capability id is the stable machine identity; this maps it to an expert-facing label for the UI.
 # Curated knowledge (draft — to be corrected by the domain expert). Missing ids fall back to a
 # prettified id in the frontend. NO real company names. Keep labels short + concrete.
 labels:
  # ── ISMS / ISO 27001 core ───────────────────────────────────────────────
  information_security_management: "Informationssicherheits-Managementsystem (ISMS)"
  access_control_and_authentication: "Zugriffskontrolle & Authentifizierung"
  asset_and_configuration_management: "Asset- & Konfigurationsverwaltung"
  cryptography: "Kryptographie / Verschlüsselung"
  incident_management: "Security-Incident-Management"
  security_awareness_training: "Security-Awareness-Schulungen"
  supplier_security: "Lieferanten-Sicherheit"
  security_logging_and_monitoring: "Security-Logging & Monitoring"
  technical_vulnerability_management: "Technisches Schwachstellen-Management"
  # ── TISAX / VDA-spezifisch ──────────────────────────────────────────────
  prototype_protection: "Prototypenschutz (physisch & logisch)"
  tisax_label_scope_selection: "TISAX-Label-/Scope-Festlegung"
  tisax_assessment_via_enx: "TISAX-Assessment über die ENX-Plattform"
  vda_isa_self_assessment: "VDA-ISA-Selbstauskunft"
  data_protection_processing_on_behalf: "Auftragsverarbeitung (Art. 28 DSGVO)"
  physical_security: "Physische Sicherheit / Zutrittskontrolle"
  # ── QM / ISO 9001 ───────────────────────────────────────────────────────
  document_and_change_control: "Dokumenten- & Änderungslenkung"
  supplier_evaluation: "Lieferantenbewertung"
  release_and_approval_process: "Freigabe- & Genehmigungsprozess"
  ce_conformity_assessment_and_technical_documentation: "CE-Konformitätsbewertung & technische Dokumentation"
  # ── CRA / Produkt-Cybersecurity ─────────────────────────────────────────
  sbom_creation: "SBOM-Erstellung (Software-Stückliste)"
  coordinated_vulnerability_disclosure: "Coordinated Vulnerability Disclosure (CVD)"
  secure_development_lifecycle: "Sicherer Entwicklungslebenszyklus (SDLC)"
  secure_signed_update_distribution: "Sichere, signierte Update-Verteilung"
  security_update_support_period: "Sicherheits-Update-Supportzeitraum"
  product_cyber_risk_assessment: "Produkt-Cyber-Risikobewertung"
  exploited_vuln_and_incident_reporting: "Meldung ausgenutzter Schwachstellen & Vorfälle"
  public_security_advisories: "Öffentliche Security Advisories"
  cybersecurity_management_system: "Cybersecurity-Managementsystem (CSMS)"
  # ── MaschinenVO / Safety ────────────────────────────────────────────────
  machine_safety_risk_assessment: "Maschinen-Risikobeurteilung"
  mechanical_safety_and_guards: "Mechanische Sicherheit & Schutzeinrichtungen"
  operating_instructions_and_safety_information: "Betriebsanleitung & Sicherheitshinweise"
  protection_against_corruption_of_safety_functions: "Schutz der Sicherheitsfunktionen vor Manipulation"
  # ── Umwelt ──────────────────────────────────────────────────────────────
  environmental_management_documentation: "Umweltmanagement-Dokumentation"
@@ -0,0 +1,73 @@
 """Observation Log — append-only JSONL store + computed statistics (Task 59b/c v1).
 Pins the user's decision (2026-06-28): observations are CALIBRATION data, not product data -> an
 append-only JSONL log under knowledge/observations/, NO DB, NO migration. Distribution and confidence are
 COMPUTED from the log; only REVIEWED observations calibrate (review gate); a later review is a new line
 that supersedes by observation_id. Nothing is ever written back to a hypothesis.
 """
 from __future__ import annotations
 from compliance.onboarding import (
    ObservationRecord,
    ObservationType,
    aggregate_by_hypothesis,
    append_observation,
    load_observations,
    review_queue,
 )
 def _rec(oid, hyp, otype, reviewed=False, **kw):
    return ObservationRecord(
        observation_id=oid, hypothesis_id=hyp, observation_type=otype, reviewed=reviewed,
        timestamp="2026-07-01T00:00:00Z", customer_archetype="machine_builder+ISO27001", **kw)
 def test_append_only_round_trip(tmp_path):
    p = str(tmp_path / "obs.jsonl")
    append_observation(_rec("o1", "HYP-secure_dev", ObservationType.CONFIRMED, reviewed=True), p)
    append_observation(_rec("o2", "HYP-secure_dev", ObservationType.REFUTED, reviewed=True), p)
    recs = load_observations(p)
    assert {r.observation_id for r in recs} == {"o1", "o2"}
    assert all(r.customer_archetype == "machine_builder+ISO27001" for r in recs)  # anonymised archetype, not a name
 def test_review_supersedes_by_id_append_only(tmp_path):
    p = str(tmp_path / "obs.jsonl")
    append_observation(_rec("o1", "HYP-x", ObservationType.CONFIRMED, reviewed=False), p)   # raw answer
    append_observation(_rec("o1", "HYP-x", ObservationType.CONFIRMED, reviewed=True,
                            reviewed_by="anna"), p)                                          # later review event
    assert len(load_observations(p, reconcile=False)) == 2                                  # both lines kept (append-only)
    recs = load_observations(p)                                                             # reconciled
    assert len(recs) == 1 and recs[0].reviewed and recs[0].reviewed_by == "anna"
 def test_statistics_apply_the_review_gate(tmp_path):
    p = str(tmp_path / "obs.jsonl")
    append_observation(_rec("a", "HYP-sdl", ObservationType.CONFIRMED, reviewed=True), p)
    append_observation(_rec("b", "HYP-sdl", ObservationType.CONFIRMED, reviewed=True), p)
    append_observation(_rec("c", "HYP-sdl", ObservationType.REFUTED, reviewed=True), p)
    append_observation(_rec("d", "HYP-sdl", ObservationType.CONFIRMED, reviewed=False), p)  # unreviewed -> ignored
    stats = {s.hypothesis_id: s for s in aggregate_by_hypothesis(load_observations(p))}
    s = stats["HYP-sdl"]
    assert s.total_count == 4 and s.reviewed_count == 3
    assert s.distribution["confirmed"] == 2 and s.distribution["refuted"] == 1   # unreviewed one excluded
    assert s.confidence == round(2 / 3, 2)                                        # (2 + 0.5*0) / 3
 def test_review_queue_lists_unreviewed(tmp_path):
    p = str(tmp_path / "obs.jsonl")
    append_observation(_rec("a", "HYP-y", ObservationType.CONFIRMED, reviewed=True), p)
    append_observation(_rec("b", "HYP-y", ObservationType.PARTIAL, reviewed=False), p)
    q = review_queue(load_observations(p))
    assert [r.observation_id for r in q] == ["b"]
 def test_load_directory_of_monthly_files(tmp_path):
    d = tmp_path / "observations"
    d.mkdir()
    append_observation(_rec("a", "HYP-z", ObservationType.CONFIRMED, reviewed=True), str(d / "2026-06.jsonl"))
    append_observation(_rec("b", "HYP-z", ObservationType.REFUTED, reviewed=True), str(d / "2026-07.jsonl"))
    recs = load_observations(str(d))
    assert {r.observation_id for r in recs} == {"a", "b"}
@@ -73,6 +73,17 @@ def test_partial_signal_surfaces_as_indication_and_is_still_asked():
    assert "secure_development_lifecycle" in asked or "secure_development_lifecycle" in d["capability_delta"]
 def test_questions_carry_curated_text_and_human_labels():
    # the curated why_asked from the transition pattern must reach the question (not the generic
    # fallback "Keine Anhaltspunkte ... klären"), and surfaced capabilities get human labels.
    body = dict(_BODY, certifications=["ISO27001"], target="TISAX", scanner_findings=[])
    r = _client.post("/onboarding/advisor-start", json=body)
    assert r.status_code == 200, r.text
    d = r.json()
    assert any("Keine Anhaltspunkte" not in q["why"] for q in d["top_5_questions"])   # real expert text surfaced
    assert d["capability_labels"].get("vda_isa_self_assessment") == "VDA-ISA-Selbstauskunft"
 def test_unknown_target_is_404():
    body = dict(_BODY, target="NOPE")
    r = _client.post("/onboarding/advisor-start", json=body)
		`@@ -0,0 +1,2 @@`
							`# Append-only observation log (Task 59b). Real lines (observations.jsonl / YYYY-MM.jsonl) are written at`
							`# runtime via compliance/onboarding/observation_log.py. Anonymised archetypes only — NEVER real company names.`