From 363f27327bbb595248030bd181e28de6e01b5053 Mon Sep 17 00:00:00 2001 From: Red Queen Date: Wed, 26 Aug 2026 16:07:49 +0000 Subject: [PATCH] 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:), 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 | 60 +++ tolaria/c5-sync-service/rq_c5d.py | 144 ++++-- tolaria/c5-sync-service/test_c5d_adoption.py | 34 +- .../c5-sync-service/test_c5d_provenance.py | 424 ++++++++++++++++++ 4 files changed, 632 insertions(+), 30 deletions(-) create mode 100644 tolaria/c5-sync-service/test_c5d_provenance.py diff --git a/tolaria/c5-sync-service/rq_c5a.py b/tolaria/c5-sync-service/rq_c5a.py index 8c535e1..65d5c71 100644 --- a/tolaria/c5-sync-service/rq_c5a.py +++ b/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() diff --git a/tolaria/c5-sync-service/rq_c5d.py b/tolaria/c5-sync-service/rq_c5d.py index 531418b..7b5ed45 100644 --- a/tolaria/c5-sync-service/rq_c5d.py +++ b/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() diff --git a/tolaria/c5-sync-service/test_c5d_adoption.py b/tolaria/c5-sync-service/test_c5d_adoption.py index 0c2f421..65e61cb 100644 --- a/tolaria/c5-sync-service/test_c5d_adoption.py +++ b/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) diff --git a/tolaria/c5-sync-service/test_c5d_provenance.py b/tolaria/c5-sync-service/test_c5d_provenance.py new file mode 100644 index 0000000..e243d48 --- /dev/null +++ b/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()