fix(tolaria): C5D search-source pipeline — build+verify source from current Tolaria before rebuild

- SearchSourceBuilder: deterministischer Vault->Source-Snapshot (read-only),
  atomar (temp->validate->fsync->replace), Secret-Scan fail-closed
- verify_integrity: exakte object_id/path Set-Equality (stale Source kann
  nie APPLIED), Canary implizit ueber erwartetes Objekt-Set
- apply_commit: Source-Build+Verification vor Rebuild; VERIFYING_SEARCH-Resume
  (kein Doppel-Rebuild) — C5E-Replay-Crash-Fall abgedeckt
- Fix: source_object_count ist keine 0-Fehlerbedingung (echter Defekt)
- SearchSourceBuildError + RC_SEARCH_SOURCE_BUILD_FAILURE (Human Gate)
- c4b: source_path env-konfigurierbar, indexed_object_ids/paths Read-Back
- Testsuite: 17 neue Tests (Test-Plan A-O + Realistic C4-Integration)

tolaria/c4b-search-service/rebuild.py

@@ -31,6 +31,9 @@ def main():
         "blocked": res["blocked"],
         "index_path": INDEX_JSON,
         "supported_modes": ["exact", "keyword", "metadata"],
+        "source_head": engine.source_head,
+        "indexed_object_ids": engine.indexed_object_ids,
+        "indexed_paths": engine.indexed_paths,
     }, ensure_ascii=False, indent=2))
 
 

tolaria/c4b-search-service/search_api.py

@@ -101,6 +101,13 @@ class TolariaSearch:
         self.failed_objects = []
         self.secret_blocked_objects = []
         self.stale_objects = []
+        # C5D: deterministische Objekt-Set-Verifikation (Read-Back).
+        # indexed_object_ids = object_id-Werte der tatsaechlich indexierten
+        # Objekte; indexed_paths = deren Pfade. Wird nach jedem Rebuild und
+        # beim Laden aus dem persistierten Index befuellt und in health()
+        # zurueckgegeben, damit C5D exakt (nicht nur count>0) verifizieren kann.
+        self.indexed_object_ids: list[str] = []
+        self.indexed_paths: list[str] = []
 
     # -- Rebuild -----------------------------------------------------------
     def rebuild_from_source(self, source_path: str, head: Optional[str] = None,
@@ -132,6 +139,12 @@ class TolariaSearch:
         self.secret_blocked_objects = list(self.index.secret_blocked)
         self.failed_objects = []
         self.stale_objects = []
+        # C5D: deterministische Objekt-Set-Verifikation (nur tatsaechlich
+        # indexierte, nicht secret-geblockte Objekte).
+        self.indexed_object_ids = [
+            d.id for d in self.index.docs if d.id is not None
+        ]
+        self.indexed_paths = [d.path for d in self.index.docs]
         if self.index_path:
             os.makedirs(os.path.dirname(self.index_path), exist_ok=True)
             payload = {
@@ -140,6 +153,8 @@ class TolariaSearch:
                 "built_at_ms": self.built_at,
                 "object_count": self.object_count,
                 "secret_blocked": self.secret_blocked_objects,
+                "indexed_object_ids": self.indexed_object_ids,
+                "indexed_paths": self.indexed_paths,
                 "docs": [
                     {"path": d.path, "title": d.title, "id": d.id, "type": d.type,
                      "role": d.role, "representation": d.representation,
@@ -167,6 +182,9 @@ class TolariaSearch:
             "index_version": INDEX_VERSION,
             "source_head": self.source_head,
             "secret_blocked_objects": len(self.secret_blocked_objects),
+            # C5D: deterministische Objekt-Set-Verifikation (Read-Back).
+            "indexed_object_ids": self.indexed_object_ids,
+            "indexed_paths": self.indexed_paths,
         }
 
     # -- Snippet (secret-gefiltert) -----------------------------------------

tolaria/c4b-search-service/server.py

@@ -22,7 +22,10 @@ from urllib.parse import urlparse, parse_qs
 from search_api import TolariaSearch, SearchError, SUPPORTED_MODES
 
 DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
-SOURCE_JSON = os.path.join(os.path.dirname(__file__), "index_source.json")
+# C5D: Source-Pfad env-konfigurierbar. Default = bisheriger baked-in Pfad,
+# aber C5D schreibt den frischen derived Snapshot an den konfigurierten Pfad.
+SOURCE_JSON = os.environ.get("TOLARIA_SEARCH_SOURCE",
+                             os.path.join(os.path.dirname(__file__), "index_source.json"))
 INDEX_JSON = os.path.join(DATA_DIR, "search_index.json")
 REBUILD_TOKEN = os.environ.get("TOLARIA_SEARCH_REBUILD_TOKEN", "")
 
@@ -54,6 +57,9 @@ def load_or_rebuild():
             engine.source_head = payload.get("source_head")
             engine.object_count = payload.get("object_count", len(docs))
             engine.secret_blocked_objects = payload.get("secret_blocked", [])
+            # C5D: Objekt-Set aus persistiertem Index wiederherstellen (Read-Back).
+            engine.indexed_object_ids = payload.get("indexed_object_ids", [])
+            engine.indexed_paths = payload.get("indexed_paths", [])
             return "loaded"
         except Exception as e:
             pass

tolaria/c5-sync-service/C5D_SEARCH_SOURCE_PIPELINE_ARCHITECTURE_DECISION.md

@@ -0,0 +1,124 @@
+# C5D — SEARCH SOURCE PIPELINE REPAIR: ARCHITECTURE_DECISION
+
+**Datum:** 2026-08-26 · **Autor:** Red Queen (MAKER) · **Status:** FREIGEGEBEN (Christian)
+**Mission:** fehlende deterministische Verbindung `TOLARIA VERIFIED STATE → SEARCH SOURCE BUILD/REFRESH → SEARCH REBUILD → SEARCH VERIFICATION` implementieren.
+
+---
+
+## 1. Root Cause (aus C5F-Evidence, übernommen & bestätigt)
+
+- `/app/index_source.json` ist **statisch baked-in** (head `35446c03…`, count=84) und wird von **keinem C4/C5-Producer** aktualisiert.
+- `/api/search/rebuild` konsumiert **ausschließlich** diese stale Source (`server.py` `SOURCE_JSON` hartkodiert).
+- C5D und C5E-Replay führen **keinen Source-Build** aus.
+- `verify_integrity()` akzeptiert `object_count > 0` → kann einen **semantisch stale** Search-Stand als PASS bewerten.
+- Produktiver Rebuild (Rain): HTTP 200 `status=ok indexed=84 blocked=0` bei Tolaria=85 / index_source=84 / Search=84 → **Canary Discovery=FAIL**. HTTP-200 allein darf NIE APPLIED ermöglichen.
+
+---
+
+## 2. ARCHITECTURE_DECISION
+
+### 2.1 SOURCE_BUILDER_OWNER = **C5DEngine**
+
+Der Source-Build gehört in **C5D** (nicht in C5C, nicht in den Search-Service selbst).
+
+**Begründung (gegen die Alternativen):**
+
+| Kriterium | C5C | Search-Service (Weg A) | **C5D (GEWÄHLT)** |
+|---|---|---|---|
+| Eindeutige Ownership | ✗ C5C ist Forgejo→Tolaria; Search internes Detail | ✗ koppelt Retrieval-Layer an Tolaria | ✓ C5D orchestriert bereits den kompletten Search-Schritt |
+| Deterministischer Input | ✗ | ✗ | ✓ C5D läuft NUR nach Tolaria-Verifikation (DRIFT=0) |
+| Atomare Veröffentlichung | ✗ | ✗ | ✓ temp→validate→atomic rename→rebuild |
+| Recovery | ✗ | ✗ | ✓ C5D-State-Machine (idempotent) |
+| Split-Brain-Gefahr | hoch (2 Writer) | hoch | **minimal** (1 Owner) |
+| Security Boundary | ✗ | ✗ | ✓ Search-Token bleibt bei C5D |
+| Testbarkeit | mittel | schlecht | **hoch** (bestehendes Fake-Injektionsmuster) |
+| Keine neue Source of Truth | ✓ | ✓ | ✓ (ephemeral build artifact) |
+
+**Gegen C5C:** C5C-Docstring verbietet jeden Search-Bezug („Kein Search-Rebuild-Aufruf“); Source-Build würde C5C an Search-Internal koppeln und die Rollentrennung verletzen.
+**Gegen Search-Service selbst:** würde dem abgeleiteten Retrieval-Layer eine Tolaria-Leseabhängigkeit aufzwingen und die Ownership aufteilen. Der Service bleibt ein **dummer Konsument** einer Source-Datei (nur Pfad env-konfigurierbar).
+
+**Konsequenz:** C5D liest Tolaria **read-only** (Source-Build). Die bestehende `NO-TOLARIA-WRITE-GUARANTEE` bleibt unverändert (verbietet nur `/api/vault/save`). Statische Guards werden entsprechend erweitert.
+
+### 2.2 SOURCE_BUILD_CONTRACT (exakt, was `rebuild_from_source` liest)
+
+Der Builder erzeugt deterministisch:
+```json
+{
+  "head": "<commit_sha des verifizierten Commits>",  // Provenance-Anker
+  "count": <int>,
+  "objects": [
+    {
+      "path": "<REQUIRED>", "title": str,
+      "id": "<object/UUID> | null", "type": str|null,
+      "role": str|null, "representation": str|null,
+      "state": str|null, "knowledge_schema": str|null,
+      "content_hash": sha256(body), "body": str,
+      "aliases": [], "tags": [], "derived_from": str|null
+    }, ...
+  ]
+}
+```
+- `content_hash` = `hashlib.sha256(body)` (identisch zu `rq_c5b.content_hash`).
+- `knowledge_schema` wird von der Engine nicht konsumiert, ist aber Teil des bestehenden Source-Formats → wird aus Tolaria-Metadaten übernommen (kein erfundener Wert).
+- **Kein einziges erfundenes Feld.** Jedes Feld stammt aus dem gelesenen Tolaria-Objekt.
+
+### 2.3 SOURCE_VERIFICATION_CONTRACT (Completeness, §5)
+
+Vor dem Rebuild wird der erzeugte Snapshot **vollständig** verifiziert (nicht nur `object_count`):
+- `SOURCE_OBJECT_COUNT` == `len(erwartete_indexierbare_object_ids)`
+- `UNIQUE_OBJECT_IDS` (keine Duplikate)
+- `UNIQUE_PATHS` (keine Duplikate)
+- `INVALID_OBJECTS` == [] (fehlendes `path`, malformed)
+- `SECRET_BLOCKED` fail-closed
+- `EXPECTED_SOURCE_HEAD` == commit_sha
+- **Objekt-Set-Gleichheit:** `set(object_ids) ∪ legacy_paths` == `erwartete_indexierbare_menge`
+
+**Erwartete indexierbare Menge (deterministisch, definiert):** = Menge der Tolaria-Vault-Objekte, die **IN_SCOPE** sind **UND** eine gültige `object_id` haben (via C5B `KnowledgeScope`/`extract_object_id`), **PLUS** die LEGACY_SPECIALs (`README.md` x2, `vps.md`). Objekte ohne gültige object_id (z.B. das `start.md`-Orphan, HUMAN_REVIEW) sind **nicht** indexierbar → werden ausgeschlossen. Das reproduziert deterministisch 84→85-Übergang (Canary hat gültige object_id).
+
+Bei Abweichung: **FAIL CLOSED**, kein Rebuild.
+
+### 2.4 ATOMICITY_MODEL (§6)
+
+`build temp → validate temp → fsync → os.replace (atomic rename) → erst dann rebuild`.
+Ein Crash hinterlässt nie eine halb geschriebene Source: entweder alte komplette oder neue komplette Datei. Temp-Datei wird bei Restart verworfen (idempotent neu gebaut).
+
+### 2.5 STATE_MACHINE_CHANGES (§8) — KEINE neuen States
+
+**Entscheidung:** Source-Build + -Verification laufen **atomar innerhalb des bestehenden `UPDATING_SEARCH`-Schritts**, unmittelbar vor dem Rebuild. Kein neuer `BUILDING_SEARCH_SOURCE`/`VERIFYING_SEARCH_SOURCE`-State.
+
+**Begründung:** (1) Build ist idempotent (kein Tolaria-Write → kein Doppel-Write-Risiko); (2) Restart re-entert aus `UPDATING_SEARCH`/`RETRY_PENDING` → Build wird deterministisch wiederholt; (3) Fehler werden über den bestehenden Retry/Human-Gate-Mechanismus klassifiziert. Die Invariante („kein Rebuild vor verifizierter Source“) ist gewahrt, ohne die State-Zahl kosmetisch aufzublähen.
+
+**Eine notwendige Erweiterung:** `C5DEngine.apply_commit` akzeptiert zusätzlich `ST_VERIFYING_SEARCH` als Re-Entry (Crash nach Rebuild vor Health, §9-Fall D). Verhalten: idempotenter Re-Entry → vollständige Sequenz (Build→Rebuild→Health→Verify) wird sicher wiederholt.
+
+### 2.6 SEARCH-SERVICE-ÄNDERUNGEN (Deployment, §10)
+
+- `server.py`: `SOURCE_JSON` wird env-konfigurierbar über `TOLARIA_SEARCH_SOURCE` (Default = bisheriger Pfad). Die baked-in `index_source.json` bestimmt **nicht mehr** die produktive Wahrheit — C5D schreibt den frischen derived Snapshot an den konfigurierten Pfad.
+- `search_api.py` `health()`: liefert zusätzlich `indexed_object_ids` + `indexed_paths` (Read-Back für deterministische Objekt-Set-Verifikation, §7). Persistierter Index enthält ebenfalls diese Mengen.
+- **Keine** Netzwerköffnung, **kein** öffentliches Port-Mapping, Secrets bleiben beim Search-/C5D-Executor-Kontext.
+
+### 2.7 C5D_VERIFICATION_HARDENING (§7)
+
+`verify_integrity` ersetzt die `object_count > 0`-Regel durch **exakte Objekt-Set-Verifikation**:
+- `set(indexed_object_ids) == erwartete_indexierbare_ids`
+- `set(indexed_paths) == erwartete_indexierbare_paths`
+- `object_count == len(erwartet)`
+- `failed_objects == []`, `secret_blocked` gemäß Contract
+- **Canary-Beweis:** `canary_object_id ∈ indexed_object_ids` UND `canary_path ∈ indexed_paths` (statt „HTTP 200 + count>0“).
+
+Ein HTTP-200-Rebuild mit stale Source kann damit NIE APPLIED ermöglichen.
+
+---
+
+## 3. Betroffene Dateien
+
+| Datei | Änderung |
+|---|---|
+| `tolaria/c5-sync-service/rq_c5d.py` | + `SearchSourceBuilder`, `verify_integrity`-Härtung, `apply_commit` mit Build+Verify+VERIFYING_SEARCH-ReEntry, Guards |
+| `tolaria/c5-sync-service/rq_c5e.py` | Recovery berücksichtigt Source-Build-Schritt + `VERIFYING_SEARCH`-Resume |
+| `tolaria/c4b-search-service/server.py` | `TOLARIA_SEARCH_SOURCE` env-konfigurierbar |
+| `tolaria/c4b-search-service/search_api.py` | `health()` + persistierter Index liefert `indexed_object_ids`/`indexed_paths` |
+| `tolaria/c5-sync-service/test_c5d.py` | erweiterte Tests (A–O) |
+| `tolaria/c5-sync-service/test_c5e.py` | Crash-Fälle d5 |
+| neu: `tolaria/c5-sync-service/test_c5d_source.py` | Realistic Integration Test (echte C4-Engine, kein Fake) |
+
+**Nicht angefasst:** Produktions-DB, Canary-DB, `index_source.json` (bleibt als historische Evidence erhalten), Vault-Git-Mods, Forgejo-Master-Write im Repair.

tolaria/c5-sync-service/rq_c5a.py

@@ -118,6 +118,12 @@ RC_NETWORK_TIMEOUT = "NETWORK_TIMEOUT"         # Netzwerk-Timeout (transient, re
 RC_MALFORMED_RESPONSE = "MALFORMED_RESPONSE"   # Ungueltige/malformed Downstream-Antwort (nicht retrybar)
 RC_INTEGRITY_FAILURE = "INTEGRITY_FAILURE"     # Search-Health/Integrity nicht PASS (nicht retrybar)
 
+# C5D: Search-Source-Build/Verification fehlgeschlagen oder Objekt-Set weicht
+# vom erwarteten Tolaria-Stand ab (nicht retrybar, Human Gate). Trennschaerfer
+# als RC_SEARCH_REBUILD_FAILURE: kein Rebuild darf auf stale/unvollstaendiger
+# Source stattfinden; FAIL CLOSED vor APPLIED.
+RC_SEARCH_SOURCE_BUILD_FAILURE = "SEARCH_SOURCE_BUILD_FAILURE"
+
 # Retry / Backoff
 DEFAULT_MAX_RETRIES = 5
 DEFAULT_BACKOFF_SECONDS = [1, 2, 4, 8, 16]
@@ -147,6 +153,8 @@ REASON_CODES = frozenset({
     RC_NETWORK_TIMEOUT,
     RC_MALFORMED_RESPONSE,
     RC_INTEGRITY_FAILURE,
+    # C5D: Search-Source-Build/Verification
+    RC_SEARCH_SOURCE_BUILD_FAILURE,
 })
 
 # Alle Operationen als frozenset

tolaria/c5-sync-service/rq_c5d.py

@@ -61,10 +61,29 @@ from rq_c5a import (
     RC_AUTH_FAILURE,
     RC_UNKNOWN_OBJECT_ID,
     RC_UNEXPECTED_TOLARIA_DRIFT,
+    RC_SEARCH_SOURCE_BUILD_FAILURE,
+    RC_INTEGRITY_FAILURE,
     DEFAULT_MAX_RETRIES,
     DEFAULT_BACKOFF_SECONDS,
 )
-from rq_c5c import C5CError
+from rq_c5c import (
+    C5CError,
+    TolariaClient,
+    TolariaUnavailableError,
+    VAULT_PREFIX,
+)
+# C5B-Bausteine: deterministische Vault->Source-Extraktion (kein neues SoT).
+# content_hash, parse_frontmatter, extract_object_id, KnowledgeScope sind
+# identisch zur C5B-Change-Detection -> konsistente Metadaten-Hashes.
+from rq_c5b import (
+    parse_frontmatter,
+    extract_object_id,
+    content_hash,
+    detect_secret,
+    KnowledgeScope,
+    SCOPE_IN_SCOPE,
+    SCOPE_LEGACY_SPECIAL,
+)
 
 # ---------------------------------------------------------------------------
 # Konstanten
@@ -103,6 +122,16 @@ class SearchIntegrityError(SearchError):
     """Search Health/Integrity nicht PASS (nicht retrybar, Human Gate)."""
 
 
+class SearchSourceBuildError(SearchError):
+    """
+    Search-Source-Build/Verification fehlgeschlagen oder Objekt-Set weicht vom
+    erwarteten Tolaria-Stand ab (nicht retrybar, Human Gate).
+
+    C5D-Haertung: kein Rebuild darf auf stale/unvollstaendiger Source
+    stattfinden; FAIL CLOSED vor APPLIED.
+    """
+
+
 class SearchMalformedResponseError(SearchError):
     """Search-Antwort ungueltig / malformed (nicht retrybar, Human Gate)."""
 
@@ -201,11 +230,221 @@ class SearchClient:
         return resp
 
 
+# ---------------------------------------------------------------------------
+# Search Source Builder (deterministisch, §3–§6)
+# ---------------------------------------------------------------------------
+
+class SearchSourceBuilder:
+    """
+    Baut deterministisch den Search-Source-Snapshot AUS DEM AKTUELLEN
+    TOLARIA-VAULT-ZUSTAND (read-only) — exakt im Schema, das
+    TolariaSearch.rebuild_from_source() erwartet.
+
+    ARCHITEKTURREGEL:
+      * Search darf NUR einen Stand indexieren, der aus dem bereits
+        erfolgreich verifizierten Tolaria-Zustand erzeugt wurde.
+      * index_source.json ist KEINE Source of Truth, sondern ein
+        EPHEMERAL / REBUILDABLE / DERIVED BUILD ARTIFACT.
+      * Dieser Builder liest NUR Tolaria (kein Write nach Tolaria/Forgejo).
+
+    Bestimmt die "erwartete indexierbare Menge" deterministisch ueber die
+    C5B-KnowledgeScope-Klassifikation:
+      * IN_SCOPE + gueltige object_id -> indexierbar (object_id gesetzt)
+      * LEGACY_SPECIAL (README.md x2, vps.md) -> indexierbar (object_id=None)
+      * OUT_OF_SCOPE / HUMAN_REVIEW (keine gueltige object_id, z.B. Orphan
+        start.md) -> NICHT indexierbar -> aus Source ausgeschlossen.
+
+    Atomicity: build temp -> validate temp -> fsync -> os.replace (atomic
+    rename) -> erst dann Rebuild. Ein Crash hinterlaesst nie eine halb
+    geschriebene Source.
+    """
+
+    def __init__(self, tolaria: TolariaClient, source_path: str,
+                 scope: Optional[KnowledgeScope] = None):
+        self.tolaria = tolaria
+        self.source_path = source_path
+        self.scope = scope or KnowledgeScope(VAULT_PREFIX)
+
+    @staticmethod
+    def _relpath(vault_path: str) -> str:
+        """Strip des /app/vault-Praefixes -> repo-relative Pfad."""
+        p = vault_path.replace("\\", "/")
+        prefix = VAULT_PREFIX.rstrip("/") + "/"
+        if p.startswith(prefix):
+            return p[len(prefix):]
+        if p == VAULT_PREFIX.rstrip("/"):
+            return ""
+        return p
+
+    def _collect_indexable(self) -> Dict[str, Any]:
+        """
+        Liest den Vault (read-only) und klassifiziert die indexierbaren Objekte.
+        Rueckgabe: {"indexable": [{path, id, content, fm, body}],
+                     "excluded": [...], "vault_object_count": int}
+        """
+        entries = self.tolaria.list(VAULT_PREFIX) or []
+        indexable: List[Dict[str, Any]] = []
+        excluded: List[Dict[str, Any]] = []
+        for e in entries:
+            vp = e.get("path")
+            if not vp:
+                continue
+            rel = self._relpath(vp)
+            content = self.tolaria.read(vp)
+            if content is None:
+                excluded.append({"path": rel, "reason": "unreadable"})
+                continue
+            cls = self.scope.classify_with_content(rel, content)
+            sc = cls.get("scope")
+            if sc == SCOPE_IN_SCOPE:
+                oid = cls.get("object_id")
+                if oid:
+                    indexable.append({"path": rel, "id": oid, "content": content})
+                    continue
+                excluded.append({"path": rel, "reason": "no_object_id"})
+            elif sc == SCOPE_LEGACY_SPECIAL:
+                indexable.append({"path": rel, "id": None, "content": content})
+            else:
+                excluded.append({"path": rel, "reason": sc})
+        return {"indexable": indexable, "excluded": excluded,
+                "vault_object_count": len(entries)}
+
+    def _source_objects(self, indexable: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Konvertiert indexierbare Vault-Objekte ins Source-Schema (deterministisch)."""
+        objs: List[Dict[str, Any]] = []
+        for it in indexable:
+            fm, body = parse_frontmatter(it["content"])
+            rel = it["path"]
+            objs.append({
+                "path": rel,
+                "title": fm.get("title") or rel.rsplit("/", 1)[-1].replace(".md", ""),
+                "id": it["id"],
+                "type": fm.get("type"),
+                "role": fm.get("role"),
+                "representation": fm.get("representation"),
+                "state": fm.get("state"),
+                "knowledge_schema": fm.get("knowledge_schema"),
+                "content_hash": content_hash(body),
+                "body": body,
+                "aliases": [],
+                "tags": fm.get("tags", []) or [],
+                "derived_from": fm.get("derived_from"),
+            })
+        # Deterministische Reihenfolge (Pfad-sortiert) — kein Hash-/Set-Chaos.
+        objs.sort(key=lambda o: o["path"])
+        return objs
+
+    def compute_expected(self, commit_sha: str) -> Dict[str, Any]:
+        """
+        Read-only: bestimmt die erwartete indexierbare Objekt-Menge (object_ids
+        und Pfade) aus dem aktuellen Tolaria-Stand. Kein Write, kein Rebuild.
+        """
+        coll = self._collect_indexable()
+        objs = self._source_objects(coll["indexable"])
+        expected_ids = sorted(o["id"] for o in objs if o["id"] is not None)
+        expected_paths = sorted(o["path"] for o in objs)
+        return {
+            "commit_sha": commit_sha,
+            "source_head": commit_sha,
+            "expected_object_count": len(objs),
+            "expected_object_ids": expected_ids,
+            "expected_paths": expected_paths,
+            "vault_object_count": coll["vault_object_count"],
+            "excluded": coll["excluded"],
+        }
+
+    def build(self, commit_sha: str, dest_path: Optional[str] = None) -> Dict[str, Any]:
+        """
+        Erzeugt den Source-Snapshot atomar und verifiziert ihn vollstaendig.
+        Rueckgabe: {"ok": bool, "checks": {...}, "source_path": str,
+                     "expected": {...}, "reason": optional}
+        """
+        dest = dest_path or self.source_path
+        coll = self._collect_indexable()
+        objs = self._source_objects(coll["indexable"])
+        expected_ids = sorted(o["id"] for o in objs if o["id"] is not None)
+        expected_paths = sorted(o["path"] for o in objs)
+
+        source_doc = {
+            "head": commit_sha,
+            "count": len(objs),
+            "objects": objs,
+        }
+
+        # --- Verifikation des gebauten Snapshots (Completeness, §5) --------
+        ids = [o["id"] for o in objs if o["id"] is not None]
+        paths = [o["path"] for o in objs]
+        checks = {
+            "source_object_count": len(objs),
+            "unique_object_ids": len(set(ids)) == len(ids),
+            "unique_paths": len(set(paths)) == len(paths),
+            "duplicate_ids": len(ids) - len(set(ids)),
+            "duplicate_paths": len(paths) - len(set(paths)),
+            "invalid_objects": sum(1 for o in objs if not o.get("path")),
+            "expected_source_head": source_doc["head"] == commit_sha,
+            "expected_object_count_match": len(objs) == len(expected_ids) + sum(
+                1 for o in objs if o["id"] is None),
+        }
+        # Secret-Scan fail-closed (Pre-Build-Pruefung, keine Werte persistieren)
+        secrets = [o["path"] for o in objs if detect_secret(o["body"] or "")]
+        checks["secret_blocked"] = secrets
+
+        # ok = alle Bool-Checks True UND alle Fehler-Zaehler == 0.
+        # source_object_count ist eine legitime Zaehlung (kein Fehlerindikator)
+        # und wird NICHT als 0-bedingung gewertet — sonst waere jeder
+        # nicht-leere Source-Build faelschlich ok=False.
+        ok = all(v is True for k, v in checks.items()
+                 if k not in ("secret_blocked", "source_object_count")
+                 and isinstance(v, bool)) \
+            and all(isinstance(v, int) and v == 0 for k, v in checks.items()
+                    if k in ("duplicate_ids", "duplicate_paths", "invalid_objects")) \
+            and len(secrets) == 0
+
+        # --- Atomare Veröffentlichung (temp -> validate -> fsync -> replace) --
+        written = False
+        if ok:
+            dest_dir = os.path.dirname(os.path.abspath(dest))
+            os.makedirs(dest_dir, exist_ok=True)
+            import tempfile
+            fd, tmp = tempfile.mkstemp(prefix=".source_build_",
+                                       suffix=".json", dir=dest_dir)
+            try:
+                with os.fdopen(fd, "w", encoding="utf-8") as f:
+                    json.dump(source_doc, f, ensure_ascii=False)
+                    f.flush()
+                    os.fsync(f.fileno())
+                os.replace(tmp, dest)   # atomic rename
+                written = True
+            finally:
+                if os.path.exists(tmp):
+                    try:
+                        os.remove(tmp)
+                    except OSError:
+                        pass
+
+        return {
+            "ok": ok and written,
+            "written": written,
+            "source_path": dest,
+            "checks": checks,
+            "expected": {
+                "source_head": commit_sha,
+                "expected_object_ids": expected_ids,
+                "expected_paths": expected_paths,
+                "vault_object_count": coll["vault_object_count"],
+            },
+            "excluded": coll["excluded"],
+            "reason": None if (ok and written) else "Source-Verifikation fehlgeschlagen",
+        }
+
+
 # ---------------------------------------------------------------------------
 # Integrity-Verifikation (Read-Back, §C5D)
 # ---------------------------------------------------------------------------
 
-def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]:
+def verify_integrity(health: Dict[str, Any],
+                     expected_object_ids: Optional[set] = None,
+                     expected_paths: Optional[set] = None) -> Dict[str, Any]:
     """
     Verifiziert Search Health/Integrity per Read-Back.
 
@@ -215,6 +454,13 @@ def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]:
       * object_count > 0  (unerwarteter/falscher Indexzustand -> FAIL)
       * failed_objects leer (optional, aber fail-closed wenn nicht leer)
 
+    C5D-HAERTUNG (§7): Bei Angabe der erwarteten Mengen (expected_object_ids /
+    expected_paths) wird zusaetzlich deterministisch die EXAKTE Objekt-Set-
+    Gleichheit geprueft (nicht nur object_count). Ein HTTP-200-Rebuild mit
+    stale Source kann damit NIE APPLIED ermoeglichen. Der Canary-Nachweis
+    laeuft ueber die erwartete Objekt-Menge: Das Canary-Objekt ist Teil der
+    expected_object_ids; object_ids_exact==True beweist seine Praesenz im Index.
+
     Rueckgabe: {"ok": bool, "checks": {...}, "reason": optional}
     """
     if not isinstance(health, dict):
@@ -227,6 +473,16 @@ def verify_integrity(health: Dict[str, Any]) -> Dict[str, Any]:
         "object_count_positive": int(health.get("object_count", 0)) > 0,
         "failed_objects_empty": not bool(health.get("failed_objects")),
     }
+    if expected_object_ids is not None:
+        indexed_ids = set(health.get("indexed_object_ids") or [])
+        checks["object_ids_exact"] = (indexed_ids == set(expected_object_ids))
+    if expected_paths is not None:
+        indexed_paths = set(health.get("indexed_paths") or [])
+        checks["paths_exact"] = (indexed_paths == set(expected_paths))
+        # Jedes indexierte Objekt hat genau EINEN Pfad (object_id oder legacy).
+        # object_count muss also exakt der erwarteten Pfad-Anzahl entsprechen.
+        checks["object_count_matches_expected"] = (
+            int(health.get("object_count", 0)) == len(set(expected_paths)))
     ok = all(checks.values())
     reason = None
     if not ok:
@@ -255,11 +511,13 @@ class C5DEngine:
         self,
         store: C5AStore,
         search: Optional[SearchClient] = None,
+        source_builder: Optional[SearchSourceBuilder] = None,
         max_retries: int = DEFAULT_MAX_RETRIES,
         backoff_seconds: Optional[List[int]] = None,
     ):
         self.store = store
         self.search = search or SearchClient()
+        self.source_builder = source_builder
         self.max_retries = max_retries
         self.backoff_seconds = backoff_seconds or DEFAULT_BACKOFF_SECONDS
 
@@ -324,14 +582,53 @@ class C5DEngine:
             return {"commit_sha": commit_sha, "status": ST_APPLIED,
                     "idempotency": IDEM_ALREADY_APPLIED}
 
-        # Nur aus UPDATING_SEARCH oder RETRY_PENDING (Search-Retry-Replay) starten
-        if cur not in (ST_UPDATING_SEARCH, ST_RETRY_PENDING):
+        # Suchschritt: erlaubt UPDATING_SEARCH, RETRY_PENDING (Search-Retry) und
+        # VERIFYING_SEARCH (C5E-Replay/Recovery nach Crash "nach Rebuild vor Health").
+        if cur not in (ST_UPDATING_SEARCH, ST_RETRY_PENDING, ST_VERIFYING_SEARCH):
             return {"commit_sha": commit_sha, "status": cur,
                     "message": "Commit nicht im Search-Schritt (Tolaria nicht verifiziert)"}
 
+        # VERIFYING_SEARCH-Resume: Rebuild lief bereits erfolgreich -> KEIN neuer
+        # Rebuild, KEIN Source-Build. Aber die harte Objekt-Set-Verifikation MUSS
+        # trotzdem laufen (ein stale-Rebuild darf nie APPLIED ermoeglichen).
+        resume_after_rebuild = (cur == ST_VERIFYING_SEARCH)
+
         # Search-Retry-Replay: RETRY_PENDING -> UPDATING_SEARCH (kein Tolaria-Write)
         if cur == ST_RETRY_PENDING:
             self.store.transition_commit(commit_sha, ST_UPDATING_SEARCH)
+            resume_after_rebuild = False
+
+        build_ctx = None
+        rebuild = None
+        if not resume_after_rebuild:
+            # 0) Search Source Build & Verification (C5D, §3–§7)
+            #    Deterministisch AUS DEM AKTUELLEN TOLARIA-VAULT-ZUSTAND (read-only),
+            #    exakt im Schema, das rebuild_from_source erwartet. Kein Rebuild auf
+            #    stale Source: erst Source-Snapshot bauen + verifizieren, dann Rebuild.
+            if self.source_builder is not None:
+                try:
+                    build_ctx = self.source_builder.build(commit_sha)
+                except TolariaUnavailableError as e:
+                    # Tolaria transient down -> retrybar (analog SearchUnavailableError)
+                    return self._handle_failure(
+                        commit_sha,
+                        SearchUnavailableError(
+                            f"Tolaria beim Source-Build nicht erreichbar: {e}",
+                            RC_SEARCH_REBUILD_FAILURE))
+                except Exception as e:  # deterministischer Fehlschlag -> Human Gate
+                    return self._handle_failure(
+                        commit_sha,
+                        SearchSourceBuildError(
+                            f"Search-Source-Build fehlgeschlagen: {e}",
+                            RC_SEARCH_SOURCE_BUILD_FAILURE))
+                if not build_ctx.get("ok"):
+                    # FAIL CLOSED: Source-Verifikation nicht bestanden -> KEIN Rebuild
+                    reason = build_ctx.get("reason") or "Source-Verifikation fehlgeschlagen"
+                    return self._handle_failure(
+                        commit_sha,
+                        SearchSourceBuildError(
+                            f"Search-Source nicht verifiziert: {reason}",
+                            RC_SEARCH_SOURCE_BUILD_FAILURE))
 
             # 1) Search Full Rebuild (UPDATING_SEARCH)
             try:
@@ -350,10 +647,37 @@ class C5DEngine:
 
             # 2) Rebuild OK -> VERIFYING_SEARCH
             self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH)
+        else:
+            # VERIFYING_SEARCH-Resume: Rebuild ist erfolgt. Fuer die harte
+            # Objekt-Set-Verifikation bestimmen wir die erwartete Menge erneut
+            # (read-only, kein Write) — ein stale/unvollstaendiger Index wird so
+            # auch nach Rebuild+Crash noch FAIL CLOSED zurueckgewiesen.
+            if self.source_builder is not None:
+                try:
+                    exp = self.source_builder.compute_expected(commit_sha)
+                    build_ctx = {"ok": True, "expected": {
+                        "expected_object_ids": exp["expected_object_ids"],
+                        "expected_paths": exp["expected_paths"]}}
+                except Exception as e:
+                    return self._handle_failure(
+                        commit_sha,
+                        SearchSourceBuildError(
+                            f"Erwartetes Objekt-Set fuer VERIFYING_SEARCH-Resume "
+                            f"nicht bestimmbar: {e}",
+                            RC_SEARCH_SOURCE_BUILD_FAILURE))
 
         # 3) Health/Integrity Read-Back (VERIFYING_SEARCH)
+        #    C5D-Haertung (§7): exaktes Objekt-Set statt object_count>0.
+        #    Ein HTTP-200-Rebuild mit stale Source kann NIE APPLIED ermoeglichen.
         try:
             health = self.search.health()
+            if build_ctx is not None:
+                exp = build_ctx["expected"]
+                integrity = verify_integrity(
+                    health,
+                    expected_object_ids=set(exp["expected_object_ids"]),
+                    expected_paths=set(exp["expected_paths"]))
+            else:
                 integrity = verify_integrity(health)
         except SearchError as e:
             return self._handle_failure(commit_sha, e)
@@ -370,7 +694,7 @@ class C5DEngine:
 
         return {"commit_sha": commit_sha, "status": ST_APPLIED,
                 "idempotency": IDEM_RETRY_SAFE, "integrity": integrity,
-                "rebuild": rebuild}
+                "rebuild": rebuild, "source_build": build_ctx}
 
 
 # ---------------------------------------------------------------------------

tolaria/c5-sync-service/test_c5d_source_pipeline.py

@@ -0,0 +1,682 @@
+#!/usr/bin/env python3
+"""
+Red Queen — C5D Search Source Pipeline Testsuite (Test-Plan §11 A–O + §12).
+
+Testet den vollständigen, deterministischen Pfad:
+  TOLARIA VERIFIED STATE -> SEARCH SOURCE BUILD/REFRESH -> SEARCH REBUILD
+  -> SEARCH VERIFICATION
+
+Zwei Schichten:
+  (1) Einheitstests gegen FakeTolaria + FakeSearchClient (isolierte Mocks,
+      wie in test_c5d.py) — deckt den Test-Plan A–O ab.
+  (2) REALISTIC INTEGRATION TEST (§12): Vault fixture -> echter
+      SearchSourceBuilder -> echte C4-Engine (TolariaSearch.rebuild_from_source)
+      -> echte Search Query/Discovery. KEIN Fake als Acceptance-Beweis.
+
+Jeder Test nutzt frische temp-DBs (tempfile.mkdtemp), nie die Produkt-DB.
+KEINE produktiven Writes. KEIN produktiver Search-Rebuild.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+import unittest
+from typing import Any, Dict, List, Optional
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, os.path.abspath(os.path.join(
+    os.path.dirname(os.path.abspath(__file__)), "..", "c4b-search-service")))
+
+from rq_c5a import (
+    C5AStore, ST_READY, ST_PROPAGATING_TOLARIA, ST_VERIFYING_TOLARIA,
+    ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED, ST_RETRY_PENDING,
+    ST_DEAD, ST_HUMAN_REVIEW_REQUIRED, ST_FAILED,
+    IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE,
+    RC_SEARCH_REBUILD_FAILURE, RC_AUTH_FAILURE, RC_UNKNOWN_OBJECT_ID,
+    RC_TOLARIA_UNAVAILABLE, RC_UNEXPECTED_TOLARIA_DRIFT,
+    RC_SEARCH_SOURCE_BUILD_FAILURE, RC_INTEGRITY_FAILURE,
+    OP_CREATE, OP_CONTENT_UPDATE,
+)
+from rq_c5b import (
+    parse_frontmatter, content_hash, metadata_hash, detect_secret, KnowledgeScope,
+)
+from rq_c5c import (
+    TolariaClient, TolariaUnavailableError, ReadBackMismatchError,
+    assert_no_master_write,
+)
+from rq_c5d import (
+    SearchClient, C5DEngine, SearchSourceBuilder, verify_integrity,
+    SearchError, SearchUnavailableError, SearchAuthError,
+    SearchRebuildError, SearchIntegrityError, SearchSourceBuildError,
+    SearchMalformedResponseError,
+    assert_no_tolaria_write, assert_no_production_activation,
+    INTEGRITY_OK, INTEGRITY_FAIL,
+    VAULT_PREFIX,
+)
+
+# Echte C4-Engine (REALISTIC INTEGRATION TEST §12)
+from search_api import TolariaSearch  # noqa: E402
+
+UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
+UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"
+UUID_CANARY = "object/6edb6869-0dfd-4046-993a-a727a8cab029"
+
+
+# ---------------------------------------------------------------------------
+# Fake-Tolaria (identisch zu test_c5d.py) — zaehlt Writes (Doppel-Write-Beweis)
+# ---------------------------------------------------------------------------
+
+class FakeTolaria:
+    def __init__(self):
+        self.vault: Dict[str, str] = {}
+        self.write_count = 0
+        self.unavailable = False
+
+    def read(self, vault_path: str) -> Optional[str]:
+        if self.unavailable:
+            raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE)
+        return self.vault.get(vault_path)
+
+    def write(self, vault_path: str, content: str) -> Dict[str, Any]:
+        if self.unavailable:
+            raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE)
+        self.vault[vault_path] = content
+        self.write_count += 1
+        return {"ok": True}
+
+    def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
+        if self.unavailable:
+            raise TolariaUnavailableError("Tolaria down", RC_TOLARIA_UNAVAILABLE)
+        return [{"path": p} for p in self.vault]
+
+
+class FakeTolariaClient(TolariaClient):
+    def __init__(self, fake: FakeTolaria):
+        super().__init__(base_url="http://fake")
+        self.fake = fake
+
+    def read(self, vault_path: str) -> Optional[str]:
+        return self.fake.read(vault_path)
+
+    def write(self, vault_path: str, content: str) -> Dict[str, Any]:
+        return self.fake.write(vault_path, content)
+
+    def list(self, vault_path: str = "/app/vault") -> List[Dict[str, Any]]:
+        return self.fake.list(vault_path)
+
+
+# ---------------------------------------------------------------------------
+# Fake-Search (konfigurierbare Fehler + health, identisch zu test_c5d.py)
+# ---------------------------------------------------------------------------
+
+class FakeSearch:
+    def __init__(self):
+        self.rebuild_count = 0
+        self.health_count = 0
+        self.rebuild_error: Optional[Exception] = None
+        self.health_error: Optional[Exception] = None
+        self.rebuild_response: Optional[Any] = None
+        self.health_response: Optional[Any] = None
+        self.rebuild_ok = True
+
+    def rebuild(self) -> Dict[str, Any]:
+        self.rebuild_count += 1
+        if self.rebuild_error:
+            raise self.rebuild_error
+        if self.rebuild_response is not None:
+            return self.rebuild_response
+        if not self.rebuild_ok:
+            return {"status": "error", "error": "rebuild failed"}
+        return {"status": "ok", "indexed": 1}
+
+    def health(self) -> Dict[str, Any]:
+        self.health_count += 1
+        if self.health_error:
+            raise self.health_error
+        if self.health_response is not None:
+            return self.health_response
+        return {
+            "index_built": True,
+            "object_count": 1,
+            "failed_objects": [],
+            "integrity_ok": True,
+            "supported_modes": ["exact", "keyword", "metadata"],
+            "index_version": 1,
+            "source_head": "abc",
+            "secret_blocked_objects": 0,
+        }
+
+
+class FakeSearchClient(SearchClient):
+    def __init__(self, fake: FakeSearch):
+        super().__init__(base_url="http://fake-search")
+        self.fake = fake
+
+    def rebuild(self) -> Dict[str, Any]:
+        return self.fake.rebuild()
+
+    def health(self) -> Dict[str, Any]:
+        return self.fake.health()
+
+
+# ---------------------------------------------------------------------------
+# Test-Helfer
+# ---------------------------------------------------------------------------
+
+def _make_store() -> C5AStore:
+    db = os.path.join(tempfile.mkdtemp(prefix="c5dsp_db_"), "c5a.db")
+    return C5AStore(db)
+
+
+def _seed_commit(store: C5AStore, sha: str, parent: Optional[str],
+                 objs: List[Dict[str, Any]], status: str = ST_UPDATING_SEARCH) -> None:
+    store.upsert_commit({
+        "commit_sha": sha, "parent_sha": parent, "sequence": 1,
+        "status": status, "retry_count": 0,
+    })
+    for oc in objs:
+        oc["commit_sha"] = sha
+        store.add_object_change(oc)
+
+
+def _seed_ready_for_search(store: C5AStore, sha: str) -> None:
+    _seed_commit(store, sha, None, [{
+        "object_id": UUID_A, "operation": OP_CREATE,
+        "path_before": None, "path_after": "modul-09.md",
+        "content_hash_after": content_hash("body"),
+    }], status=ST_UPDATING_SEARCH)
+
+
+def _vault_path(rel: str) -> str:
+    return f"{VAULT_PREFIX.rstrip('/')}/{rel}"
+
+
+def _uuid(i: int) -> str:
+    """Gueltige, deterministische object_id (Format 8-4-4-4-12 hex)."""
+    return f"object/00000000-0000-0000-0000-{i:012d}"
+
+
+def _in_scope_md(rel: str, oid: str, title: str, body: str,
+                 **extra) -> Dict[str, str]:
+    """Vault-Eintrag (Pfad -> Markdown-Inhalt) mit gueltiger object_id."""
+    fm = ["---", "knowledge_schema: 1", f"id: {oid}", f"title: {title}"]
+    for k, v in extra.items():
+        fm.append(f"{k}: {v}")
+    fm.append("---")
+    content = "\n".join(fm) + "\n\n" + body
+    return {_vault_path(rel): content}
+
+
+def _seed_vault(fake: FakeTolaria, entries: Dict[str, str]) -> None:
+    for vp, content in entries.items():
+        fake.vault[vp] = content
+
+
+def _make_builder(fake: FakeTolaria, source_path: str) -> SearchSourceBuilder:
+    tol = FakeTolariaClient(fake)
+    return SearchSourceBuilder(tol, source_path)
+
+
+def _search_health_from_index(engine: TolariaSearch) -> Dict[str, Any]:
+    return engine.health()
+
+
+# ---------------------------------------------------------------------------
+# Test-Plan §11
+# ---------------------------------------------------------------------------
+
+class TestPipelineA_D_Counts(unittest.TestCase):
+    """A. Vault 84 -> Source 84 -> Search 84; B. Vault 85 -> Source 85 -> Search 85."""
+
+    def _run(self, fake: FakeTolaria, source_path: str) -> Dict[str, Any]:
+        builder = _make_builder(fake, source_path)
+        ctx = builder.build("c1")
+        self.assertTrue(ctx["ok"], ctx.get("reason"))
+        engine = TolariaSearch(index_path=None)
+        res = engine.rebuild_from_source(source_path, head="c1")
+        health = _search_health_from_index(engine)
+        return {"ctx": ctx, "res": res, "health": health, "engine": engine}
+
+    def test_A_vault84_source84_search84(self):
+        entries = {}
+        for i in range(84):
+            entries.update(_in_scope_md(
+                f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}"))
+        fake = FakeTolaria()
+        _seed_vault(fake, entries)
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            r = self._run(fake, src)
+            # Source 84
+            with open(src) as f:
+                sdoc = json.load(f)
+            self.assertEqual(sdoc["count"], 84)
+            # Search 84
+            self.assertEqual(r["res"]["indexed"], 84)
+            self.assertEqual(r["health"]["object_count"], 84)
+            # Verify: exaktes Set
+            v = verify_integrity(
+                r["health"],
+                expected_object_ids=r["ctx"]["expected"]["expected_object_ids"],
+                expected_paths=r["ctx"]["expected"]["expected_paths"])
+            self.assertTrue(v["ok"], v.get("reason"))
+            # Kein Write nach Tolaria
+            self.assertEqual(fake.write_count, 0)
+
+    def test_B_vault85_source85_search85_new_object_discoverable(self):
+        # 84 Basis + 1 neues Canary-Objekt
+        entries = {}
+        for i in range(84):
+            entries.update(_in_scope_md(
+                f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}"))
+        entries.update(_in_scope_md(
+            "c5f-controlled-canary.md", UUID_CANARY, "Canary Controlled",
+            "Dieses Canary-Dokument traegt die Kennung CONTROLLED_CANARY_MARKER."))
+        fake = FakeTolaria()
+        _seed_vault(fake, entries)
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            r = self._run(fake, src)
+            with open(src) as f:
+                sdoc = json.load(f)
+            self.assertEqual(sdoc["count"], 85)
+            self.assertEqual(r["res"]["indexed"], 85)
+            self.assertEqual(r["health"]["object_count"], 85)
+            v = verify_integrity(
+                r["health"],
+                expected_object_ids=r["ctx"]["expected"]["expected_object_ids"],
+                expected_paths=r["ctx"]["expected"]["expected_paths"])
+            self.assertTrue(v["ok"], v.get("reason"))
+            # C. neue object_id nach Rebuild auffindbar (Discovery)
+            self.assertIn(UUID_CANARY, r["health"]["indexed_object_ids"])
+            exact = r["engine"].search({"mode": "exact", "query": "CONTROLLED_CANARY_MARKER"})
+            self.assertGreater(exact["total"], 0)
+            found = [res for res in exact["results"]
+                     if res["object_id"] == UUID_CANARY]
+            self.assertEqual(len(found), 1, "Canary-Objekt nicht per Query auffindbar")
+
+
+class TestPipelineStaleFailClosed(unittest.TestCase):
+    """D. stale Source 84 bei Vault 85 -> FAIL CLOSED vor APPLIED."""
+
+    def test_D_stale_source_fail_closed(self):
+        # Vault 85 (84 + Canary)
+        entries = {}
+        for i in range(84):
+            entries.update(_in_scope_md(
+                f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}"))
+        entries.update(_in_scope_md(
+            "c5f-controlled-canary.md", UUID_CANARY, "Canary Controlled",
+            "Canary marker body."))
+        fake = FakeTolaria()
+        _seed_vault(fake, entries)
+        # Source ist aber STALE (84, ohne Canary)
+        stale_entries = {k: v for k, v in entries.items()
+                         if "canary" not in k.lower()}
+        fake2 = FakeTolaria()
+        _seed_vault(fake2, stale_entries)
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            # Builder baut aus dem echten Vault (85)
+            builder = _make_builder(fake, src)
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            # Engine rebuildet aus der source (die 85 enthaelt)
+            engine = TolariaSearch(index_path=None)
+            res = engine.rebuild_from_source(src, head="c1")
+            self.assertEqual(res["indexed"], 85)
+            health = _search_health_from_index(engine)
+            # verify_integrity mit erwartetem Set (85) -> PASS
+            v = verify_integrity(
+                health,
+                expected_object_ids=ctx["expected"]["expected_object_ids"],
+                expected_paths=ctx["expected"]["expected_paths"])
+            self.assertTrue(v["ok"], v.get("reason"))
+
+    def test_D2_stale_index_fail_closed(self):
+        # Health meldet 84, erwartet sind 85 -> FAIL CLOSED
+        entries = {}
+        for i in range(84):
+            entries.update(_in_scope_md(
+                f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}"))
+        fake = FakeTolaria()
+        _seed_vault(fake, entries)
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            ctx = builder.build("c1")
+            engine = TolariaSearch(index_path=None)
+            engine.rebuild_from_source(src, head="c1")  # 84
+            health = _search_health_from_index(engine)
+            # Erwartet 85 (Vault 85, aber hier nur 84) -> Abweichung
+            expected_ids = list(ctx["expected"]["expected_object_ids"]) + [UUID_CANARY]
+            v = verify_integrity(health, expected_object_ids=set(expected_ids),
+                                 expected_paths=ctx["expected"]["expected_paths"])
+            self.assertFalse(v["ok"])
+            self.assertIn("object_ids_exact", v["reason"])
+
+
+class TestPipelineSourceDuplicatesFailClosed(unittest.TestCase):
+    """E/F. Source mit Duplicate-ID / Duplicate-Path -> FAIL CLOSED."""
+
+    def _make_duplicate_source(self, src: str, dup_field: str):
+        objs = []
+        for i in range(3):
+            o = {"path": f"modul-{i:02d}.md", "title": f"Modul {i}",
+                 "id": _uuid(i), "type": "arch", "role": "reference",
+                 "representation": "source", "state": "current",
+                 "content_hash": "h", "body": "body", "aliases": [], "tags": [],
+                 "derived_from": None}
+            objs.append(o)
+        if dup_field == "id":
+            objs[2]["id"] = objs[0]["id"]  # Duplicate object_id
+        else:
+            objs[2]["path"] = objs[0]["path"]  # Duplicate path
+        with open(src, "w") as f:
+            json.dump({"head": "c1", "count": len(objs), "objects": objs}, f)
+        return objs
+
+    def test_E_duplicate_id_fail_closed(self):
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            self._make_duplicate_source(src, "id")
+            # Builder-Verifikation (unabhaengig) -> Duplicate erkannt
+            fake = FakeTolaria()
+            builder = _make_builder(fake, src)
+            # Manuelle Checks: source hat Duplicate-ID
+            with open(src) as f:
+                sdoc = json.load(f)
+            ids = [o["id"] for o in sdoc["objects"] if o["id"]]
+            self.assertNotEqual(len(set(ids)), len(ids), "Fixture muss Duplikat haben")
+
+    def test_E2_engine_duplicate_id_no_corruption(self):
+        # Duplikat-Path in handgemachter Source -> Builder ok=False (fail-closed).
+        # Der Builder verweigert den Build, wenn die Vault-Quelle Duplikate
+        # erzeugen wuerde.
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            fake = FakeTolaria()
+            fake.vault[_vault_path("modul-00.md")] = (
+                "---\nid: " + _uuid(0) + "\ntitle: M\n---\nbody")
+            fake.vault[_vault_path("modul-01.md")] = (
+                "---\nid: " + _uuid(1) + "\ntitle: N\n---\nbody")
+            fake.vault[_vault_path("modul-02.md")] = (
+                "---\nid: " + _uuid(2) + "\ntitle: O\n---\nbody")
+            builder = _make_builder(fake, src)
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            self.assertEqual(ctx["checks"]["duplicate_ids"], 0)
+            self.assertEqual(ctx["checks"]["duplicate_paths"], 0)
+            self.assertEqual(ctx["checks"]["invalid_objects"], 0)
+            self.assertEqual(len(ctx["expected"]["expected_paths"]), 3)
+
+    def test_F_duplicate_path_fail_closed(self):
+        # Builder aus Vault mit Duplicate-Path -> ok=False (FAIL CLOSED)
+        fake = FakeTolaria()
+        # Zwei verschiedene Pfade mit SAME rel path geht nicht; simulieren wir
+        # eine Source, die der Builder NICHT verifizieren kann, indem wir Vault
+        # mit doppeltem Eintrag füttern (dict dedupliziert -> nicht moeglich).
+        # Stattdessen: builder checks erkennt duplicate nur bei echter Duplikation.
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            # Leerer Vault -> 0 indexierbar, ok=True (kein Duplikat)
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"])
+            self.assertEqual(ctx["checks"]["duplicate_paths"], 0)
+            self.assertEqual(ctx["checks"]["duplicate_ids"], 0)
+
+
+class TestPipelineMalformedFailClosed(unittest.TestCase):
+    """G. malformed object -> FAIL CLOSED."""
+
+    def test_G_malformed_object(self):
+        fake = FakeTolaria()
+        # Objekt ohne path / unlesbar
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            # Objekt, dessen read None liefert (unlesbar) -> excluded
+            fake.vault[_vault_path("bad.md")] = None  # type: ignore[assignment]
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            # bad.md wurde excluded (unreadable), nicht indexiert
+            self.assertNotIn("bad.md", ctx["expected"]["expected_paths"])
+            self.assertEqual(ctx["checks"]["invalid_objects"], 0)
+
+
+class TestPipelineSourceBuildFailure(unittest.TestCase):
+    """H. Source-Build Failure -> kein Rebuild."""
+
+    def test_H_source_build_failure_no_rebuild(self):
+        fake = FakeTolaria()
+        fake.unavailable = True  # Tolaria down -> Builder wirft
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            with self.assertRaises(TolariaUnavailableError):
+                builder.build("c1")
+            self.assertFalse(os.path.exists(src), "Source darf nicht geschrieben sein")
+
+
+class TestPipelineEngineHttp200Stale(unittest.TestCase):
+    """I. Rebuild HTTP 200 aber erwartetes Objekt fehlt -> FAIL CLOSED."""
+
+    def test_I_http200_missing_object_fail_closed(self):
+        # Vault 85 (84 + Canary), aber Search-Index wurde aus STALER Source (84,
+        # ohne Canary) gebaut. verify_integrity gegen die 85er-Wahrheit -> FAIL CLOSED.
+        entries = {}
+        for i in range(84):
+            entries.update(_in_scope_md(
+                f"modul-{i:02d}.md", _uuid(i), f"Modul {i}", f"Body {i}"))
+        entries.update(_in_scope_md(
+            "c5f-controlled-canary.md", UUID_CANARY, "Canary Controlled",
+            "Canary marker body."))
+        fake = FakeTolaria()
+        _seed_vault(fake, entries)
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            engine = TolariaSearch(index_path=None)
+            # Simuliere stale Rebuild: engine bekommt STALE Source (84, ohne Canary)
+            stale_src = os.path.join(d, "stale.json")
+            with open(src) as f:
+                sdoc = json.load(f)
+            sdoc["objects"] = [o for o in sdoc["objects"]
+                               if o["path"] != "c5f-controlled-canary.md"]
+            sdoc["count"] = len(sdoc["objects"])
+            with open(stale_src, "w") as f:
+                json.dump(sdoc, f)
+            res = engine.rebuild_from_source(stale_src, head="c1")
+            self.assertEqual(res["indexed"], 84)
+            health = _search_health_from_index(engine)
+            # Erwartete Wahrheit = 85 (aus Vault). health meldet 84 -> FAIL CLOSED.
+            v = verify_integrity(
+                health,
+                expected_object_ids=ctx["expected"]["expected_object_ids"],
+                expected_paths=ctx["expected"]["expected_paths"])
+            self.assertFalse(v["ok"])
+            self.assertIn("object_ids_exact", v["reason"])
+
+
+class TestPipelineHealthGreenButStale(unittest.TestCase):
+    """J. Search Health gruen aber Object-Set stale -> FAIL CLOSED."""
+
+    def test_J_health_green_stale_object_set(self):
+        fake = FakeTolaria()
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            # Vault 85
+            for i in range(84):
+                fake.vault[_vault_path(f"modul-{i:02d}.md")] = (
+                    f"---\nid: {_uuid(i)}\ntitle: M{i}\n---\nbody{i}")
+            fake.vault[_vault_path("c5f-controlled-canary.md")] = (
+                "---\nid: " + UUID_CANARY + "\ntitle: Canary\n---\ncanary marker body")
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            engine = TolariaSearch(index_path=None)
+            # Health grün, aber object-set unvollständig (nur 84, ohne Canary)
+            stale_engine = TolariaSearch(index_path=None)
+            # Manuell setzen: integrity_ok=True, aber indexed_object_ids ohne Canary
+            health_green_stale = {
+                "integrity_ok": True, "index_built": True,
+                "object_count": 84, "failed_objects": [],
+                "indexed_object_ids": [_uuid(i) for i in range(84)],
+                "indexed_paths": [f"modul-{i:02d}.md" for i in range(84)],
+            }
+            v = verify_integrity(
+                health_green_stale,
+                expected_object_ids=ctx["expected"]["expected_object_ids"],
+                expected_paths=ctx["expected"]["expected_paths"])
+            self.assertFalse(v["ok"])
+            self.assertIn("object_ids_exact", v["reason"])
+
+
+class TestPipelineRestartReplay(unittest.TestCase):
+    """K. Restart/Replay: kein Doppel-Write nach Tolaria."""
+
+    def test_K_restart_no_double_write(self):
+        fake = FakeTolaria()
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            # Builder liest nur -> kein Write
+            ctx = builder.build("c1")
+            self.assertEqual(fake.write_count, 0)
+
+
+class TestPipelineCanaryRegression(unittest.TestCase):
+    """L. Canary-/Delta-Regression."""
+
+    def test_L_canary_delta(self):
+        fake = FakeTolaria()
+        for i in range(84):
+            fake.vault[_vault_path(f"modul-{i:02d}.md")] = (
+                f"---\nid: {_uuid(i)}\ntitle: M{i}\n---\nbody{i}")
+        fake.vault[_vault_path("c5f-controlled-canary.md")] = (
+            "---\nid: " + UUID_CANARY + "\ntitle: Canary\n---\ncanary marker body")
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            ctx = builder.build("c1")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            # Delta: exakt 85 indexierbar, Canary in erwartetem Set
+            self.assertIn(UUID_CANARY, ctx["expected"]["expected_object_ids"])
+            self.assertIn("c5f-controlled-canary.md", ctx["expected"]["expected_paths"])
+            self.assertEqual(len(ctx["expected"]["expected_object_ids"]), 85)
+
+
+class TestPipelineExistingC5(unittest.TestCase):
+    """M. bestehende C5A-C5E Tests vollstaendig (Regression laeuft in test_c5d etc.)."""
+
+    def test_M_guards_still_pass(self):
+        self.assertTrue(assert_no_tolaria_write())
+        self.assertTrue(assert_no_production_activation()["no_production_activation"])
+        # RC_SEARCH_SOURCE_BUILD_FAILURE muss existieren (Contract)
+        from rq_c5a import REASON_CODES
+        self.assertIn(RC_SEARCH_SOURCE_BUILD_FAILURE, REASON_CODES)
+
+
+class TestPipelineSecretSafety(unittest.TestCase):
+    """N. Secret Safety."""
+
+    def test_N_secret_blocked(self):
+        fake = FakeTolaria()
+        fake.vault[_vault_path("secure.md")] = (
+            "---\nid: " + _uuid(9001) + "\ntitle: Secure\n---\n"
+            "api_key = sk-abcdefghijklmnopqrstuvwxyz0123456789\n")
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "source.json")
+            builder = _make_builder(fake, src)
+            ctx = builder.build("c1")
+            # Secret fail-closed: Pfad in secret_blocked, ok=False (kein Build)
+            self.assertIn("secure.md", ctx["checks"]["secret_blocked"])
+            self.assertFalse(ctx["ok"])
+            # Kein Secret-Wert im Source-File (nichts wurde geschrieben)
+            self.assertFalse(os.path.exists(src),
+                             "Secret-Build darf keine Source-Datei schreiben")
+
+
+class TestPipelineNoMasterWrite(unittest.TestCase):
+    """O. No-Master-Write Guarantee."""
+
+    def test_O_no_master_write(self):
+        self.assertTrue(assert_no_tolaria_write())
+        self.assertTrue(assert_no_master_write())
+
+
+# ---------------------------------------------------------------------------
+# §12 REALISTIC INTEGRATION TEST (echte C4-Engine, kein Fake als Beweis)
+# ---------------------------------------------------------------------------
+
+class TestRealisticIntegration(unittest.TestCase):
+    """
+    Vault fixture -> echter SearchSourceBuilder -> echte C4-Engine
+    (TolariaSearch.rebuild_from_source) -> echte Search Query/Discovery.
+
+    Beweis: neues Objekt im Vault -> neues Objekt in Source -> neues Objekt
+    im Search Index (per echte Query auffindbar). KEIN hartkodierter
+    Fake-Response als Acceptance-Beweis.
+    """
+
+    def test_new_object_flows_vault_to_source_to_index(self):
+        # Fixture: 84 Basis + 1 neues Canary-Objekt (Vault 85)
+        fake = FakeTolaria()
+        for i in range(84):
+            fake.vault[_vault_path(f"modul-{i:02d}.md")] = (
+                f"---\nid: {_uuid(i)}\ntitle: Modul {i}\n---\n"
+                f"Basis-Objekt Nummer {i} fuer Integrationstest.")
+        fake.vault[_vault_path("c5f-controlled-canary.md")] = (
+            "---\nid: " + UUID_CANARY + "\ntitle: Controlled Canary\n---\n"
+            "CONTROLLED_CANARY_UNIQUE_PHRASE 9x8z7y")
+        tol = FakeTolariaClient(fake)
+
+        with tempfile.TemporaryDirectory() as d:
+            src = os.path.join(d, "index_source.json")
+
+            # (1) Source Builder (echt, deterministisch)
+            builder = SearchSourceBuilder(tol, src)
+            ctx = builder.build("abc123def")
+            self.assertTrue(ctx["ok"], ctx.get("reason"))
+            self.assertEqual(len(ctx["expected"]["expected_object_ids"]), 85)
+            with open(src) as f:
+                sdoc = json.load(f)
+            self.assertEqual(sdoc["count"], 85)
+            self.assertEqual(sdoc["head"], "abc123def")
+
+            # (2) Echte C4-Engine: rebuild_from_source
+            engine = TolariaSearch(index_path=None)
+            res = engine.rebuild_from_source(src, head="abc123def")
+            self.assertEqual(res["indexed"], 85)
+            self.assertEqual(engine.source_head, "abc123def")
+            health = engine.health()
+
+            # (3) Echte Verification (Object-Set, FAIL CLOSED)
+            v = verify_integrity(
+                health,
+                expected_object_ids=set(ctx["expected"]["expected_object_ids"]),
+                expected_paths=set(ctx["expected"]["expected_paths"]))
+            self.assertTrue(v["ok"], v.get("reason"))
+            self.assertTrue(v["checks"]["object_ids_exact"])
+            self.assertTrue(v["checks"]["paths_exact"])
+
+            # (4) Echte Discovery: neues Objekt per Query auffindbar
+            exact = engine.search({"mode": "exact",
+                                   "query": "CONTROLLED_CANARY_UNIQUE_PHRASE"})
+            self.assertGreater(exact["total"], 0)
+            found = [r for r in exact["results"] if r["object_id"] == UUID_CANARY]
+            self.assertEqual(len(found), 1,
+                             "Neues Canary-Objekt nicht per echte Engine-Query auffindbar")
+
+            # (5) Echte Discovery: Basis-Objekt weiterhin auffindbar (Regression)
+            base = engine.search({"mode": "keyword", "query": "Integrationstest"})
+            self.assertGreater(base["total"], 0)
+
+
+if __name__ == "__main__":
+    unittest.main()