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

@@ -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}
 
 
 # ---------------------------------------------------------------------------

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()