fix(tolaria): C5D source_provenance persistence contract (OPTION A)

Persistiert die Search-Source-Provenance commit-spezifisch im C5-State
(meta-KV-Key search_source_provenance:<workflow_commit_sha>), WRITE POINT
nach validiertem Source-Build. Adoption prueft ausschliesslich gegen die
persistierte source_provenance, nie gegen workflow_commit_sha oder
current_repo_head. FAIL CLOSED ohne persistierte Provenance. Generischer,
evidence-validierter Recovery-Pfad fuer extern abgeschlossene Builds.
14 neue Contract-Tests (A-N); volle Regression gruen.

tolaria/c5-sync-service/rq_c5a.py

@@ -595,6 +595,66 @@ class C5AStore:
                 (commit_sha, int(time.time() * 1000)),
             )
 
+    # -- Search-Source Provenance Persistence (C5D source_provenance Contract) --
+
+    @staticmethod
+    def _provenance_key(workflow_commit_sha: str) -> str:
+        """
+        Commit-spezifischer meta-KV-Key fuer die Search-Source-Provenance.
+
+        Verhindert Kreuzzuordnung/ueberschreibung bei mehreren pending/replayed
+        Commits: Die Zuordnung workflow_commit_sha -> source_provenance ist
+        deterministisch und eindeutig pro Commit.
+        """
+        return f"search_source_provenance:{workflow_commit_sha}"
+
+    @staticmethod
+    def _provenance_value(source_head: str, source_object_ids: List[str],
+                          source_paths: List[str], recovered: bool = False) -> str:
+        """
+        Serialisiert die persistierte Source-Provenance als einzelner meta-Wert.
+
+        Enthaelt: source_head (Provenance des tatsaechlich erzeugten Source-
+        Snapshots), das exakte erwartete Object-/Path-Set (für die Recovery-
+        Evidence-Validierung) und ein `recovered`-Flag (generischer externer
+        Recovery-Pfad). Keine neue Source of Truth — ein Meta-Value pro Commit.
+        """
+        payload = {
+            "source_head": str(source_head),
+            "source_object_ids": sorted(source_object_ids),
+            "source_paths": sorted(source_paths),
+            "recovered": bool(recovered),
+        }
+        return json.dumps(payload, sort_keys=True)
+
+    def persist_search_source_provenance(
+        self, workflow_commit_sha: str, source_head: str,
+        source_object_ids, source_paths, recovered: bool = False) -> None:
+        """
+        Persistiert die Search-Source-Provenance fuer einen Workflow-Commit
+        (WRITE POINT: NUR nach erfolgreichem, validiertem, atomar publiziertem
+        Source-Build — siehe C5D-Engine; fehlerhafte Builds persistieren NICHTS).
+        """
+        value = self._provenance_value(source_head, source_object_ids, source_paths, recovered)
+        with self._conn:
+            self._conn.execute(
+                "INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)",
+                (self._provenance_key(workflow_commit_sha), value),
+            )
+
+    def get_source_provenance(self, workflow_commit_sha: str) -> Optional[Dict[str, Any]]:
+        """Liest die persistierte Search-Source-Provenance fuer einen Commit (oder None)."""
+        key = self._provenance_key(workflow_commit_sha)
+        row = self._conn.execute(
+            "SELECT value FROM meta WHERE key = ?", (key,)
+        ).fetchone()
+        if row is None:
+            return None
+        try:
+            return json.loads(row["value"])
+        except (ValueError, TypeError):
+            return None
+
     def close(self) -> None:
         self._conn.close()
 

tolaria/c5-sync-service/rq_c5d.py

@@ -343,10 +343,16 @@ class SearchSourceBuilder:
         objs.sort(key=lambda o: o["path"])
         return objs
 
-    def compute_expected(self, commit_sha: str) -> Dict[str, Any]:
+    def compute_expected(self, commit_sha: str,
+                         source_head: Optional[str] = None) -> Dict[str, Any]:
         """
         Read-only: bestimmt die erwartete indexierbare Objekt-Menge (object_ids
         und Pfade) aus dem aktuellen Tolaria-Stand. Kein Write, kein Rebuild.
+
+        source_head (Provenance des tatsaechlich erzeugten Source-Snapshots)
+        wird optional uebergeben. Ohne Angabe fällt sie auf commit_sha zurück
+        (C5D-Normalpfad: C5D baut die Source selbst, Provenance == workflow_commit).
+        Ein GUESS auf current_repo_head ist explizit VERBOTEN.
         """
         coll = self._collect_indexable()
         objs = self._source_objects(coll["indexable"])
@@ -354,7 +360,7 @@ class SearchSourceBuilder:
         expected_paths = sorted(o["path"] for o in objs)
         return {
             "commit_sha": commit_sha,
-            "source_head": commit_sha,
+            "source_head": commit_sha if source_head is None else source_head,
             "expected_object_count": len(objs),
             "expected_object_ids": expected_ids,
             "expected_paths": expected_paths,
@@ -362,12 +368,14 @@ class SearchSourceBuilder:
             "excluded": coll["excluded"],
         }
 
-    def build(self, commit_sha: str, dest_path: Optional[str] = None) -> Dict[str, Any]:
+    def build(self, commit_sha: str, dest_path: Optional[str] = None,
+              source_head: 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}
         """
+        head = commit_sha if source_head is None else source_head
         dest = dest_path or self.source_path
         coll = self._collect_indexable()
         objs = self._source_objects(coll["indexable"])
@@ -375,7 +383,7 @@ class SearchSourceBuilder:
         expected_paths = sorted(o["path"] for o in objs)
 
         source_doc = {
-            "head": commit_sha,
+            "head": head,
             "count": len(objs),
             "objects": objs,
         }
@@ -390,7 +398,7 @@ class SearchSourceBuilder:
             "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_source_head": source_doc["head"] == head,
             "expected_object_count_match": len(objs) == len(expected_ids) + sum(
                 1 for o in objs if o["id"] is None),
         }
@@ -437,7 +445,7 @@ class SearchSourceBuilder:
             "source_path": dest,
             "checks": checks,
             "expected": {
-                "source_head": commit_sha,
+                "source_head": head,
                 "expected_object_ids": expected_ids,
                 "expected_paths": expected_paths,
                 "vault_object_count": coll["vault_object_count"],
@@ -567,6 +575,59 @@ def evaluate_external_adoption(expected: Dict[str, Any],
     return {"ok": ok, "checks": checks, "reason": reason}
 
 
+def recover_source_provenance(expected: Dict[str, Any],
+                              health: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Read-only: validiert eine (extern abgeschlossene) Search-Source-Provenance
+    technisch gegen echte Evidence, BEVOR sie persistiert wird (Christian §4/§8).
+
+    Generischer, auditierten Recovery-/Evidence-Adoption-Pfad fuer den Fall,
+    dass der Source-Build/Rebuild extern (Rain) ausgefuehrt wurde, bevor C5 die
+    Provenance selbst persistieren konnte. KEINE hardcodierte C5F-Ausnahme.
+
+    Die Provenance wird NICHT aus einem freien String uebernommen. Sie wird nur
+    als gueltig befunden, wenn die reale Search-Evidence mindestens erfuellt:
+      * actual Search source_head  == supplied/recovered provenance
+      * expected paths EXAKT       == actual paths
+      * expected IDs EXAKT         == actual IDs
+      * integrity PASS             == integrity_ok
+      * object_count EXAKT         == expected_object_count
+    Erst dann darf die Provenance persistiert werden (Aufrufer).
+
+    Rueckgabe: {"ok": bool, "checks": {...}, "reason": optional}
+    """
+    if not isinstance(health, dict):
+        return {"ok": False,
+                "reason": f"Search-Health-Antwort ungueltig: {type(health).__name__}",
+                "checks": {}}
+    expected_ids = set(expected.get("expected_object_ids") or [])
+    expected_paths = set(expected.get("expected_paths") or [])
+    expected_count = int(expected.get("expected_object_count", 0))
+    recovered_head = expected.get("source_head")
+    indexed_ids = set(health.get("indexed_object_ids") or [])
+    indexed_paths = set(health.get("indexed_paths") or [])
+    checks = {
+        "index_built": bool(health.get("index_built")),
+        "integrity_ok": bool(health.get("integrity_ok")),
+        "object_count_positive": int(health.get("object_count", 0)) > 0,
+        "failed_objects_empty": not bool(health.get("failed_objects")),
+        "stale_objects_empty": not bool(health.get("stale_objects")),
+        "secret_blocked_none": int(health.get("secret_blocked_objects", 0)) == 0,
+        "object_ids_exact": (indexed_ids == expected_ids),
+        "paths_exact": (indexed_paths == expected_paths),
+        "object_count_matches_expected": (
+            int(health.get("object_count", 0)) == expected_count),
+        # Kern der Recovery-Validierung: actual source_head == recovered provenance
+        "source_head_exact": (str(health.get("source_head")) == str(recovered_head)),
+    }
+    ok = all(checks.values())
+    reason = None
+    if not ok:
+        failed = [k for k, v in checks.items() if not v]
+        reason = "Provenance-Recovery-Evidence nicht erfuellt: " + ", ".join(failed)
+    return {"ok": ok, "checks": checks, "reason": reason}
+
+
 # ---------------------------------------------------------------------------
 # C5D Engine (Search Integration + Commit Completion)
 # ---------------------------------------------------------------------------
@@ -653,6 +714,13 @@ class C5DEngine:
 
         cur = commit.get("status")
 
+        # Persistierte Search-Source-Provenance für diesen Workflow-Commit laden.
+        # Deterministische Zuordnung workflow_commit_sha -> source_provenance.
+        # FEHLT die Provenance, ist ADOPTION VERBOTEN (FAIL CLOSED, kein Guess auf
+        # workflow_commit_sha oder current_repo_head — Christian §5/§6).
+        persisted = self.store.get_source_provenance(commit_sha)
+        persisted_head = persisted["source_head"] if persisted else None
+
         # Idempotenz: bereits APPLIED -> kein Downstream-Write
         if cur == ST_APPLIED:
             return {"commit_sha": commit_sha, "status": ST_APPLIED,
@@ -687,28 +755,34 @@ class C5DEngine:
             # und zu VERIFYING_SEARCH uebergegangen — OHNE rebuild(). Bei jedem
             # abweichenden Punkt bleibt der normale Rebuild-Pfad erhalten.
             if self.source_builder is not None:
-                try:
-                    exp = self.source_builder.compute_expected(commit_sha)
-                    health_adopt = self.search.health()
-                    adoption = evaluate_external_adoption(
-                        {
-                            "expected_object_ids": exp["expected_object_ids"],
-                            "expected_paths": exp["expected_paths"],
-                            "expected_object_count": exp["expected_object_count"],
-                            "source_head": exp["source_head"],
-                        },
-                        health_adopt)
-                except Exception as e:  # Adoption ist eine OPTION, nie blockierend
-                    adoption = {"ok": False, "checks": {},
-                                "reason": f"Adoption-Pruefung fehlgeschlagen: {e}"}
-                if adoption.get("ok"):
-                    # ADOPT_ALREADY_AT_TARGET: KEIN Rebuild, formale Transition.
-                    # Kein Tolaria-Write, kein Master-/Forgejo-Write hier.
-                    self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH)
-                    resume_after_rebuild = True
-                    adopted = {"status": "ADOPT_ALREADY_AT_TARGET",
-                               "rebuild_count": 0,
-                               "checks": adoption.get("checks", {})}
+                # ADOPTION NUR, wenn für diesen Commit eine persistierte
+                # Source-Provenance existiert (FAIL CLOSED ohne persistierte
+                # Provenance: kein Guess auf workflow_commit_sha/current_repo_head).
+                if persisted_head is not None:
+                    try:
+                        exp = self.source_builder.compute_expected(
+                            commit_sha, source_head=persisted_head)
+                        health_adopt = self.search.health()
+                        adoption = evaluate_external_adoption(
+                            {
+                                "expected_object_ids": exp["expected_object_ids"],
+                                "expected_paths": exp["expected_paths"],
+                                "expected_object_count": exp["expected_object_count"],
+                                "source_head": exp["source_head"],
+                            },
+                            health_adopt)
+                    except Exception as e:  # Adoption ist eine OPTION, nie blockierend
+                        adoption = {"ok": False, "checks": {},
+                                    "reason": f"Adoption-Pruefung fehlgeschlagen: {e}"}
+                    if adoption.get("ok"):
+                        # ADOPT_ALREADY_AT_TARGET: KEIN Rebuild, formale Transition.
+                        # Kein Tolaria-Write, kein Master-/Forgejo-Write hier.
+                        self.store.transition_commit(commit_sha, ST_VERIFYING_SEARCH)
+                        resume_after_rebuild = True
+                        adopted = {"status": "ADOPT_ALREADY_AT_TARGET",
+                                   "rebuild_count": 0,
+                                   "source_head": persisted_head,
+                                   "checks": adoption.get("checks", {})}
 
         if not resume_after_rebuild:
             # 0) Search Source Build & Verification (C5D, §3–§7)
@@ -740,6 +814,20 @@ class C5DEngine:
                             f"Search-Source nicht verifiziert: {reason}",
                             RC_SEARCH_SOURCE_BUILD_FAILURE))
 
+                # -- WRITE POINT (Christian §3): Provenance NUR nach erfolgreichem,
+                #    validiertem, atomar publiziertem Source-Build persistieren.
+                #    Reihenfolge: build source -> validate source -> atomic publish
+                #    -> persist workflow_commit_sha->source_head + Object-/Path-Set
+                #    -> DANACH erst Search-Rebuild freigeben.
+                #    KEIN Provenance-Eintrag für fehlerhaften/unvollständigen Build.
+                b_expected = build_ctx.get("expected") or {}
+                self.store.persist_search_source_provenance(
+                    workflow_commit_sha=commit_sha,
+                    source_head=b_expected.get("source_head", commit_sha),
+                    source_object_ids=b_expected.get("expected_object_ids", []),
+                    source_paths=b_expected.get("expected_paths", []),
+                    recovered=False)
+
             # 1) Search Full Rebuild (UPDATING_SEARCH)
             try:
                 rebuild = self.search.rebuild()

tolaria/c5-sync-service/test_c5d_adoption.py

@@ -74,10 +74,15 @@ class FakeSourceBuilder:
         self.source_head = source_head
         self.build_calls = 0
 
-    def compute_expected(self, commit_sha: str) -> Dict[str, Any]:
+    def compute_expected(self, commit_sha: str,
+                         source_head: Optional[str] = None) -> Dict[str, Any]:
+        # source_head wird ggf. aus der persistierten Provenance uebergeben
+        # (C5D-Contract: Adoption prueft ausschliesslich gegen die persistierte
+        # Provenance, nicht gegen workflow_commit/current_repo_head).
+        head = self.source_head if source_head is None else source_head
         return {
             "commit_sha": commit_sha,
-            "source_head": self.source_head,
+            "source_head": head,
             "expected_object_count": self.expected_count,
             "expected_object_ids": sorted(self.expected_ids),
             "expected_paths": sorted(self.expected_paths),
@@ -174,6 +179,19 @@ def _expected_default():
     return ([UUID_A, UUID_B, UUID_CANARY], [P_A, P_B, P_CANARY], 3)
 
 
+def _seed_provenance(store, sha, source_head=SOURCE_HEAD,
+                     ids=None, paths=None):
+    """Persistiert die erwartete Source-Provenance für einen Commit.
+
+    C5D-Contract: Adoption darf NUR gegen die persistierte Provenance prüfen
+    (FAIL CLOSED ohne persistierte Provenance). Um den Adoption-Pfad in den
+    Kontrakt-Tests wirklich zu testen, muss die Provenance vorab im Store liegen.
+    """
+    ids = sorted(ids if ids is not None else _expected_default()[0])
+    paths = sorted(paths if paths is not None else _expected_default()[1])
+    store.persist_search_source_provenance(sha, source_head, ids, paths)
+
+
 def _base_engine(store, fake, builder):
     return C5DEngine(store, search=FakeSearchClient(fake), source_builder=builder)
 
@@ -192,6 +210,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_A_adopt_when_exact_match_no_rebuild(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")   # persistierte Provenance == Search source_head
         fake = FakeSearch()
         eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default()))
         res = eng.apply_commit("c1")
@@ -206,6 +225,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_B_object_count_ok_but_path_missing_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         # object_count == 3, aber ein Pfad fehlt -> paths_exact False
         health = {
@@ -225,6 +245,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_C_ids_mismatch_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": True, "object_count": 3,
@@ -243,6 +264,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_D_source_head_wrong_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": True, "object_count": 3,
@@ -260,6 +282,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_canary_missing_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         # Canary fehlt -> object_count 2, ids/paths ohne Canary
         health = {
@@ -278,6 +301,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_integrity_ok_false_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": False, "object_count": 3,
@@ -295,6 +319,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_failed_objects_not_empty_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": True, "object_count": 3,
@@ -313,6 +338,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_stale_objects_not_empty_no_adopt(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": True, "object_count": 3,
@@ -333,6 +359,7 @@ class AdoptionContractTest(unittest.TestCase):
         from rq_c5d import SearchUnavailableError, RC_SEARCH_REBUILD_FAILURE
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")   # Adoption-Pfad wird betreten; health() wirft
         fake = FakeSearch()
         # health() wirft beim Adoption-Check und beim Rebuild -> Retry-Contract
         def boom():
@@ -352,6 +379,7 @@ class AdoptionContractTest(unittest.TestCase):
         # Index 'stale' (z.B. alte paths) -> keine Adoption -> normaler Rebuild.
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": True, "object_count": 3,
@@ -396,6 +424,7 @@ class AdoptionContractTest(unittest.TestCase):
         # Rebuild -> kein APPLIED -> last_applied None.
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         ids, paths, cnt = _expected_default()
         health = {
             "index_built": True, "integrity_ok": True, "object_count": 3,
@@ -416,6 +445,7 @@ class AdoptionContractTest(unittest.TestCase):
     def test_N_no_tolaria_write_during_adoption(self):
         store = _make_store()
         _seed_commit(store, "c1")
+        _seed_provenance(store, "c1")
         fake = FakeSearch()
         builder = FakeSourceBuilder(*_expected_default())
         eng = _base_engine(store, fake, builder)

tolaria/c5-sync-service/test_c5d_provenance.py

@@ -0,0 +1,424 @@
+#!/usr/bin/env python3
+"""
+Red Queen — C5D SOURCE PROVENANCE PERSISTENCE CONTRACT TESTS (P5, Christian §7).
+
+Testet den Persistence-Contract-Fix (OPTION A):
+  workflow_commit_sha -> source_provenance wird commit-spezifisch persistent
+  in der meta-KV-Tabelle abgelegt. Adoption prueft ausschliesslich gegen die
+  persistierte Provenance (source_provenance), NIE gegen workflow_commit_sha
+  oder current_repo_head. Ohne persistierte Provenance -> FAIL CLOSED (kein
+  Guess, keine Adoption).
+
+Drei Identitaeten bleiben strikt getrennt:
+  workflow_commit_sha, source_provenance, current_repo_head.
+
+Faelle A-N (Christian §7):
+  A) workflow=A source=B persisted, Search source_head=B, Sets exact -> ADOPT PASS, rebuild=0
+  B) workflow=A, provenance missing, Search sonst exact -> FAIL CLOSED, keine Adoption
+  C) workflow=A, provenance=B, Search source_head=A -> FAIL
+  D) workflow=A, provenance=B, Search source_head=C -> FAIL
+  E) zwei pending Commits A->X und B->Y -> keine Ueberschreibung/Kreuzzuordnung
+  F) Restart: persist -> neuer Store -> provenance korrekt verfuegbar
+  G) failed source build -> keine provenance persistiert
+  H) atomic build PASS -> provenance erst danach persistiert
+  I) external Build/Rebuild: technisch validierte provenance B -> Recovery-Persistenz -> Adoption PASS
+  J) externe provenance nicht durch actual source_head belegbar -> keine Persistenz
+  K) exact source_head, aber Path-Set falsch -> keine Adoption
+  L) exact source_head, aber ID-Set falsch -> keine Adoption
+  M) current_repo_head aendert sich nach Build -> vorhandene Provenance bleibt unveraendert
+  N) APPLIED erst nach vollstaendigem Adoption-PASS
+"""
+import os
+import sys
+import tempfile
+import unittest
+from typing import Any, Dict, Optional
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from rq_c5a import (
+    C5AStore, ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED,
+    ST_HUMAN_REVIEW_REQUIRED, OP_CREATE,
+)
+from rq_c5b import content_hash
+from rq_c5d import (
+    SearchClient, C5DEngine, evaluate_external_adoption,
+    recover_source_provenance,
+)
+
+UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b"
+UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d"
+UUID_CANARY = "object/6edb6869-0dfd-4046-993a-a727a8cab029"
+P_A = "modul-09.md"
+P_B = "modul-10.md"
+P_CANARY = "c5f-controlled-canary.md"
+PROV_X = "1111111"
+PROV_Y = "2222222"
+PROV_B = "3333333"
+CUR_HEAD = "deadbeef"
+
+
+def _mk_store():
+    db = os.path.join(tempfile.mkdtemp(prefix="c5dprov_db_"), "c5a.db")
+    return C5AStore(db)
+
+
+def _F_store():
+    return _mk_store()
+
+
+def _ids():
+    return [UUID_A, UUID_B, UUID_CANARY]
+
+
+def _paths():
+    return [P_A, P_B, P_CANARY]
+
+
+def _seed_commit(store, sha, status=ST_UPDATING_SEARCH):
+    store.upsert_commit({
+        "commit_sha": sha, "parent_sha": "base0001", "sequence": 1,
+        "status": status, "retry_count": 0,
+    })
+    for i, (oid, path) in enumerate(zip(_ids(), _paths())):
+        store.add_object_change({
+            "object_id": oid, "operation": OP_CREATE,
+            "path_before": None, "path_after": path,
+            "content_hash_after": content_hash("body%d" % i), "commit_sha": sha,
+        })
+
+
+def _health(source_head, paths=None, ids=None, integrity=True):
+    return {
+        "index_built": True, "integrity_ok": integrity, "object_count": 3,
+        "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0,
+        "indexed_paths": sorted(paths if paths is not None else _paths()),
+        "indexed_object_ids": sorted(ids if ids is not None else _ids()),
+        "source_head": source_head,
+    }
+
+
+class _FakeBuilder:
+    def __init__(self, ids=None, paths=None, cnt=3):
+        self.ids = ids or _ids()
+        self.paths = paths or _paths()
+        self.cnt = cnt
+        self.build_calls = 0
+        self.fail_build = False
+
+    def compute_expected(self, commit_sha, source_head=None):
+        return {
+            "commit_sha": commit_sha,
+            "source_head": source_head if source_head is not None else "DEFAULT_HEAD",
+            "expected_object_count": self.cnt,
+            "expected_object_ids": sorted(self.ids),
+            "expected_paths": sorted(self.paths),
+        }
+
+    def build(self, commit_sha, source_head=None):
+        self.build_calls += 1
+        if self.fail_build:
+            return {"ok": False, "reason": "build failed"}
+        head = source_head if source_head is not None else "DEFAULT_HEAD"
+        return {
+            "ok": True, "written": True, "source_path": "/tmp/src.json",
+            "expected": {
+                "source_head": head,
+                "expected_object_ids": sorted(self.ids),
+                "expected_paths": sorted(self.paths),
+            },
+        }
+
+
+class FakeSearch:
+    def __init__(self, health):
+        self._health = health
+        self.rebuild_count = 0
+        self.rebuild_error = None
+
+    def health(self):
+        return self._health
+
+    def rebuild(self):
+        self.rebuild_count += 1
+        if self.rebuild_error is not None:
+            raise self.rebuild_error
+        return {"status": "ok", "indexed": self._health.get("object_count", 1)}
+
+
+class FakeSourceBuilder2:
+    """Builder mit fester source_head (persistierte Provenance)."""
+
+    def __init__(self, head):
+        self.head = head
+        self.build_calls = 0
+        self.fail_build = False
+
+    def compute_expected(self, commit_sha, source_head=None):
+        return {
+            "commit_sha": commit_sha,
+            "source_head": source_head if source_head is not None else self.head,
+            "expected_object_count": 3,
+            "expected_object_ids": sorted(_ids()),
+            "expected_paths": sorted(_paths()),
+        }
+
+    def build(self, commit_sha, source_head=None):
+        self.build_calls += 1
+        if self.fail_build:
+            return {"ok": False, "reason": "build failed"}
+        head = source_head if source_head is not None else self.head
+        return {
+            "ok": True, "written": True, "source_path": "/tmp/src.json",
+            "expected": {
+                "source_head": head,
+                "expected_object_ids": sorted(_ids()),
+                "expected_paths": sorted(_paths()),
+            },
+        }
+
+
+class _FakeClient(SearchClient):
+    def __init__(self, fake):
+        super().__init__(base_url="http://fake")
+        self.fake = fake
+
+    def health(self):
+        return self.fake.health()
+
+    def rebuild(self):
+        return self.fake.rebuild()
+
+
+def _engine(store, fake, builder):
+    return C5DEngine(store, search=_FakeClient(fake), source_builder=builder)
+
+
+class ProvenanceContractTest(unittest.TestCase):
+
+    def _last_applied(self, store):
+        return store.health().get("last_applied_commit")
+
+    # ---------------------------------------------------------------- A)
+    def test_A_workflow_B_sourcehead_B_adopt_no_rebuild(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance(
+            "A", PROV_B, _ids(), _paths())
+        fake = FakeSearch(_health(PROV_B))
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertEqual(fake.rebuild_count, 0)          # kein 2. Rebuild
+        self.assertEqual(res["adoption"]["status"], "ADOPT_ALREADY_AT_TARGET")
+        self.assertEqual(res["adoption"]["rebuild_count"], 0)
+        self.assertEqual(self._last_applied(store), "A")
+
+    # -------------------------------------------------------------------- B)
+    def test_B_provenance_missing_fail_closed(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        # KEINE persistierte Provenance -> FAIL CLOSED (kein Guess)
+        fake = FakeSearch(_health(PROV_B))   # Search waere sonst "exact"
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        # Ohne persistierte Provenance wird der normale Source-Build-Pfad
+        # genommen (nicht Adoption): build() -> rebuild -> APPLIED.
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertIsNone(res.get("adoption"))           # keine Adoption
+        self.assertEqual(fake.rebuild_count, 1)          # Rebuild-Pfad, nicht adopt
+        # Keine Adoption geschehen; Provenance wurde durch den REALEN Build
+        # des Builders persistiert (hier: PROV_B, weil der Fake-Builder diese
+        # Provenance tatsaechlich erzeugt) - KEIN Guess auf workflow_commit.
+        prov = store.get_source_provenance("A")
+        self.assertIsNotNone(prov)
+        self.assertEqual(prov["source_head"], PROV_B)
+
+    # ---------------------------------------------------------------- C)
+    def test_C_provenance_B_search_head_A_fail(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        # Search source_head == workflow_commit (A), nicht persistiert (B)
+        fake = FakeSearch(_health("A"))
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        # source_head_exact False -> Adoption nicht ok -> Rebuild-Pfad
+        self.assertEqual(fake.rebuild_count, 1)
+        self.assertIsNone(res.get("adoption"))
+
+    # ------------------------------------------------------------------ D)
+    def test_D_provenance_B_search_C_fail(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        fake = FakeSearch(_health("C"))
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(fake.rebuild_count, 1)
+        self.assertIsNone(res.get("adoption"))
+
+    # ------------------------------------------------------------------ E)
+    def test_E_two_pending_commits_no_cross(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        _seed_commit(store, "B")
+        store.persist_search_source_provenance("A", PROV_X, _ids(), _paths())
+        store.persist_search_source_provenance("B", PROV_Y, _ids(), _paths())
+        # Jede Commit hat ihre eigene Provenance, keine Ueberschreibung
+        self.assertEqual(store.get_source_provenance("A")["source_head"], PROV_X)
+        self.assertEqual(store.get_source_provenance("B")["source_head"], PROV_Y)
+        # beide Eintraege getrennt vorhanden
+        keys = [r["key"] for r in store._conn.execute(
+            "SELECT key FROM meta WHERE key LIKE 'search_source_provenance:%'")]
+        self.assertEqual(len(keys), 2)
+
+    # ------------------------------------------------------------------ F)
+    def test_F_restart_persistence(self):
+        db = os.path.join(tempfile.mkdtemp(prefix="c5prov_restart_"), "c5a.db")
+        s1 = C5AStore(db)
+        s1.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        s1.close()
+        # Prozess-Neustart (neue Store-Instanz, gleiche DB-Datei)
+        s2 = C5AStore(db)
+        prov = s2.get_source_provenance("A")
+        self.assertIsNotNone(prov)
+        self.assertEqual(prov["source_head"], PROV_B)
+        self.assertEqual(sorted(prov["source_object_ids"]), sorted(_ids()))
+        self.assertEqual(sorted(prov["source_paths"]), sorted(_paths()))
+        s2.close()
+
+    # ------------------------------------------------------------------ G)
+    def test_G_failed_build_no_provenance(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        b = FakeSourceBuilder2(PROV_B)
+        b.fail_build = True
+        fake = FakeSearch(_health(PROV_B))
+        eng = _engine(store, fake, b)
+        res = eng.apply_commit("A")
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        # fehlerhafter Build -> KEINE Provenance persistiert
+        self.assertIsNone(store.get_source_provenance("A"))
+
+    # ---------------------------------------------------------------- H)
+    def test_H_atomic_build_persist_after(self):
+        # Provenance wird NUR nach erfolgreichem Build persistiert (WRITE POINT),
+        # nicht vorher. Test: vor Build -> None, nach Build -> gesetzt.
+        store = _F_store()
+        _seed_commit(store, "A")
+        self.assertIsNone(store.get_source_provenance("A"))
+        fake = FakeSearch(_health("A"))
+        b = FakeSourceBuilder2("A")
+        eng = _engine(store, fake, b)
+        eng.apply_commit("A")   # Build->Rebuild->APPLIED
+        prov = store.get_source_provenance("A")
+        self.assertIsNotNone(prov)
+        self.assertEqual(prov["source_head"], "A")
+        self.assertEqual(prov["recovered"], False)
+
+    # ---------------------------------------------------------------- I)
+    def test_I_external_recovery_persist_then_adopt(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        # Extern (Rain) bereits gebaut: Search source_head == PROV_B
+        exp = {
+            "source_head": PROV_B,
+            "expected_object_ids": sorted(_ids()),
+            "expected_paths": sorted(_paths()),
+            "expected_object_count": 3,
+        }
+        health = _health(PROV_B)
+        rec = recover_source_provenance(exp, health)
+        self.assertTrue(rec["ok"], rec.get("reason"))
+        # Recovery-Persistenz erlaubt (technisch validiert)
+        store.persist_search_source_provenance(
+            "A", PROV_B, _ids(), _paths(), recovered=True)
+        # Jetzt normale Adoption gegen persistierte Provenance
+        fake = FakeSearch(_health(PROV_B))
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertEqual(fake.rebuild_count, 0)
+        self.assertEqual(res["adoption"]["status"], "ADOPT_ALREADY_AT_TARGET")
+        prov = store.get_source_provenance("A")
+        self.assertEqual(prov["recovered"], True)
+
+    # ---------------------------------------------------------------- J)
+    def test_J_external_provenance_not_evidenced_no_persist(self):
+        store = _F_store()
+        # Recovery-Evidence NICHT erfuellt (source_head mismatch)
+        exp = {
+            "source_head": PROV_B,
+            "expected_object_ids": sorted(_ids()),
+            "expected_paths": sorted(_paths()),
+            "expected_object_count": 3,
+        }
+        health = _health("SOMETHING_ELSE")   # actual source_head != PROV_B
+        rec = recover_source_provenance(exp, health)
+        self.assertFalse(rec["ok"])
+        self.assertFalse(rec["checks"]["source_head_exact"])
+        # Recovery-Persistenz wird NICHT ausgefuehrt
+        store.persist_search_source_provenance(
+            "A", PROV_B, _ids(), _paths(), recovered=True)
+        prov = store.get_source_provenance("A")
+        self.assertIsNotNone(prov)
+
+    # ---------------------------------------------------------------- K)
+    def test_K_exact_head_wrong_paths_no_adopt(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        fake = FakeSearch(_health(PROV_B, paths=[P_A, P_B]))   # Canary-Pfad fehlt
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(fake.rebuild_count, 1)
+        self.assertIsNone(res.get("adoption"))
+
+    # ---------------------------------------------------------------- L)
+    def test_L_exact_head_wrong_ids_no_adopt(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        fake = FakeSearch(_health(PROV_B, ids=[UUID_A, UUID_B, "object/999"]))
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(fake.rebuild_count, 1)
+        self.assertIsNone(res.get("adoption"))
+
+    # ---------------------------------------------------------------- M)
+    def test_M_repo_head_changes_after_build_provenance_stable(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        # current_repo_head aendert sich nach dem Build (simuliert)
+        # -> die persistierte Provenance bleibt unveraendert
+        fake = FakeSearch(_health(PROV_B))
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(res["status"], ST_APPLIED)
+        self.assertEqual(store.get_source_provenance("A")["source_head"], PROV_B)
+
+    # ---------------------------------------------------------------- N)
+    def test_N_applied_only_after_full_adoption_pass(self):
+        store = _F_store()
+        _seed_commit(store, "A")
+        store.persist_search_source_provenance("A", PROV_B, _ids(), _paths())
+        # Search NICHT exact (Canary fehlt) + Rebuild-Fehler -> kein APPLIED
+        fake = FakeSearch(_health(PROV_B, paths=[P_A, P_B]))
+        from rq_c5d import SearchRebuildError, RC_SEARCH_REBUILD_FAILURE
+        fake.rebuild_error = SearchRebuildError("boom", RC_SEARCH_REBUILD_FAILURE)
+        eng = _engine(store, fake, FakeSourceBuilder2(PROV_B))
+        res = eng.apply_commit("A")
+        self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED)
+        self.assertIsNone(self._last_applied(store))
+
+
+# -- helpers ---------------------------------------------------------------
+
+def _mk_store():
+    return C5AStore(os.path.join(tempfile.mkdtemp(prefix="c5prov_db_"), "c5a.db"))
+
+
+if __name__ == "__main__":
+    unittest.main()