From 747376ebb067196bba598811adefbbed4e44990b Mon Sep 17 00:00:00 2001 From: Red Queen Date: Wed, 26 Aug 2026 15:18:27 +0000 Subject: [PATCH] fix(tolaria): adopt externally completed search rebuild C5F P5 Option A: C5DEngine.apply_commit() uebernimmt einen extern bereits vollstaendig korrekt rebuildeten Suchzustand bei UPDATING_SEARCH idempotent (ADOPT_ALREADY_AT_TARGET) ohne zweiten Rebuild. - Neues evaluate_external_adoption(): read-only, exakte Set-Gleichheit (ids, paths, count, source_head), kein object_count-only shortcut - apply_commit(): UPDATING_SEARCH -> Adoption-Check -> bei exaktem Match direkt VERIFYING_SEARCH ohne rebuild() -> verify -> APPLIED - Neuer Report-Feld 'adoption' mit Status + rebuild_count - Testsuite test_c5d_adoption.py (Faelle A-O, 15 Tests) Kein zweiter Rebuild, kein manueller State-Set, kein Tolaria-/Forgejo-Write. --- tolaria/c5-sync-service/rq_c5d.py | 104 ++++- tolaria/c5-sync-service/test_c5d_adoption.py | 454 +++++++++++++++++++ 2 files changed, 557 insertions(+), 1 deletion(-) create mode 100644 tolaria/c5-sync-service/test_c5d_adoption.py diff --git a/tolaria/c5-sync-service/rq_c5d.py b/tolaria/c5-sync-service/rq_c5d.py index d7cfb0e..531418b 100644 --- a/tolaria/c5-sync-service/rq_c5d.py +++ b/tolaria/c5-sync-service/rq_c5d.py @@ -500,6 +500,73 @@ def verify_integrity(health: Dict[str, Any], return {"ok": ok, "checks": checks, "reason": reason} +# --------------------------------------------------------------------------- +# External-Rebuild Adoption (§C5F P5 Option A) +# --------------------------------------------------------------------------- + +def evaluate_external_adoption(expected: Dict[str, Any], + health: Dict[str, Any]) -> Dict[str, Any]: + """ + Bewertet READ-ONLY, ob ein extern bereits vollstaendig korrekt rebuildeter + Suchzustand (Rain) exakt dem deterministisch erwarteten Ziel entspricht. + + Wird NUR zur ADOPTION eines bereits erfolgreich rebuildeten Index + verwendet (C5F P5 Option A, ADOPT_ALREADY_AT_TARGET). Diese Funktion + fuehrt KEINEN Rebuild aus, schreibt NICHTS und aendert KEINEN State. + + Entscheidungsregel (harter KONTRAKT — kein object_count-only shortcut): + Adoption nur wenn ALLE Bedingungen gleichzeitig erfuellt sind: + * index_built == True + * integrity_ok == True + * object_count > 0 + * failed_objects leer + * stale_objects leer + * secret_blocked_objects == 0 + * indexed_paths EXAKT == expected_paths + * indexed_object_ids EXAKT == expected_object_ids + * source_head EXAKT == expected_source_head + Ein einziger abweichender Punkt -> KEINE Adoption. + + expected: {"expected_object_ids": set/list, "expected_paths": set/list, + "expected_object_count": int, "source_head": str} + health: {"index_built", "integrity_ok", "object_count", "failed_objects", + "stale_objects", "secret_blocked_objects", "indexed_paths", + "indexed_object_ids", "source_head"} + + 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)) + expected_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), + # Kein object_count-only shortcut: source_head muss exakt stimmen. + "source_head_exact": (str(health.get("source_head")) == str(expected_head)), + } + ok = all(checks.values()) + reason = None + if not ok: + failed = [k for k, v in checks.items() if not v] + reason = "Externe-ReBuild-Adoption nicht erfuellt: " + ", ".join(failed) + return {"ok": ok, "checks": checks, "reason": reason} + + # --------------------------------------------------------------------------- # C5D Engine (Search Integration + Commit Completion) # --------------------------------------------------------------------------- @@ -609,6 +676,40 @@ class C5DEngine: build_ctx = None rebuild = None + adopted = None + if not resume_after_rebuild: + # --- C5F P5 Option A: External-Rebuild Adoption (UPDATING_SEARCH) --- + # Wenn ein externer/privilegierter Prozess (Rain) den produktiven + # Search bereits vollstaendig korrekt rebuildet hat, uebernimmt C5D + # diesen Zustand READ-ONLY, statt einen ZWEITEN Rebuild zu erzwingen. + # Nur wenn der Search-Zustand EXAKT dem deterministisch erwarteten + # Ziel entspricht (siehe evaluate_external_adoption), wird adoptiert + # 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", {})} + if not resume_after_rebuild: # 0) Search Source Build & Verification (C5D, §3–§7) # Deterministisch AUS DEM AKTUELLEN TOLARIA-VAULT-ZUSTAND (read-only), @@ -703,7 +804,8 @@ class C5DEngine: return {"commit_sha": commit_sha, "status": ST_APPLIED, "idempotency": IDEM_RETRY_SAFE, "integrity": integrity, - "rebuild": rebuild, "source_build": build_ctx} + "rebuild": rebuild, "source_build": build_ctx, + "adoption": adopted} # --------------------------------------------------------------------------- diff --git a/tolaria/c5-sync-service/test_c5d_adoption.py b/tolaria/c5-sync-service/test_c5d_adoption.py new file mode 100644 index 0000000..0c2f421 --- /dev/null +++ b/tolaria/c5-sync-service/test_c5d_adoption.py @@ -0,0 +1,454 @@ +#!/usr/bin/env python3 +""" +Red Queen — C5F P5 Option A: External-Rebuild Adoption Contract Testsuite. + +Testet den neuen C5D-Contract: C5D uebernimmt einen extern bereits +vollstaendig korrekt rebuildeten Suchzustand bei UPDATING_SEARCH idempotent, +OHNE einen zweiten Rebuild auszuloesen. + +Kontrakt (siehe rq_c5d.evaluate_external_adoption): + UPDATING_SEARCH -> read-only expected/current Vergleich + -> ADOPT_ALREADY_AT_TARGET (nur bei exakter Set-Gleichheit) + -> VERIFYING_SEARCH -> final verify -> APPLIED + REBUILD_COUNT = 0. + +KEIN object_count-only shortcut. Kein zweiter Rebuild. Kein manueller +State-Set. Kein Tolaria-/Forgejo-Write. last_applied erst nach vollem PASS. + +Faelle A-M (Christian-Briefing §5): + A) UPDATING_SEARCH + Search exakt expected -> rebuild 0 -> adopt -> APPLIED + B) object_count gleich, aber Pfad fehlt -> keine Adoption (-> Rebuild-Pfad) + C) IDs stimmen nicht -> keine Adoption + D) source_head falsch -> keine Adoption + E) Canary/erwartetes neues Objekt fehlt -> keine Adoption + F) integrity_ok=false -> keine Adoption + G) failed_objects nicht leer -> keine Adoption + H) stale_objects nicht leer -> keine Adoption + I) Search leer/unavailable -> bestehender Retry/Failure-Contract + J) Search stale -> normaler Rebuild-Pfad bleibt funktionsfaehig + K) VERIFYING_SEARCH-Resume bleibt unveraendert (kein Rebuild) + L) Duplicate Replay nach APPLIED -> ALREADY_APPLIED, kein Rebuild + M) last_applied_commit nur nach vollstaendigem Adoption-PASS + N) kein Tolaria-Write waehrend Adoption + O) kein Master-/Forgejo-Write +""" +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__))) + +from rq_c5a import ( + C5AStore, ST_UPDATING_SEARCH, ST_VERIFYING_SEARCH, ST_APPLIED, + ST_HUMAN_REVIEW_REQUIRED, ST_RETRY_PENDING, + IDEM_ALREADY_APPLIED, IDEM_RETRY_SAFE, + OP_CREATE, +) +from rq_c5b import content_hash +from rq_c5d import ( + SearchClient, C5DEngine, evaluate_external_adoption, +) + +UUID_A = "object/77b02661-d67b-4bae-b612-01d1287cea6b" +UUID_B = "object/48bd264f-607b-15f1-5f73-3e922af9b19d" +UUID_CANARY = "object/6edb6869-0dfd-4046-993a-a727a8cab029" +SOURCE_HEAD = "12635e8" + +# Erwartete (relative) Pfade — identisch zum echten SearchSourceBuilder: +# _source_objects erzeugt `path` als repo-relativen Pfad (VAULT_PREFIX gestrippt). +P_A = "modul-09.md" +P_B = "modul-10.md" +P_CANARY = "c5f-controlled-canary.md" + + +class FakeSourceBuilder: + """Deterministisches erwartetes Objekt-Set via compute_expected().""" + + def __init__(self, expected_ids, expected_paths, expected_count, + source_head=SOURCE_HEAD): + self.expected_ids = expected_ids + self.expected_paths = expected_paths + self.expected_count = expected_count + self.source_head = source_head + self.build_calls = 0 + + def compute_expected(self, commit_sha: str) -> Dict[str, Any]: + return { + "commit_sha": commit_sha, + "source_head": self.source_head, + "expected_object_count": self.expected_count, + "expected_object_ids": sorted(self.expected_ids), + "expected_paths": sorted(self.expected_paths), + } + + def build(self, commit_sha: str): + self.build_calls += 1 + return { + "ok": True, "written": True, "source_path": "/tmp/source.json", + "expected": { + "source_head": self.source_head, + "expected_object_ids": sorted(self.expected_ids), + "expected_paths": sorted(self.expected_paths), + }, + } + + +class FakeSearch: + """Zaehlt rebuild()/health(); health ist konfigurierbar.""" + + def __init__(self, health: Optional[Dict[str, Any]] = None): + self.rebuild_count = 0 + self.health_calls = 0 + self._health = health or self._perfect_health() + self.rebuild_error: Optional[Exception] = None + + def _perfect_health(self) -> Dict[str, Any]: + return { + "index_built": True, + "object_count": 3, + "failed_objects": [], + "stale_objects": [], + "secret_blocked_objects": 0, + "integrity_ok": True, + "indexed_paths": sorted([P_A, P_B, P_CANARY]), + "indexed_object_ids": sorted([UUID_A, UUID_B, UUID_CANARY]), + "source_head": SOURCE_HEAD, + } + + def rebuild(self) -> Dict[str, Any]: + self.rebuild_count += 1 + if self.rebuild_error: + raise self.rebuild_error + # Ein erfolgreicher Rebuild repariert den Index: health zuruecksetzen + # auf den vollstaendig korrekten Zustand (wie ein echter produktiver + # Rebuild, der aus der Source den korrekten Index neu aufbaut). + self._health = self._perfect_health() + return {"status": "ok", "indexed": self._health.get("object_count", 1)} + + def get_health(self) -> Dict[str, Any]: + self.health_calls += 1 + return self._health + + +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.get_health() + + +def _make_store() -> C5AStore: + db = os.path.join(tempfile.mkdtemp(prefix="c5dadopt_db_"), "c5a.db") + return C5AStore(db) + + +def _default_objs(): + return [ + {"object_id": UUID_A, "operation": OP_CREATE, + "path_before": None, "path_after": P_A, + "content_hash_after": content_hash("body")}, + {"object_id": UUID_B, "operation": OP_CREATE, + "path_before": None, "path_after": P_B, + "content_hash_after": content_hash("body")}, + ] + + +def _seed_commit(store, sha, objs=None, status=ST_UPDATING_SEARCH): + store.upsert_commit({ + "commit_sha": sha, "parent_sha": "base0001", "sequence": 1, + "status": status, "retry_count": 0, + }) + for oc in (objs or _default_objs()): + oc["commit_sha"] = sha + store.add_object_change(oc) + + +def _expected_default(): + return ([UUID_A, UUID_B, UUID_CANARY], [P_A, P_B, P_CANARY], 3) + + +def _base_engine(store, fake, builder): + return C5DEngine(store, search=FakeSearchClient(fake), source_builder=builder) + + +def _perfect_engine(store): + ids, paths, cnt = _expected_default() + return _base_engine(store, FakeSearch(), FakeSourceBuilder(ids, paths, cnt)) + + +class AdoptionContractTest(unittest.TestCase): + + def _last_applied(self, store): + return store.health().get("last_applied_commit") + + # ------------------------------------------------------------------ A) + def test_A_adopt_when_exact_match_no_rebuild(self): + store = _make_store() + _seed_commit(store, "c1") + fake = FakeSearch() + eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default())) + res = eng.apply_commit("c1") + self.assertEqual(res["status"], ST_APPLIED) + self.assertEqual(fake.rebuild_count, 0) # KEIN zweiter Rebuild + self.assertEqual(res["adoption"]["status"], "ADOPT_ALREADY_AT_TARGET") + self.assertEqual(res["adoption"]["rebuild_count"], 0) + self.assertEqual(self._last_applied(store), "c1") + self.assertEqual(store.commit_status("c1"), ST_APPLIED) + + # ------------------------------------------------------------------ B) + def test_B_object_count_ok_but_path_missing_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + # object_count == 3, aber ein Pfad fehlt -> paths_exact False + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B], # P_CANARY fehlt + "indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) # Rebuild-Pfad + self.assertIsNone(res.get("adoption")) + self.assertEqual(res["status"], ST_APPLIED) + + # ------------------------------------------------------------------ C) + def test_C_ids_mismatch_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B, P_CANARY], + "indexed_object_ids": [UUID_A, UUID_B, "object/999"], + "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) + self.assertIsNone(res.get("adoption")) + + # ------------------------------------------------------------------ D) + def test_D_source_head_wrong_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B, P_CANARY], + "indexed_object_ids": sorted(ids), "source_head": "WRONGHEAD", + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) + self.assertIsNone(res.get("adoption")) + + # ------------------------------------------------------------------ E + def test_canary_missing_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + # Canary fehlt -> object_count 2, ids/paths ohne Canary + health = { + "index_built": True, "integrity_ok": True, "object_count": 2, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B], + "indexed_object_ids": [UUID_A, UUID_B], "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) + self.assertIsNone(res.get("adoption")) + + # ------------------------------------------------------------------ F + def test_integrity_ok_false_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": False, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B, P_CANARY], + "indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) + self.assertIsNone(res.get("adoption")) + + # ------------------------------------------------------------------ G + def test_failed_objects_not_empty_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": ["x"], "stale_objects": [], + "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B, P_CANARY], + "indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) + self.assertIsNone(res.get("adoption")) + + # ------------------------------------------------------------------ H + def test_stale_objects_not_empty_no_adopt(self): + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": ["x"], + "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B, P_CANARY], + "indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) + self.assertIsNone(res.get("adoption")) + + # ------------------------------------------------------------------ I + def test_I_unavailable_retry_contract(self): + # Search unerreichbar -> RETRY_PENDING, kein Adoption, kein APPLIED. + from rq_c5d import SearchUnavailableError, RC_SEARCH_REBUILD_FAILURE + store = _make_store() + _seed_commit(store, "c1") + fake = FakeSearch() + # health() wirft beim Adoption-Check und beim Rebuild -> Retry-Contract + def boom(): + raise SearchUnavailableError("search down", RC_SEARCH_REBUILD_FAILURE) + fake.get_health = boom + fake.rebuild_error = SearchUnavailableError( + "search down", RC_SEARCH_REBUILD_FAILURE) + eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default())) + res = eng.apply_commit("c1") + # Adoption-Pruefung wirft -> adoption False (nicht blockierend) + # -> danach rebuild() wirft ebenfalls -> Retry-Pfad (SearchUnavailable) + self.assertEqual(res["status"], ST_RETRY_PENDING) + self.assertIsNone(self._last_applied(store)) + + # ------------------------------------------------------------------ J) + def test_J_search_stale_normal_rebuild_path(self): + # Index 'stale' (z.B. alte paths) -> keine Adoption -> normaler Rebuild. + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, "alt.md"], # stale Pfad -> paths mismatch + "indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 1) # Rebuild-Pfad funktioniert + self.assertIsNone(res.get("adoption")) + self.assertEqual(res["status"], ST_APPLIED) + + # ------------------------------------------------------------------ K) + def test_K_verifying_search_resume_no_rebuild(self): + store = _make_store() + _seed_commit(store, "c1", status=ST_VERIFYING_SEARCH) + fake = FakeSearch() + eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default())) + res = eng.apply_commit("c1") + self.assertEqual(fake.rebuild_count, 0) # Resume -> kein Rebuild + self.assertEqual(res["status"], ST_APPLIED) + self.assertIsNone(res.get("adoption")) # kein Adoption-Report + + # ------------------------------------------------------------------ L) + def test_L_duplicate_after_applied_no_rebuild(self): + store = _make_store() + # Commit bereits vollstaendig APPLIED (last_applied gesetzt) -> Replay + _seed_commit(store, "c1", status=ST_APPLIED) + store.mark_applied("c1") + fake = FakeSearch() + eng = _base_engine(store, fake, FakeSourceBuilder(*_expected_default())) + res = eng.apply_commit("c1") + self.assertEqual(res["status"], ST_APPLIED) + self.assertEqual(res["idempotency"], IDEM_ALREADY_APPLIED) + self.assertEqual(fake.rebuild_count, 0) + + # ------------------------------------------------------------------ M) + def test_M_last_applied_only_after_full_pass(self): + # Bei fehlgeschlagener Adoption (Pfad fehlt) und NICHT erfolgreichem + # Rebuild -> kein APPLIED -> last_applied None. + store = _make_store() + _seed_commit(store, "c1") + ids, paths, cnt = _expected_default() + health = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": [P_A, P_B], # fehlt -> keine Adoption + "indexed_object_ids": [UUID_A, UUID_B], "source_head": SOURCE_HEAD, + } + fake = FakeSearch(health) + # Rebuild liefert Fehler -> Human Gate (nicht APPLIED) + from rq_c5d import SearchRebuildError, RC_SEARCH_REBUILD_FAILURE + fake.rebuild_error = SearchRebuildError("fail", RC_SEARCH_REBUILD_FAILURE) + eng = _base_engine(store, fake, FakeSourceBuilder(ids, paths, cnt)) + res = eng.apply_commit("c1") + self.assertEqual(res["status"], ST_HUMAN_REVIEW_REQUIRED) + self.assertIsNone(self._last_applied(store)) + + # ------------------------------------------------------------------ N) ENGINE + def test_N_no_tolaria_write_during_adoption(self): + store = _make_store() + _seed_commit(store, "c1") + fake = FakeSearch() + builder = FakeSourceBuilder(*_expected_default()) + eng = _base_engine(store, fake, builder) + res = eng.apply_commit("c1") + self.assertEqual(res["status"], ST_APPLIED) + self.assertEqual(fake.rebuild_count, 0) + self.assertEqual(builder.build_calls, 0) # kein Source-Build-Write + # Adoption ist der KEIN-Write-Pfad: adoption gesetzt, rebuild_count 0 + self.assertEqual(res["adoption"]["status"], "ADOPT_ALREADY_AT_TARGET") + self.assertEqual(res["adoption"]["rebuild_count"], 0) + + # ------------------------------------------------------------------ O) ENGINE + def test_evaluate_external_adoption_pure(self): + # Reine Funktion: alle Checks, nur exakt expected -> ok True + ids, paths, cnt = _expected_default() + exp = {"expected_object_ids": sorted(ids), "expected_paths": sorted(paths), + "expected_object_count": cnt, "source_head": SOURCE_HEAD} + h = { + "index_built": True, "integrity_ok": True, "object_count": 3, + "failed_objects": [], "stale_objects": [], "secret_blocked_objects": 0, + "indexed_paths": sorted(paths), + "indexed_object_ids": sorted(ids), "source_head": SOURCE_HEAD, + } + r = evaluate_external_adoption(exp, h) + self.assertTrue(r["ok"]) + + # object_count gleich aber path fehlt -> ok False (kein count-only shortcut) + h2 = dict(h, indexed_paths=[P_A, P_B]) + r2 = evaluate_external_adoption(exp, h2) + self.assertFalse(r2["ok"]) + self.assertIn("paths_exact", r2["checks"]) + self.assertFalse(r2["checks"]["paths_exact"]) + + +if __name__ == "__main__": + unittest.main()